PySpark如何基于文件系统事件持续读取新增二进制文件？

一、PySpark原生的文件处理能力 #

PySpark本身没有原生支持基于文件系统事件（如inotify）的触发机制，它的结构化流处理（Structured Streaming）处理文件源时，本质还是基于短轮询（默认10秒间隔，可配置）检测新增文件，并非纯事件驱动。

不过结构化流处理可以满足你处理新增文件+访问历史数据的需求：

• 若需关联历史数据，可将历史处理结果写入Delta Lake、Hive表这类可查询存储，流处理过程中随时读取进行关联计算。
• 结构化流会通过检查点机制自动跟踪已处理文件，不会重复处理，默认处理目录下所有未处理文件，后续新增文件也会被自动检测。

你可以把批处理代码改造为流处理模式，示例如下：

from pyspark.sql import SparkSession

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

# 构建流读取数据源
stream_df = spark.readStream.format("binaryFile")\
    .option("pathGlobFilter", "*.pcap.tgz")\
    .option("compression", "gzip")\
    .load(folder)

# 这里添加解包、提取IP-UDP数据包的处理逻辑
processed_df = ... # 你的业务处理代码

# 自定义批次处理函数，完成结果写入+文件移动
def handle_batch(batch_df, batch_id):
    # 写入处理结果到存储
    batch_df.write.mode("append").save("/path/to/result_store")
    # 获取当前批次的文件路径，移动到done文件夹
    file_paths = batch_df.select("path").rdd.map(lambda row: row.path).collect()
    import shutil
    import os
    done_folder = "/path/to/done"
    os.makedirs(done_folder, exist_ok=True)
    for path in file_paths:
        dest_path = os.path.join(done_folder, os.path.basename(path))
        shutil.move(path, dest_path)

# 启动流任务
query = processed_df.writeStream\
    .option("checkpointLocation", "/path/to/checkpoint")\
    .foreachBatch(handle_batch)\
    .start()

query.awaitTermination()

二、纯事件驱动方案：借助外部工具 #

如果完全不想用轮询，必须基于文件系统事件触发，可依赖外部工具：

• inotify-tools（Linux）：编写Shell/Python脚本监听目标文件夹的文件创建事件，触发PySpark批处理任务，处理完成后移动文件到done目录。
• Apache NiFi：通过内置的文件监听组件配置事件触发流程，自动调用PySpark处理，完成后将文件移动到指定目录，全程事件驱动。
• Apache Flume：配置Spooling Directory Source监听文件夹，将文件数据传递给Spark处理，处理完成后自动标记或移动文件。

三、历史数据访问的适配方案 #

不管用结构化流还是外部工具触发的批处理，都可以通过以下方式满足历史数据访问需求：

• 将所有处理结果统一存储到Delta Lake、Iceberg这类支持ACID的湖仓，后续任务随时读取历史数据进行关联分析。
• 批处理模式下，每次触发任务时，可根据业务需要读取历史结果数据，和新文件数据一起计算。

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