PySpark定时任务如何每次读取1个未处理的avro增量文件

核心实现思路 #

你的场景核心需要解决3个问题：动态新增目录的文件发现、已消费文件的去重、单文件处理的流量控制。整体逻辑是状态驱动的增量拉取，不依赖Spark原生的全目录扫描读取，而是先独立做文件级的元数据遍历，和已消费记录做差集得到待处理文件，每次取1个提交给Spark处理，处理完成后更新消费状态。
目录结构参考如下：

root/2021/12/01/file121.avro
root/2022/06/01/file611.avro
root/2022/06/01/file612.avro
root/2022/06/01/file613.avro
root/2022/06/03/file631.avro
root/2022/06/03/file632.avro
root/2022/06/05/file651.avro
root/2022/06/05/file652.avro
root/2022/06/05/file653.avro

关键策略说明 #

• 文件遍历：直接用Hadoop FileSystem API递归遍历根目录，拉取所有.avro后缀的文件路径，性能远高于Spark读取全目录元数据，天然适配按年/月/日/小时动态新增的多级目录结构
• 状态存储：轻量场景可以直接用HDFS/本地磁盘上的文本文件存储已消费文件的全路径（每行1条），生产环境也可以替换为Hive表、Redis、JDBC数据库，只要支持读写全量已消费路径即可
• 消费顺序：拉取到所有avro文件后，按文件修改时间或者路径中的日期层级排序，保证先生成的文件优先被消费，避免数据乱序
• 完整性保证：只有文件处理逻辑全部执行成功后，才把当前文件路径写入已消费状态列表，任务失败时不更新状态，下次调度自动重试
• 异常文件过滤：遍历文件时过滤掉.tmp/.writing等临时后缀的文件，同时增加文件修改时间阈值（比如最后修改时间超过1分钟才纳入待消费列表），避免读取正在写入的不完整文件

代码参考 #

提交任务时需要提前引入对应Spark版本的spark-avro依赖，以下是可直接运行的PySpark代码：

from pyspark.sql import SparkSession
from py4j.java_gateway import java_import

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("IncrementalAvroConsumer") \
    .getOrCreate()

# 导入Hadoop FileSystem相关类
java_import(spark._jvm, 'org.apache.hadoop.fs.Path')
java_import(spark._jvm, 'org.apache.hadoop.fs.FileSystem')
java_import(spark._jvm, 'org.apache.hadoop.conf.Configuration')

hadoop_conf = spark._jsc.hadoopConfiguration()
fs = spark._jvm.FileSystem.get(hadoop_conf)

# 配置项
ROOT_AVRO_PATH = "hdfs:///path/to/your/avro/root"  # Avro根目录
CONSUMED_STATE_PATH = "hdfs:///path/to/your/consumed_files.txt"  # 已消费文件状态存储路径
FILE_SUFFIX = ".avro"
FILE_READY_DELAY = 60 * 1000  # 文件修改后至少等待1分钟才认为写入完成，单位毫秒

def get_all_avro_files(root_path: str) -> list:
    """递归获取根目录下所有符合要求的已就绪avro文件全路径"""
    avro_files = []
    root_status = fs.listStatus(spark._jvm.Path(root_path))
    for status in root_status:
        if status.isDirectory():
            # 递归遍历子目录
            avro_files.extend(get_all_avro_files(status.getPath().toString()))
        else:
            path = status.getPath().toString()
            # 过滤后缀、临时文件，判断是否达到就绪时间
            if path.endswith(FILE_SUFFIX) \
                and not path.endswith(f".tmp{FILE_SUFFIX}") \
                and (spark._jvm.System.currentTimeMillis() - status.getModificationTime()) > FILE_READY_DELAY:
                avro_files.append((path, status.getModificationTime()))
    # 按修改时间升序排序，保证先产生的文件先消费
    avro_files.sort(key=lambda x: x[1])
    return [f[0] for f in avro_files]

def get_consumed_files(state_path: str) -> set:
    """读取已消费文件列表"""
    if not fs.exists(spark._jvm.Path(state_path)):
        return set()
    input_stream = fs.open(spark._jvm.Path(state_path))
    reader = spark._jvm.java.io.BufferedReader(spark._jvm.java.io.InputStreamReader(input_stream))
    consumed = set()
    line = reader.readLine()
    while line:
        consumed.add(line.strip())
        line = reader.readLine()
    reader.close()
    input_stream.close()
    return consumed

def mark_file_consumed(state_path: str, file_path: str):
    """处理完成后将文件标记为已消费"""
    output_stream = fs.append(spark._jvm.Path(state_path)) if fs.exists(spark._jvm.Path(state_path)) else fs.create(spark._jvm.Path(state_path))
    writer = spark._jvm.java.io.BufferedWriter(spark._jvm.java.io.OutputStreamWriter(output_stream))
    writer.write(file_path + "\n")
    writer.close()
    output_stream.close()

if __name__ == "__main__":
    # 1. 拉取所有avro文件
    all_avro_files = get_all_avro_files(ROOT_AVRO_PATH)
    # 2. 拉取已消费文件
    consumed_files = get_consumed_files(CONSUMED_STATE_PATH)
    # 3. 计算待消费文件
    pending_files = [f for f in all_avro_files if f not in consumed_files]
    
    if not pending_files:
        print("No pending avro files to process, exit.")
        spark.stop()
        exit(0)
    
    # 4. 每次只取第一个待处理文件
    current_process_file = pending_files[0]
    print(f"Start processing file: {current_process_file}")
    
    # 5. 读取avro文件做业务处理
    df = spark.read.format("avro").load(current_process_file)
    # 这里替换为你的实际业务处理逻辑，比如写入数仓、做转换等
    df.show()
    
    # 6. 处理完成后标记为已消费
    mark_file_consumed(CONSUMED_STATE_PATH, current_process_file)
    print(f"File {current_process_file} process success, marked as consumed.")
    
    spark.stop()

生产环境优化点 #

• 如果是多节点并行调度任务，需要给状态文件加分布式锁，避免多实例同时拿到同一个文件重复消费
• 状态文件可以定期归档，避免文件过大导致读取变慢，比如每月初把上月的已消费记录归档到历史表
• 可以增加死信队列逻辑，连续多次处理失败的文件移动到专门的错误目录，避免阻塞后续文件消费
• 如果Avro文件是存放在对象存储（S3、OSS等），替换对应的FileSystem配置即可，逻辑不需要改动

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