Spark Structured Streaming固定间隔/单次微批触发与Parquet Sink异常及小文件问题

一、固定间隔/单次微批触发模式是否支持Parquet文件接收器？ #

Spark 2.4.5的Structured Streaming完全支持用processingTime固定间隔触发或once=True单次触发模式搭配Parquet文件接收器。你遇到的无数据写入问题，并非模式不兼容，大概率是以下配置或数据问题导致：

• Kafka无可用数据：检查目标Topic是否有新数据，同时确认Spark消费的offset配置（比如startingOffsets设为latest但Topic没有新消息，或设为earliest但权限不足无法读取历史数据）。
• 微批无输出数据：如果你的流处理逻辑中存在过滤、聚合等操作，导致当前微批处理后无数据输出，Parquet接收器只会生成用于跟踪状态的_spark_metadata目录，不会写入数据文件。
• Checkpoint异常：检查checkpoint路径的权限是否正常，若之前的任务残留了异常的checkpoint数据，可能导致当前任务无法正常触发写入。可以尝试清理旧的checkpoint目录后重新启动任务。
• 触发间隔设置不合理：如果processingTime间隔设置过小，可能导致微批还未拉取到足够数据就触发；若间隔过大，需要等待更长时间才能看到数据写入。

单次触发模式的正确示例：

from pyspark.sql import SparkSession
from pyspark.sql.streaming import Trigger

spark = SparkSession.builder.appName("KafkaToParquet").getOrCreate()

df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-host:9092") \
    .option("subscribe", "test-topic") \
    .option("startingOffsets", "earliest") \
    .load()

# 解析Kafka消息的示例逻辑
parsed_df = df.selectExpr("CAST(value AS STRING) as content")

query = parsed_df.writeStream \
    .format("parquet") \
    .option("path", "hdfs:///path/to/parquet") \
    .option("checkpointLocation", "hdfs:///path/to/checkpoint") \
    .trigger(Trigger.Once()) \
    .start()

query.awaitTermination()

二、默认触发模式下处理大量小文件的方案 #

默认触发模式（微批间隔为0）会尽可能频繁地处理数据，因此容易生成大量小文件。针对Spark 2.4.5，可通过以下方式解决：

1. 合理设置触发间隔 #

调整processingTime触发间隔，让每个微批处理更多数据，从源头减少小文件数量。例如设置5分钟触发一次：

.trigger(Trigger.ProcessingTime("5 minutes"))

注意：需先解决第一个问题中触发模式无数据的问题，确保该配置能正常写入数据。

2. 重分区控制文件数量 #

在写入前对DataFrame进行重分区，指定每个微批生成的文件数。如果数据量不大，用coalesce避免shuffle；数据量大则用repartition：

# 每个微批生成4个Parquet文件
parsed_df.coalesce(4).writeStream...

3. 配置Spark参数自动合并小文件 #

通过以下参数让Spark自动合并历史小文件：

• spark.sql.streaming.fileSink.log.compactInterval：设置合并元数据日志的间隔（默认10个批次），合并后会触发小文件合并。
• spark.sql.streaming.fileSink.log.cleanupDelay：设置清理旧元数据日志的延迟（默认1小时），确保合并完成后再清理日志。
• spark.sql.files.maxRecordsPerFile：限制每个文件的最大记录数，避免生成过小的文件。

在SparkSession初始化时配置：

spark = SparkSession.builder \
    .appName("KafkaToParquet") \
    .config("spark.sql.streaming.fileSink.log.compactInterval", "5") \
    .config("spark.sql.streaming.fileSink.log.cleanupDelay", "30min") \
    .config("spark.sql.files.maxRecordsPerFile", "100000") \
    .getOrCreate()

4. 离线定时合并小文件 #

如果实时处理无法完全避免小文件，可定时运行批处理任务合并HDFS上的Parquet文件：

# 批处理合并示例
batch_df = spark.read.parquet("hdfs:///path/to/parquet")
batch_df.repartition(10).write.mode("overwrite").parquet("hdfs:///path/to/merged-parquet")

注意：运行前需确保流任务暂时停止，或使用分区写入的方式避免冲突（比如按日期分区，每天合并前一天的分区）。

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