Spark availableNow触发器无法归档源文件问题咨询

我使用Spark 3.4.1读取S3存储桶中按yyyy/mm/dd路径模式每日生成的JSON文件，将其转换为Iceberg格式，JSON文件和Iceberg数据存储在该桶的不同路径下。

流读取器配置如下：

jsondf = spark.readStream.format("json").schema(myschema) \
    .option("cleanSource", "archive") \
    .option("sourceArchiveDir", "s3a://mybucket/myarchivepath") \
    .load("s3a://mybucket/sourcefolder/yyyy/mm/dd") \
    .select("*")

连续流写入器可正常工作，文件出现时能完成归档。但由于文件量不大，我希望使用availableNow=true触发器（Spark 3.4.1中Once触发器已被弃用），配置如下：

jsondf.writeStream.trigger(availableNow=True) \
    .format("iceberg") \
    .option("checkpointLocation", "s3a://mybucket/chkpointfolder") \
    .outputMode("append") \
    .start(jsontable)

使用触发器时归档失败，甚至未观察到流读取器读取任何文件，请问原因是什么？

1. 路径匹配逻辑问题 #

你指定的load路径是静态字符串s3a://mybucket/sourcefolder/yyyy/mm/dd，这会导致Spark仅扫描该固定路径下的文件，而非每日生成的新日期子目录。即使连续流模式下能工作，也可能是因为启动时该路径恰好有文件，后续新日期的文件无法被扫描到。

解决办法：
使用通配符匹配所有日期路径，并开启递归文件查找：

jsondf = spark.readStream.format("json").schema(myschema) \
    .option("cleanSource", "archive") \
    .option("sourceArchiveDir", "s3a://mybucket/myarchivepath") \
    .option("recursiveFileLookup", "true") \
    .load("s3a://mybucket/sourcefolder/*/*/*") \
    .select("*")

2. Checkpoint目录的状态干扰 #

availableNow触发器依赖checkpoint目录跟踪已处理的文件，如果之前的流运行（比如连续流模式）已经将目标文件标记为已处理，新启动的availableNow流会直接跳过这些文件，表现为“未读取任何文件”。

解决办法：
删除旧的checkpoint目录s3a://mybucket/chkpointfolder，重新启动流，确保触发器能扫描到所有未处理的文件。

3. availableNow与cleanSource=archive的执行时序问题 #

availableNow模式会在处理完所有可用数据后立即停止流，而cleanSource=archive的归档操作可能需要依赖流的持续运行状态来完成文件移动。如果流终止过快，归档逻辑可能来不及执行。

解决办法：
添加maxFilesPerTrigger参数，限制每次触发器处理的文件数量，让归档操作有足够时间完成：

jsondf.writeStream.trigger(availableNow=True) \
    .format("iceberg") \
    .option("checkpointLocation", "s3a://mybucket/chkpointfolder") \
    .option("maxFilesPerTrigger", 500)  # 根据实际文件量调整
    .outputMode("append") \
    .start(jsontable)

4. Iceberg流写入的状态管理 #

Iceberg的流写入在availableNow模式下，可能需要额外配置来确保状态正确提交。可以尝试添加Iceberg的元数据清理参数，避免元数据堆积影响状态识别：

jsondf.writeStream.trigger(availableNow=True) \
    .format("iceberg") \
    .option("checkpointLocation", "s3a://mybucket/chkpointfolder") \
    .option("write.metadata.delete-after-commit.enabled", "true") \
    .outputMode("append") \
    .start(jsontable)

5. 日志排查 #

开启Spark的DEBUG级日志，重点查看org.apache.spark.sql.execution.streaming.FileStreamSource和org.apache.spark.sql.execution.streaming.StreamExecution相关日志，确认流启动时是否扫描到文件，以及处理过程中是否有归档失败的错误信息，定位具体问题环节。

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