在 Docker 设置中解决 Spark Worker 与 Kafka 的连接问题

处理 Spark Worker 和 Kafka Broker 之间的连接问题

解决方案 1：使用 Python 调试并解决 Spark 和 Kafka 与 Docker 中的连接问题

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

调试 Dockerized Kafka 中的 DNS 解析问题

解决方案 2：修改 Docker Compose 配置以实现正确的 DNS 解析

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

测试 Kafka 消费者连接

解决方案 3：用于测试连接的 Python Kafka Consumer

# 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 中的 Spark 和 Kafka 的常见问题

1. 目的是什么 KAFKA_ADVERTISED_LISTENERS？
2. 它指定 Kafka 客户端连接的通告地址，确保 Docker 网络内外的正确通信。
3. 如何在 Docker Compose 中定义自定义网络？
4. 您可以在下面添加网络 networks key 并将其包含在服务中，例如`networks: my_network`。
5. 为什么 Docker 容器中的 DNS 解析失败？
6. 容器可能无法通过名称来识别彼此，除非它们属于链接其 DNS 的同一个 Docker 网络。
7. 的作用是什么 .option("subscribe", "topic") 在 Spark 流中？
8. 它将 Spark Structured Streaming DataFrame 订阅到指定的 Kafka 主题以进行实时数据摄取。
9. 重试如何改进 Kafka-Spark 集成？
10. 重试配置，例如 max.poll.records，帮助处理瞬态错误并确保数据处理的一致性。

简化 Spark 和 Kafka 集成

在 Docker 中设置 Spark 和 Kafka 可能很复杂，但通过正确的配置，它变得易于管理。重点关注侦听器设置和网络配置以避免连接问题。确保 Zookeeper 和 Kafka 等所有组件都良好同步，以获得最佳性能。

现实世界的用例，例如监控财务数据或物联网流，凸显了稳健配置的重要性。此处共享的工具和脚本使您具备克服常见障碍并构建高效、实时数据管道的知识。 🛠️