Spark Structured Streaming中cleanSource配置不生效问题求助

在AWS Glue 3.0（对应Spark 3.1.1）环境中，使用Spark Structured Streaming的trigger(once=True)模式将S3上的Parquet表流式写入Delta表时，已配置cleanSource=archive和sourceArchiveDir，但源Parquet文件未被完整归档。省略sourceArchiveDir时JVM会抛出参数缺失异常，说明配置已被正确读取；将spark.sql.streaming.fileSource.cleaner.numThreads设为0强制主线程执行清理，仍出现部分归档、进程提前终止的情况。

1. 线程执行时序问题：Spark的文件清理/归档操作由独立JVM线程负责，trigger(once=True)模式下，流式任务完成数据处理后会立即标记流为完成状态，PySpark的Python进程会在所有活跃流终止后直接退出，此时JVM侧的清理线程可能还未完成S3上的文件移动（S3对象存储的文件移动本质是复制+删除，耗时可能长于数据处理）。
2. Glue环境资源回收：Glue任务主逻辑执行完毕后，服务会快速回收资源，不会等待后台清理线程完成收尾操作。

方案1：增加主动等待时间（临时 workaround） #

在所有流终止后，增加一段等待时间，给JVM足够时间完成S3文件归档，时间需根据文件大小、数量调整：

import time

def wait_for_streams(spark: SparkSession):
    while len(spark.streams.active) > 0:
        spark.streams.awaitAnyTermination()
    # 按需调整等待时长，示例为60秒
    time.sleep(60)

方案2：强制主线程同步执行清理（推荐） #

通过Spark配置强制清理逻辑在主线程同步完成，并主动触发日志清理操作：

1. 创建SparkSession时添加配置：

spark = SparkSession.builder \
    .config("spark.sql.streaming.fileSource.cleaner.numThreads", "0") \
    .config("spark.sql.streaming.fileSource.log.cleanupDelay", "0") \
    .config("spark.sql.streaming.fileSource.log.compactInterval", "1") \
    .getOrCreate()

2. 修改wait_for_streams函数，主动触发清理：

from py4j.java_gateway import java_import

def wait_for_streams(spark: SparkSession):
    while len(spark.streams.active) > 0:
        spark.streams.awaitAnyTermination()
    
    # 调用Spark内部API强制执行文件清理
    java_import(spark._jvm, "org.apache.spark.sql.execution.streaming.FileStreamSource")
    spark._jvm.FileStreamSource.cleanupOldLogFiles(spark._jsparkSession)

该方式确保所有归档操作完成后，Python进程才会退出。

方案3：改用连续处理模式（特定场景适用） #

若业务允许，将trigger(once=True)改为连续处理模式，但需调整任务触发逻辑，仅适合需要持续运行的场景，复杂度较高。

1. 任务运行后，检查源目录文件是否全部移动至归档目录；
2. 查看Glue任务日志，确认无"File cleanup not completed before process exit"类警告；
3. 可选：在任务结束前打印清理状态，确认文件处理完成：

# 流终止后添加
# 查看处理文件数量
source_log = spark._jvm.org.apache.spark.sql.execution.streaming.FileStreamSource.Log(
    spark._jsparkSession, source, spark._jsparkSession.sessionState().newHadoopConfiguration()
)
print(f"已处理文件数: {source_log.get().length()}")

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