Spark Structured Streaming多Sink输出不一致问题排查及代码优化咨询

一、数据丢失问题排查 #

1. 核心逻辑错误：流等待方式不对 #

你的代码最后串行调用ds.awaitTermination()和dsf.awaitTermination()，但第一个awaitTermination()会一直阻塞程序，直到console流停止，导致Kafka流的awaitTermination()永远不会执行。当console流终止后，程序直接退出，Kafka流可能还没写完所有数据，这是部分数据丢失的关键原因。

2. UDF异常与null值处理 #

• 检查function_name是否在处理失败时返回null：Kafka Sink默认会丢弃value为null的记录，但Console Sink会正常输出null。如果8条成功处理的数据中有2条返回null，就会出现Console输出8条但Kafka只写入6条的情况。
• 确认UDF是否未捕获异常：如果UDF抛出未处理的异常，Spark可能会跳过该条数据，但Console和Kafka流的容错策略不一致，也会导致数据计数差异。

3. Checkpoint配置验证 #

确保Kafka流的checkpointLocation是独立且首次运行时不存在的目录，如果目录已存在且包含旧偏移信息，可能导致流重复消费或跳过数据。

二、代码优化建议 #

1. 修复流等待逻辑 #

使用spark.streams.awaitAnyTermination()替代串行等待，或者用线程分别监控两个流：

ds.start()
dsf.start()

# 等待任意一个流终止，Spark 3.0+也可用awaitAllTermination()等待所有流结束
spark.streams.awaitAnyTermination()

2. 规范UDF定义与错误处理 #

• 明确指定UDF返回类型，避免Spark推断错误：

from pyspark.sql.types import StringType

# 直接传入函数，无需lambda包装
calludf = F.udf(function_name, StringType())

• 在UDF中添加异常捕获，记录错误并标记异常数据：

import logging
logging.basicConfig(level=logging.INFO)

def function_name(x):
    try:
        # 你的业务处理逻辑
        return processed_result
    except Exception as e:
        logging.error(f"处理数据失败: {x}, 错误信息: {str(e)}")
        # 返回明确的错误标记，而非null
        return f"ERROR_PROCESSING: {str(e)}"

3. 优化数据处理流程 #

合并冗余操作，过滤无效数据：

df = dfraw.select(F.col('value').cast('string').alias('value')) \
          .withColumn('value', calludf(F.col('value'))) \
          # 可选：过滤处理失败的记录（如果不需要写入错误数据）
          .filter(F.col('value').isNotNull() & ~F.col('value').startswith('ERROR_PROCESSING'))

4. 完善流配置 #

• 为每个流配置独立的Checkpoint目录（包括Console流，如果需要容错）：

ds = df.selectExpr("CAST(value AS STRING)") \
    .writeStream \
    .format('console') \
    .option('truncate', False) \
    .option("checkpointLocation", "/path/to/console_checkpoint") \
    .start()

dsf = df.selectExpr("CAST(value AS STRING)") \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", KAFKA_CONSUMER_IP) \
    .option("topic", topic_name_two) \
    .option("checkpointLocation", "/path/to/kafka_checkpoint") \
    .option("kafka.retries", "3") \  # 添加重试配置，避免临时网络问题丢数据
    .option("kafka.request.timeout.ms", "30000") \
    .start()

5. 数据验证 #

添加临时视图方便实时查看数据量：

df.createOrReplaceTempView("processed_data")
# 在Spark Shell中可执行：SELECT COUNT(*) FROM processed_data 验证数据量

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