spark structured Streaming 写入时小文件合并的应用代码

在Spark Structured Streaming中，可以通过设置以下参数来减少小文件的数量和大小：

1. `trigger`: 控制作业如何触发。设置较长的触发间隔可以让Spark在处理完更多数据后再写入文件，从而减少小文件数量。

2. `maxFilesPerTrigger`: 每个触发器产生的最大文件数。较高的值会导致更少的文件，但文件大小可能会增加。

3. `checkpointLocation`: 检查点目录，可以在故障恢复时重用数据，减少重新计算的成本。

4. `outputMode`: 输出模式。如果使用`append`模式，Spark将只追加新数据，而不是重写整个文件。这将减少小文件数量。

另外，可以使用`coalesce`或`repartition`操作来合并小文件。这些操作的区别在于`coalesce`尝试将数据合并到较少的分区中，而`repartition`将数据平均分区到指定的数量。

下面是一个简单的示例代码，演示了如何使用`coalesce`和`trigger`来减少小文件数量：

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

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

# 从kafka消费数据
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "topic1") \
  .option("startingOffsets", "earliest") \
  .load()

# 转换数据
df = df.select(col("value").cast("string"))

# 每5分钟写入一个文件，每个文件最多包含1000条记录，并合并小文件
df \
  .writeStream \
  .trigger(processingTime="5 minutes") \
  .option("maxFilesPerTrigger", 1000) \
  .option("checkpointLocation", "/tmp/checkpoint") \
  .format("parquet") \
  .option("path", "/tmp/output") \
  .option("compression", "snappy") \
  .option("mergeSchema", "true") \
  .option("coalesce", "true") \
  .start() \
  .awaitTermination()