Spark微批处理周期内无法消费Kafka全部数据的问题排查

你的代码里有几个关键问题导致无法处理所有消息：

• 错误的消息处理逻辑：你用rdd.collectAsMap()把整个RDD的所有键值对转换成一个字典，然后把这个字典的字符串作为单条消息发送到Kafka。这意味着每个微批（不管里面有多少条原始消息）只会向topic "two"发送一条消息，而不是每条原始消息单独发送。
• 使用了已弃用的Spark Streaming Kafka API：KafkaUtils.createStream是基于ZooKeeper的旧API，它在偏移量管理和可靠性上存在缺陷，而且已经被官方弃用，推荐使用Direct Stream API。
• Driver端发送消息的风险：你在Driver初始化Kafka Producer，然后在foreachRDD里直接使用，这种方式不仅会把所有数据拉到Driver（导致性能瓶颈和数据丢失风险），还可能因为序列化问题引发异常。

下面是修复后的代码，针对每个问题进行了改进：

1. 使用新的Direct Stream API #

Direct Stream API直接使用Kafka的消费者API管理偏移量，不需要依赖ZooKeeper，能更好地保证消息不丢失。

2. 正确处理每条消息 #

在Executor端对每条消息单独处理，用foreachPartition为每个分区创建一个Kafka Producer（避免重复初始化的开销），然后逐条发送消息。

修复后的Spark Streaming代码 #

# Read data from topic "one" and write each message to topic "two"
import sys
import os

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from kafka import KafkaProducer

# Direct API需要使用Kafka的bootstrap servers，而非ZooKeeper地址
kafka_bootstrap_servers = "192.168.106.214:9092,192.168.106.213:9092"
input_topic = "one"
output_topic = "two"

def send_to_kafka(partition):
    # 为每个分区创建一个Kafka Producer，减少初始化开销
    producer = KafkaProducer(bootstrap_servers=kafka_bootstrap_servers, client_id="spark-producer-test4")
    try:
        for record in partition:
            # record是(key, value)元组，这里发送value部分（需要key的话可一起发送）
            producer.send(output_topic, value=record[1].encode('utf-8'))
        producer.flush()
    except Exception as e:
        print(f"Exception while sending messages: {str(e)}")
        print(f"[warning] Unable to send data to topic {output_topic}")
    finally:
        producer.close()

if __name__ == "__main__":
    sc = SparkContext(appName="KafkaStreamProcessor")
    ssc = StreamingContext(sc, 1)  # 1秒微批间隔，可根据数据量调整

    # 使用Direct Stream API创建输入流
    input_stream = KafkaUtils.createDirectStream(
        ssc,
        [input_topic],
        {
            "metadata.broker.list": kafka_bootstrap_servers,
            "group.id": "spark-consumer-test4",
            "auto.offset.reset": "latest"  # 可根据需求改为"earliest"
        }
    )

    # 对每个RDD的分区逐一处理消息
    input_stream.foreachRDD(lambda rdd: rdd.foreachPartition(send_to_kafka))

    ssc.start()
    ssc.awaitTermination()

• 偏移量管理：如果需要Exactly-Once语义，可以手动管理Kafka偏移量，将偏移量存储到HDFS或数据库中，在处理完RDD后再提交偏移量。
• 微批间隔调整：微批间隔（当前为1秒）可根据数据量和处理能力调整，太小会增加调度开销，太大则会提升延迟。
• 切换到Spark Structured Streaming：Spark Streaming已进入维护模式，官方推荐使用Structured Streaming，它提供更简洁的API和更可靠的流处理语义，以下是参考版本：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

if __name__ == "__main__":
    spark = SparkSession.builder.appName("StructuredKafkaProcessor").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")

    # 从Kafka读取数据
    df = spark.readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "192.168.106.214:9092,192.168.106.213:9092") \
        .option("subscribe", "one") \
        .option("group.id", "spark-structured-consumer") \
        .load()

    # 将数据写入Kafka的topic two
    query = df.select(col("key"), col("value")) \
        .writeStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "192.168.106.214:9092,192.168.106.213:9092") \
        .option("topic", "two") \
        .option("checkpointLocation", "/tmp/kafka-checkpoint")  # 必须设置checkpoint目录保证容错
        .start()

    query.awaitTermination()

内容的提问来源于stack exchange，提问作者chandan