ڈوکر سیٹ اپ میں کافکا کے ساتھ اسپارک ورکر کنکشن کے مسائل کو حل کرنا

ایک ڈاکرائزڈ ماحول میں چنگاری اور کافکا کو یکجا کرنے کے چیلنجز

کیا آپ کو کبھی بھی کنیکٹیویٹی کے مسئلے کا سامنا کرنا پڑا ہے۔ کافکا بروکر ایک میں چنگاری کلسٹر ایک ڈاکر سیٹ اپ کے اندر؟ آپ اکیلے نہیں ہیں! بہت سے ڈویلپرز کو ان دو طاقتور ٹولز کے درمیان مواصلت قائم کرتے وقت رکاوٹوں کا سامنا کرنا پڑتا ہے۔ 🛠️

حال ہی میں، میں نے اپنے کو بڑھانے کا آغاز کیا۔ چنگاری کلسٹر ریئل ٹائم ڈیٹا پروسیسنگ کو ہموار کرنے کے لیے کافکا بروکر کو شامل کر کے۔ تاہم، میں نے مسلسل کنکشن ٹائم آؤٹ اور DNS ریزولوشن کی غلطیوں کے ساتھ ایک روڈ بلاک کو نشانہ بنایا، جس نے اس عمل کو ایک ٹربل شوٹنگ میراتھن میں بدل دیا۔ 😅

یہ مسائل ڈوکر کمپوز اور اسپارک کی کافکا سے متعلق کنفیگریشنز میں غلط کنفیگرڈ سیٹنگز سے پیدا ہوئے ہیں۔ متعدد گائیڈز کی پیروی کرنے اور متعدد پیرامیٹرز کو موافقت کرنے کے باوجود، مضحکہ خیز "بروکر دستیاب نہیں ہو سکتا ہے" پیغام برقرار رہا، جس نے مجھے حیران اور مایوس کر دیا۔

اس آرٹیکل میں، میں اپنے تجربے کا اشتراک کروں گا اور Docker ماحول میں Spark کارکنوں اور کافکا بروکرز کے درمیان رابطے کے چیلنجوں کو حل کرنے کے لیے عملی اقدامات پیش کروں گا۔ راستے میں، آپ ان خرابیوں سے بچنے اور بغیر کسی رکاوٹ کے انضمام کو یقینی بنانے کے لیے تجاویز اور ترکیبیں سیکھیں گے۔ آئیے اندر غوطہ لگائیں! 🚀

ڈوکر میں سپارک اور کافکا انٹیگریشن کو سمجھنا

پہلا اسکرپٹ a کے درمیان تعلق قائم کرنے پر مرکوز ہے۔ چنگاری کارکن اور a کافکا بروکر. اسپارک کے سٹرکچرڈ سٹریمنگ API کا استعمال کرتے ہوئے، اسکرپٹ کافکا کے موضوع سے حقیقی وقت کا ڈیٹا پڑھتا ہے۔ اس کا آغاز اسپارک سیشن کو شروع کرنے اور اسے مطلوبہ کافکا پیکج کے ساتھ ترتیب دینے سے ہوتا ہے۔ یہ بہت اہم ہے کیونکہ یہ اسپارک کو کافکا کے ساتھ بغیر کسی رکاوٹ کے بات چیت کرنے کے لیے ضروری انحصار فراہم کرتا ہے۔ اس انحصار کی ایک مثال `org.apache.spark:spark-sql-kafka` پیکیج ہے، جو Docker ماحول میں Spark اور Kafka کے درمیان مطابقت کو یقینی بناتا ہے۔

کافکا کے پیغامات کو ہینڈل کرنے کے لیے، اسکرپٹ 'StructType' کا استعمال کرتے ہوئے اسکیما کی وضاحت کرتی ہے۔ یہ اسکیما اس بات کو یقینی بناتا ہے کہ آنے والے پیغامات کو صحیح طریقے سے پارس اور ڈھانچہ بنایا گیا ہے۔ حقیقی دنیا کے منظرناموں میں اکثر کافکا سے JSON ڈیٹا کو سنبھالنا شامل ہوتا ہے۔ مثال کے طور پر، ایک کرپٹو کرنسی مانیٹرنگ سسٹم کا تصور کریں جہاں قیمتوں کی تازہ کاری والے پیغامات کافکا کو بھیجے جاتے ہیں۔ ان پیغامات کو پڑھنے کے قابل فارمیٹ میں پارس کرنا رجحان کی پیشین گوئی کے لیے ڈیٹا پر کارروائی اور تجزیہ کرنا آسان بناتا ہے۔ 🪙

ڈوکر کمپوز کنفیگریشن رابطے کے مسائل کو حل کرنے میں اہم کردار ادا کرتی ہے۔ 'KAFKA_ADVERTISED_LISTENERS' اور 'KAFKA_LISTENERS' ترتیبات کو Docker نیٹ ورک کے اندر اندرونی اور بیرونی مواصلات میں فرق کرنے کے لیے ایڈجسٹ کیا گیا ہے۔ یہ اس بات کو یقینی بناتا ہے کہ ایک ہی ڈوکر نیٹ ورک پر چلنے والی خدمات، جیسے اسپارک اور کافکا، DNS ریزولوشن کے مسائل کے بغیر بات چیت کر سکتی ہیں۔ مثال کے طور پر، نقشہ بندی `INSIDE://kafka:9093` اندرونی کنٹینرز کو کافکا تک رسائی کی اجازت دیتی ہے، جبکہ `OUTSIDE://localhost:9093` بیرونی ایپلی کیشنز جیسے مانیٹرنگ ٹولز کو مربوط کرنے کے قابل بناتا ہے۔

دوسرا اسکرپٹ یہ ظاہر کرتا ہے کہ کافکا کنکشن کی جانچ کے لیے Python `KafkaConsumer` کو کیسے استعمال کیا جائے۔ یہ یقینی بنانے کے لیے کہ کافکا بروکر صحیح طریقے سے کام کر رہا ہے، یہ ایک سادہ لیکن موثر طریقہ ہے۔ مخصوص عنوان سے پیغامات استعمال کرکے، آپ تصدیق کر سکتے ہیں کہ آیا ڈیٹا کا بہاؤ بلا تعطل ہے۔ ایسی ایپلیکیشن پر غور کریں جہاں صارف اسٹاک مارکیٹ کے ڈیٹا کو ٹریک کرنا چاہتا ہے۔ اس کنزیومر اسکرپٹ کا استعمال کرتے ہوئے کنکشن کی جانچ اس بات کو یقینی بناتی ہے کہ کنفیگریشن کی غلطیوں کی وجہ سے کوئی بھی اہم اپ ڈیٹ چھوٹ نہ جائے۔ ان ٹولز کے ساتھ، آپ ریئل ٹائم ڈیٹا پروسیسنگ کے لیے پراعتماد طریقے سے مضبوط سسٹم لگا سکتے ہیں! 🚀

# 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")` کو شامل کرنا بھاری بوجھ کے باوجود موثر ڈیٹا کی بازیافت کو یقینی بناتا ہے۔ سٹاک کی قیمتوں کو ریئل ٹائم میں ٹریک کرنے والی پروڈکشن گریڈ ایپلی کیشن کا تصور کریں — جگہ جگہ فیل سیف ہونا نیٹ ورک کی ہچکی کے دوران بھی ڈیٹا کے بلاتعطل بہاؤ کو یقینی بناتا ہے۔ یہ تکنیکیں مل کر ایک قابل اعتماد ڈیٹا پروسیسنگ پائپ لائن کی ریڑھ کی ہڈی بنتی ہیں۔ 🚀