如何用PySpark读取目录特定文件并优化每日增量加载逻辑？

核心思路 #

• 利用Spark分布式文件处理能力替代Driver端Python循环，提升效率
• 通过Spark SQL左反连接直接筛选未加载文件，避免将全量已加载文件拉到Driver端（规避collect()带来的性能风险）
• 批量读取符合条件的文件，减少IO操作次数

具体实现步骤 #

1. 读取目录下所有Excel文件的元数据 #

用Spark内置的binaryFile格式获取文件路径，同时提取文件名：

from pyspark.sql import functions as F

# 获取目录下所有xlsx文件的路径和文件名
all_files_df = spark.read.format("binaryFile") \
    .option("pathGlobFilter", "*.xlsx") \
    .load(base_source_path) \
    .withColumn("filename", F.element_at(F.split(F.col("path"), "/"), -1)) \
    .select("path", "filename")

2. 拆分两类文件的DataFrame #

通过文件名关键字过滤，区分BookingRequests和Bookings文件：

# 筛选BookingRequests类型文件
booking_req_files_df = all_files_df.filter(F.col("filename").contains("BookingRequests"))

# 筛选Bookings类型文件
bookings_files_df = all_files_df.filter(F.col("filename").contains("Bookings"))

3. 获取已加载文件的记录 #

直接从目标表读取已加载的文件名，全程在Spark分布式环境处理：

# 从booking_requests表读取已加载的文件名
loaded_req_df = spark.sql(f"select distinct filename from {booking_requests_table}")

# 从bookings表读取已加载的文件名
loaded_book_df = spark.sql(f"select distinct filename from {bookings_table}")

4. 筛选新增未加载文件 #

用左反连接（left anti join）筛选出从未加载过的文件路径：

# 得到待加载的BookingRequests文件路径
new_req_files_df = booking_req_files_df.join(loaded_req_df, on="filename", how="left_anti")
new_req_paths = new_req_files_df.select("path").rdd.flatMap(lambda x: x).collect()

# 得到待加载的Bookings文件路径
new_book_files_df = bookings_files_df.join(loaded_book_df, on="filename", how="left_anti")
new_book_paths = new_book_files_df.select("path").rdd.flatMap(lambda x: x).collect()

5. 批量读取并写入新增文件 #

如果存在新增文件，直接传入路径列表批量读取，同时自动记录文件名到表中：

# 加载并写入新增BookingRequests数据
if new_req_paths:
    df_req = spark.read.format("com.crealytics.spark.excel") \
        .option("header", "true") \
        .load(new_req_paths)
    # 自动添加文件名字段，用于后续追踪已加载文件
    df_req_with_filename = df_req.withColumn("filename", F.element_at(F.split(F.input_file_name(), "/"), -1))
    df_req_with_filename.write.mode("append").saveAsTable(booking_requests_table)

# 加载并写入新增Bookings数据
if new_book_paths:
    df_book = spark.read.format("com.crealytics.spark.excel") \
        .option("header", "true") \
        .load(new_book_paths)
    df_book_with_filename = df_book.withColumn("filename", F.element_at(F.split(F.input_file_name(), "/"), -1))
    df_book_with_filename.write.mode("append").saveAsTable(bookings_table)

关键优化点 #

• 消除Driver端循环：用Spark分布式计算完成文件筛选，降低Driver内存压力
• 控制collect()范围：仅在最后筛选出新增文件路径时调用collect()，此时数据量远小于全量文件列表
• 批量读取优化：Spark支持直接传入路径列表，自动优化读取任务，比循环单文件读取效率提升明显
• 自动记录文件名：通过input_file_name()函数自动获取文件名，避免手动拼接出错

注意事项 #

• 确保集群已安装com.crealytics.spark.excel依赖包
• 若目录包含子目录，可添加.option("recursiveFileLookup", "true")开启递归查找
• 若文件名包含特殊分隔符，可改用正则提取文件名：

  .withColumn("filename", F.regexp_extract(F.col("path"), ".*/(.*)$", 1))
  

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