Docker Kurulumunda Kafka ile Spark Worker Bağlantı Sorunlarını Çözme

Spark ve Kafka'yı Docker Ortamına Entegre Etmenin Zorlukları

Bir entegrasyonu yaparken hiç bağlantı sorunuyla karşılaştınız mı? Kafka Brokerı bir içine Kıvılcım Kümesi Docker kurulumunda mı? Yalnız değilsin! Birçok geliştirici bu iki güçlü araç arasındaki iletişimi kurarken engellerle karşılaşıyor. 🛠️

Son zamanlarda kendimi geliştirmeye başladım. Kıvılcım Kümesi gerçek zamanlı veri işlemeyi kolaylaştırmak için bir Kafka aracısı ekleyerek. Ancak, kalıcı bağlantı zaman aşımları ve DNS çözümleme hatalarıyla birlikte bir engelle karşılaştım ve bu da süreci bir sorun giderme maratonuna dönüştürdü. 😅

Bu sorunlar, Docker Compose ve Spark'ın Kafka ile ilgili yapılandırmalarındaki yanlış yapılandırılmış ayarlardan kaynaklanıyordu. Birkaç kılavuzu izlememe ve çok sayıda parametreyi değiştirmeme rağmen, yakalanması zor "komisyoncu mevcut olmayabilir" mesajı devam etti ve beni şaşkına çevirdi ve hayal kırıklığına uğrattı.

Bu makalede, Docker ortamında Spark çalışanları ile Kafka aracıları arasındaki bağlantı sorunlarını çözmek için deneyimlerimi paylaşacağım ve pratik adımlar sunacağım. Yol boyunca bu tuzaklardan kaçınmak ve kusursuz bir entegrasyon sağlamak için ipuçları ve püf noktaları öğreneceksiniz. Hadi dalalım! 🚀

Docker'da Spark ve Kafka Entegrasyonunu Anlamak

İlk senaryo, bir kişi arasında bir bağlantı kurmaya odaklanır. Kıvılcım İşçisi ve bir Kafka Brokerı. Komut dosyası, Spark'ın Yapılandırılmış Akış API'sini kullanarak bir Kafka konusundan gerçek zamanlı verileri okur. Bir Spark oturumu başlatmak ve onu gerekli Kafka paketiyle yapılandırmakla başlar. Bu, Spark'ın Kafka ile sorunsuz bir şekilde iletişim kurabilmesi için gerekli bağımlılığı sağladığı için çok önemlidir. Bu bağımlılığın bir örneği, Docker ortamında Spark ve Kafka arasında uyumluluğu sağlayan "org.Apache.spark:spark-sql-kafka" paketidir.

Betik, Kafka mesajlarını işlemek için 'StructType'ı kullanarak bir şema tanımlar. Bu şema, gelen mesajların doğru şekilde ayrıştırılıp yapılandırılmasını sağlar. Gerçek dünya senaryoları genellikle Kafka'dan JSON verilerinin işlenmesini içerir. Örneğin fiyat güncellemelerini içeren mesajların Kafka'ya gönderildiği bir kripto para izleme sistemi düşünün. Bu mesajları okunabilir bir formatta ayrıştırmak, trend tahmini için verilerin işlenmesini ve analiz edilmesini kolaylaştırır. 🪙

Docker Compose yapılandırması, bağlantı sorunlarının çözümünde önemli bir rol oynar. `KAFKA_ADVERTISED_LISTENERS` ve `KAFKA_LISTENERS` ayarları, Docker ağı içindeki dahili ve harici iletişimi ayırt edecek şekilde ayarlanır. Bu, Spark ve Kafka gibi aynı Docker ağında çalışan hizmetlerin DNS çözümleme sorunları olmadan etkileşime girebilmesini sağlar. Örneğin, "INSIDE://kafka:9093" eşlemesi dahili kapsayıcıların Kafka'ya erişmesine izin verirken, "OUTSIDE://localhost:9093" izleme araçları gibi harici uygulamaların bağlanmasına olanak tanır.

İkinci komut dosyası, Kafka bağlantısını test etmek için Python 'KafkaConsumer'ın nasıl kullanılacağını gösterir. Bu, Kafka aracısının doğru şekilde çalıştığından emin olmak için basit ama etkili bir yaklaşımdır. Belirtilen konuya ait mesajları tüketerek veri akışının kesintisiz olup olmadığını doğrulayabilirsiniz. Bir kullanıcının borsa verilerini takip etmek istediği bir uygulamayı düşünün. Bağlantının bu tüketici komut dosyası kullanılarak test edilmesi, yapılandırma hataları nedeniyle hiçbir kritik güncellemenin kaçırılmamasını sağlar. Bu araçlarla, gerçek zamanlı veri işleme için sağlam sistemleri güvenle devreye alabilirsiniz! 🚀

# 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()

Docker Ortamında Kafka ve Spark'ı Optimize Etme

arasında sorunsuz iletişim sağlamanın kritik bir yönü Kafka Komisyoncuları Ve Kıvılcım İşçileri Docker'da ağ ayarlarını etkili bir şekilde yapılandırıyor. Docker kapsayıcıları yalıtılmış ortamlarda çalışır ve hizmetlerin etkileşime girmesi gerektiğinde genellikle DNS çözümleme sorunlarına neden olur. Bu sorunu çözmek için Docker Compose'un ağ yapılandırma seçeneklerinden yararlanabilirsiniz. Örneğin, "my_network" gibi özel bir ağ tanımlamak ve hizmetleri bağlamak, kapsayıcıların birbirlerini IP yerine adlarına göre tanımasını sağlar, bu da kurulumu basitleştirir ve sık karşılaşılan tuzaklardan kaçınır.

Bir diğer önemli husus da Kafka'nın dinleyici yapılandırmalarını optimize etmektir. Docker Compose dosyanızda `KAFKA_ADVERTISED_LISTENERS` ve `KAFKA_LISTENERS`'ı belirterek Kafka'nın istemcilerine uygun adresleri tanıtmasına izin vermiş olursunuz. Dahili ve harici dinleyiciler arasındaki bu ayrım, özellikle Spark Workers'ın Docker ağı dışından bağlanmaya çalıştığı durumlarda çatışmaları çözer. Bunun gerçek hayattaki bir örneği, erişim için ayrı bir harici dinleyici gerektiren, bir ana makineden Kafka verilerini sorgulayan bir izleme panosudur. 🔧

Son olarak Spark uygulamalarınızda güçlü hata işlemeyi uygulamak çok önemlidir. Örneğin, Kafka yapılandırmasındaki yeniden denemelerden ve geri dönüşlerden yararlanmak, geçici bağlantı sorunlarını sorunsuz bir şekilde çözebilir. `.option("kafka.consumer.max.poll.records", "500")` eklenmesi, ağır yükler altında bile verimli veri alımını sağlar. Hisse senedi fiyatlarını gerçek zamanlı olarak takip eden üretim düzeyinde bir uygulama hayal edin; arıza güvenlik önlemlerinin mevcut olması, ağ kesintileri sırasında bile kesintisiz veri akışını sağlar. Bu teknikler birlikte güvenilir bir veri işleme hattının omurgasını oluşturur. 🚀