ডকার সেটআপে কাফকার সাথে স্পার্ক ওয়ার্কার সংযোগ সমস্যা সমাধান করা

একটি ডকারাইজড পরিবেশে স্পার্ক এবং কাফকাকে একীভূত করার চ্যালেঞ্জ

একীভূত করার সময় আপনি কি কখনও সংযোগ সমস্যার সম্মুখীন হয়েছেন? কাফকা দালাল মধ্যে a স্পার্ক ক্লাস্টার একটি ডকার সেটআপের মধ্যে? আপনি একা নন! এই দুটি শক্তিশালী টুলের মধ্যে যোগাযোগ স্থাপন করার সময় অনেক ডেভেলপার বাধার সম্মুখীন হন। 🛠️

সম্প্রতি, আমি আমার উন্নত শুরু স্পার্ক ক্লাস্টার রিয়েল-টাইম ডেটা প্রসেসিং স্ট্রীমলাইন করতে একটি কাফকা ব্রোকার যোগ করে। যাইহোক, আমি অবিরাম সংযোগ টাইমআউট এবং DNS রেজোলিউশন ত্রুটি সহ একটি রোডব্লক আঘাত করেছি, যা প্রক্রিয়াটিকে একটি সমস্যা সমাধানের ম্যারাথনে পরিণত করেছে। 😅

এই সমস্যাগুলি ডকার কম্পোজ এবং স্পার্কের কাফকা-সম্পর্কিত কনফিগারেশনগুলিতে ভুল কনফিগার করা সেটিংস থেকে উদ্ভূত হয়েছে। বেশ কয়েকটি নির্দেশিকা অনুসরণ করা এবং অসংখ্য প্যারামিটারে টুইক করা সত্ত্বেও, অধরা "দালাল উপলব্ধ নাও হতে পারে" বার্তাটি অব্যাহত ছিল, যা আমাকে বিভ্রান্ত ও হতাশ করে রেখেছিল।

এই নিবন্ধে, আমি আমার অভিজ্ঞতা শেয়ার করব এবং একটি ডকার পরিবেশে স্পার্ক কর্মী এবং কাফকা ব্রোকারদের মধ্যে সংযোগের চ্যালেঞ্জগুলি সমাধান করার জন্য বাস্তব পদক্ষেপগুলি অফার করব। পথ ধরে, আপনি এই সমস্যাগুলি এড়াতে এবং একটি নিরবচ্ছিন্ন একীকরণ নিশ্চিত করার জন্য টিপস এবং কৌশলগুলি শিখবেন৷ এর মধ্যে ডুব দেওয়া যাক! 🚀

ডকারে স্পার্ক এবং কাফকা ইন্টিগ্রেশন বোঝা

প্রথম স্ক্রিপ্টটি ক-এর মধ্যে সংযোগ স্থাপনের উপর দৃষ্টি নিবদ্ধ করে স্পার্ক কর্মী এবং ক কাফকা দালাল. স্পার্কের স্ট্রাকচার্ড স্ট্রিমিং API ব্যবহার করে, স্ক্রিপ্টটি কাফকা বিষয় থেকে রিয়েল-টাইম ডেটা পড়ে। এটি একটি স্পার্ক সেশন শুরু করে এবং প্রয়োজনীয় কাফকা প্যাকেজের সাথে কনফিগার করার মাধ্যমে শুরু হয়। এটি অত্যন্ত গুরুত্বপূর্ণ কারণ এটি কাফকার সাথে নির্বিঘ্নে যোগাযোগ করার জন্য স্পার্কের জন্য প্রয়োজনীয় নির্ভরতা প্রদান করে। এই নির্ভরতার একটি উদাহরণ হল `org.apache.spark:spark-sql-kafka` প্যাকেজ, যা একটি ডকার পরিবেশে স্পার্ক এবং কাফকার মধ্যে সামঞ্জস্যতা নিশ্চিত করে।

কাফকা বার্তাগুলি পরিচালনা করতে, স্ক্রিপ্টটি `স্ট্রাকট টাইপ` ব্যবহার করে একটি স্কিমা সংজ্ঞায়িত করে। এই স্কিমা নিশ্চিত করে যে আগত বার্তাগুলি সঠিকভাবে পার্স এবং কাঠামোগত। বাস্তব-বিশ্বের পরিস্থিতিতে প্রায়ই কাফকার থেকে JSON ডেটা পরিচালনা করা জড়িত। উদাহরণ স্বরূপ, একটি ক্রিপ্টোকারেন্সি মনিটরিং সিস্টেম কল্পনা করুন যেখানে দামের আপডেট সম্বলিত বার্তাগুলি কাফকাকে পাঠানো হয়। এই বার্তাগুলিকে একটি পঠনযোগ্য বিন্যাসে পার্স করা প্রবণতা পূর্বাভাসের জন্য ডেটা প্রক্রিয়া এবং বিশ্লেষণ করা সহজ করে তোলে৷ 🪙

ডকার কম্পোজ কনফিগারেশন সংযোগ সমস্যা সমাধানে একটি গুরুত্বপূর্ণ ভূমিকা পালন করে। 'KAFKA_ADVERTISED_LISTENERS' এবং 'KAFKA_LISTENERS' সেটিংস ডকার নেটওয়ার্কের মধ্যে অভ্যন্তরীণ এবং বাহ্যিক যোগাযোগের পার্থক্য করার জন্য সামঞ্জস্য করা হয়েছে। এটি নিশ্চিত করে যে একই ডকার নেটওয়ার্কে চলমান পরিষেবাগুলি, যেমন স্পার্ক এবং কাফকা, ডিএনএস রেজোলিউশন সমস্যা ছাড়াই ইন্টারঅ্যাক্ট করতে পারে। উদাহরণ স্বরূপ, `INSIDE://kafka:9093` ম্যাপিং অভ্যন্তরীণ কন্টেইনারকে কাফকাকে অ্যাক্সেস করার অনুমতি দেয়, অন্যদিকে `OUTSIDE://localhost:9093` বাহ্যিক অ্যাপ্লিকেশন যেমন পর্যবেক্ষণ টুল সংযোগ করতে সক্ষম করে।

দ্বিতীয় স্ক্রিপ্টটি দেখায় কিভাবে কাফকা সংযোগ পরীক্ষা করার জন্য একটি পাইথন `কাফকা কনজিউমার` ব্যবহার করতে হয়। কাফকা ব্রোকার সঠিকভাবে কাজ করছে কিনা তা নিশ্চিত করার জন্য এটি একটি সহজ কিন্তু কার্যকর পদ্ধতি। নির্দিষ্ট বিষয় থেকে বার্তা গ্রহণ করে, আপনি ডেটা প্রবাহ নিরবচ্ছিন্ন কিনা তা যাচাই করতে পারেন। একটি অ্যাপ্লিকেশন বিবেচনা করুন যেখানে একজন ব্যবহারকারী স্টক মার্কেট ডেটা ট্র্যাক করতে চায়৷ এই ভোক্তা স্ক্রিপ্ট ব্যবহার করে সংযোগ পরীক্ষা করা নিশ্চিত করে যে কনফিগারেশন ত্রুটির কারণে কোনো গুরুত্বপূর্ণ আপডেট মিস করা হয়নি। এই সরঞ্জামগুলির সাহায্যে, আপনি আত্মবিশ্বাসের সাথে রিয়েল-টাইম ডেটা প্রক্রিয়াকরণের জন্য শক্তিশালী সিস্টেম স্থাপন করতে পারেন! 🚀

# Import necessary modules
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType
from pyspark.sql.functions import from_json, col
# Initialize Spark session with Kafka dependency
spark = SparkSession.builder \
    .appName("KafkaDebugReader") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
    .getOrCreate()
# Define schema for Kafka message
schema = StructType().add("message", StringType())
# Set up Kafka source for streaming data
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9093") \
    .option("subscribe", "crypto_topic") \
    .option("startingOffsets", "earliest") \
    .load()
# Parse Kafka message
messages = df.select(from_json(col("value").cast("string"), schema).alias("data")) \
    .select("data.message")
# Output data to console
query = messages.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()
query.awaitTermination()

version: '3.8'
services:
  kafka:
    image: wurstmeister/kafka
    container_name: kafka
    ports:
      - "9093:9093"
    environment:
      KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:9093,OUTSIDE://localhost:9093
      KAFKA_LISTENERS: INSIDE://:9093,OUTSIDE://0.0.0.0:9093
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
    networks:
      - my_network
  zookeeper:
    image: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    networks:
      - my_network
networks:
  my_network:
    driver: bridge

# Import KafkaConsumer from Kafka library
from kafka import KafkaConsumer
# Create a Kafka Consumer instance
consumer = KafkaConsumer(
    'crypto_topic',
    bootstrap_servers='kafka:9093',
    auto_offset_reset='earliest',
    enable_auto_commit=False,
    group_id='api_data'
)
# Poll messages from Kafka topic
for message in consumer:
    print(f"Received message: {message.value.decode('utf-8')}")
# Ensure to close the consumer
consumer.close()

একটি ডকারাইজড পরিবেশে কাফকা এবং স্পার্ক অপ্টিমাইজ করা

মধ্যে মসৃণ যোগাযোগ নিশ্চিত করার একটি গুরুত্বপূর্ণ দিক কাফকা ব্রোকারস এবং স্পার্ক ওয়ার্কার্স ডকারে কার্যকরভাবে নেটওয়ার্ক সেটিংস কনফিগার করছে। ডকার কন্টেইনারগুলি বিচ্ছিন্ন পরিবেশে কাজ করে, যখন পরিষেবাগুলিকে ইন্টারঅ্যাক্ট করার প্রয়োজন হয় তখন প্রায়ই DNS রেজোলিউশনের সমস্যা সৃষ্টি করে। এটি মোকাবেলা করার জন্য, আপনি ডকার কম্পোজের নেটওয়ার্ক কনফিগারেশন বিকল্পগুলি ব্যবহার করতে পারেন। উদাহরণস্বরূপ, `my_network` এর মতো একটি কাস্টম নেটওয়ার্ক সংজ্ঞায়িত করা এবং পরিষেবাগুলি লিঙ্ক করা নিশ্চিত করে যে কন্টেইনারগুলি একে অপরকে IP এর পরিবর্তে নাম দিয়ে চিনতে পারে, যা সেটআপকে সহজ করে এবং সাধারণ সমস্যাগুলি এড়ায়।

আরেকটি অপরিহার্য বিবেচনা হল কাফকার শ্রোতা কনফিগারেশনগুলিকে অপ্টিমাইজ করা। আপনার ডকার কম্পোজ ফাইলে `KAFKA_ADVERTISED_LISTENERS` এবং `KAFKA_LISTENERS` উল্লেখ করে, আপনি কাফকাকে তার ক্লায়েন্টদের উপযুক্ত ঠিকানার বিজ্ঞাপন দেওয়ার অনুমতি দেন। অভ্যন্তরীণ এবং বাহ্যিক শ্রোতাদের মধ্যে এই পার্থক্যটি দ্বন্দ্বের সমাধান করে, বিশেষ করে যখন স্পার্ক ওয়ার্কাররা ডকার নেটওয়ার্কের বাইরে থেকে সংযোগ করার চেষ্টা করে। এর একটি বাস্তব-জীবনের উদাহরণ হল একটি মনিটরিং ড্যাশবোর্ড যা হোস্ট মেশিন থেকে কাফকা ডেটা অনুসন্ধান করে, অ্যাক্সেসের জন্য একটি স্বতন্ত্র বাহ্যিক শ্রোতার প্রয়োজন। 🔧

অবশেষে, আপনার স্পার্ক অ্যাপ্লিকেশনগুলিতে শক্তিশালী ত্রুটি পরিচালনা করা অত্যন্ত গুরুত্বপূর্ণ। উদাহরণস্বরূপ, কাফকা কনফিগারেশনের মধ্যে পুনঃপ্রচার এবং ফলব্যাকগুলিকে সুবিধাজনকভাবে অস্থায়ী সংযোগ সমস্যাগুলি সুন্দরভাবে পরিচালনা করতে পারে। `.option("kafka.consumer.max.poll.records", "500")` যোগ করা, এমনকি ভারী বোঝার মধ্যেও দক্ষ ডেটা পুনরুদ্ধার নিশ্চিত করে৷ কল্পনা করুন যে একটি প্রোডাকশন-গ্রেড অ্যাপ্লিকেশন রিয়েল-টাইমে স্টকের দাম ট্র্যাক করছে—যাতে ব্যর্থ-নিরাপদ থাকা নেটওয়ার্ক হেঁচকির সময়ও নিরবচ্ছিন্ন ডেটা প্রবাহ নিশ্চিত করে। এই কৌশলগুলি একসাথে একটি নির্ভরযোগ্য ডেটা প্রসেসিং পাইপলাইনের মেরুদণ্ড গঠন করে। 🚀