在PySpark中如何跟踪已处理数据？适配每日多触发等场景

针对你描述的场景（每日大量文件、管道多次触发、追加写入Parquet数据湖），以下是几种实用的跟踪已处理数据的方法：

1. 维护处理元数据记录表 #

这是最可靠的方案，核心是用一张专门的表记录所有已处理文件的关键信息，每次运行时通过对比筛选出未处理文件。

实现步骤： #

• 元数据记录表字段：file_path（文件全路径）、process_time（处理时间）、batch_id（批次标识），可存储为Parquet格式放在数据湖，或存入关系型数据库。
• 每次处理流程：
  1. 列出当日源文件夹下的所有文件路径；
  2. 读取元数据记录表，过滤出当日已处理的文件；
  3. 通过左连接求差集，得到未处理文件列表；
  4. 处理未处理文件，完成后将新的文件记录写入元数据记录表。

PySpark代码示例： #

from pyspark.sql import functions as F
from datetime import date

# 当日日期，用于定位源文件夹和过滤元数据
today = date.today().strftime("%Y-%m-%d")
source_dir = f"/path/to/source/{today}/"
metadata_path = "/path/to/metadata/processed_files.parquet"

# 1. 获取当日源文件夹下的所有文件
source_files_df = spark.read.format("binaryFile")\
    .option("pathGlobFilter", "*.*")\
    .option("recursiveFileLookup", "true")\
    .load(source_dir)\
    .select("path")\
    .withColumn("processing_date", F.lit(today))

# 2. 读取已处理元数据（若元数据不存在则创建空DF）
try:
    processed_df = spark.read.parquet(metadata_path)\
        .filter(F.col("processing_date") == today)
except:
    processed_df = spark.createDataFrame([], source_files_df.schema)

# 3. 筛选未处理文件
unprocessed_files_df = source_files_df.join(processed_df, on="path", how="left_anti")

# 4. 处理未处理文件（替换为你的实际业务逻辑）
if unprocessed_files_df.count() > 0:
    raw_data_df = spark.read.format("your_source_format")\
        .load(unprocessed_files_df.select("path").rdd.flatMap(lambda x: x).collect())
    
    # 追加写入数据湖的Parquet文件
    raw_data_df.write.mode("append").parquet("/path/to/data_lake/target.parquet")
    
    # 记录新的处理元数据
    processed_batch_df = unprocessed_files_df.withColumn("process_time", F.current_timestamp())\
        .withColumn("batch_id", F.lit(f"batch_{today}_{spark.sparkContext.applicationId}"))
    processed_batch_df.write.mode("append").parquet(metadata_path)

2. 利用文件元数据（修改/创建时间） #

如果文件生成后不会被修改，且无需严格去重（比如不会重复上传完全相同的文件），可以通过文件的修改时间来增量筛选：

实现思路： #

• 每次处理完成后，记录本次处理的结束时间；
• 下次运行时，只读取源文件夹中修改时间晚于上次处理时间的文件；
• 无需额外维护元数据表，适合场景简单的情况。

PySpark代码示例： #

from datetime import datetime, timedelta
from pyspark.sql import functions as F

# 模拟获取上次处理时间（实际可存在本地文件或配置系统中）
last_process_time = datetime.now() - timedelta(hours=1)
source_dir = f"/path/to/source/{date.today().strftime('%Y-%m-%d')}/"

# 读取修改时间晚于上次处理时间的文件
unprocessed_files_df = spark.read.format("binaryFile")\
    .option("pathGlobFilter", "*.*")\
    .load(source_dir)\
    .filter(F.col("modificationTime") > F.lit(last_process_time.timestamp() * 1000))  # Spark使用毫秒时间戳

# 后续处理逻辑同方案1...

3. 文件哈希校验（防重复处理相同内容） #

如果存在文件重传但内容不变的情况，可以计算文件的哈希值，确保相同内容的文件不会被重复处理：

实现步骤： #

• 在元数据记录表中增加file_hash字段；
• 读取源文件时，计算每个文件的哈希值（如MD5）；
• 对比哈希值，仅处理哈希值未在元数据中出现的文件。

PySpark代码片段： #

# 计算文件哈希值的函数
def calculate_file_hash(file_path):
    import hashlib
    with open(file_path, "rb") as f:
        return hashlib.md5(f.read()).hexdigest()

# 注册UDF
hash_udf = F.udf(calculate_file_hash)

# 获取源文件并计算哈希
source_files_df = spark.read.format("binaryFile")\
    .load(source_dir)\
    .select("path", hash_udf("path").alias("file_hash"))

# 对比元数据中的哈希值筛选未处理文件...

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