如何在PySpark中通过for循环将多个parquet文件加载到Delta表

实现思路 #

• 先通过Hadoop API获取指定路径下所有符合命名规则的parquet文件路径，过滤掉目录下的非数据文件
• 遍历文件路径列表，逐个读取单文件为DataFrame，可按需对单文件数据做自定义预处理
• 采用追加模式逐次写入Delta表，适合内存不足无法一次性加载全量文件、或需要按文件粒度做处理校验的场景

完整代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import input_file_name

# 初始化支持Delta的SparkSession
spark = SparkSession.builder \
    .appName("ParquetBatchLoadToDelta") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 自定义配置项，替换为你的实际路径
parquet_root_dir = "hdfs:///your/parquet/file/dir"
delta_save_path = "hdfs:///your/delta/table/path"
delta_table_name = "default.my_target_delta_table"

# 步骤1：获取目录下所有符合要求的parquet文件路径
hadoop_conf = spark._jsc.hadoopConfiguration()
fs_path = spark._jvm.org.apache.hadoop.fs.Path(parquet_root_dir)
fs = fs_path.getFileSystem(hadoop_conf)
all_file_status = fs.listStatus(fs_path)

# 过滤part开头、snappy.parquet结尾的分区文件，可按需调整过滤规则
target_parquet_files = []
for file_status in all_file_status:
    file_name = file_status.getPath().getName()
    if file_name.startswith("part-") and file_name.endswith(".snappy.parquet"):
        target_parquet_files.append(file_status.getPath().toString())

print(f"共扫描到待处理文件数：{len(target_parquet_files)}")

# 步骤2：循环逐个加载写入Delta表
for index, file_path in enumerate(target_parquet_files):
    print(f"正在处理第{index+1}/{len(target_parquet_files)}个文件：{file_path}")
    # 读取单个parquet文件
    df = spark.read.parquet(file_path)
    # 可选：添加源文件路径字段方便后续数据溯源
    df = df.withColumn("source_file_path", input_file_name())

    # 写入Delta表，首次执行自动创建表，后续执行自动追加
    df.write.format("delta") \
        .mode("append") \
        .save(delta_save_path)

# 可选：将Delta表注册到Hive元数据
spark.sql(f"""
CREATE TABLE IF NOT EXISTS {delta_table_name}
USING DELTA
LOCATION '{delta_save_path}'
""")

# 可选：验证写入结果
total_count = spark.read.format("delta").load(delta_save_path).count()
print(f"Delta表写入完成，总数据量：{total_count}")

spark.stop()

注意事项 #

• 版本适配：Spark 3.0对应Delta Core版本为0.7.0/0.8.0，提交任务时需携带对应依赖，示例提交命令：

  spark-submit --packages io.delta:delta-core_2.12:0.7.0 your_script_name.py
  

• 小文件优化：逐个写入会产生较多Delta小文件，所有文件处理完成后可执行优化合并小文件：

  from delta.tables import DeltaTable
  delta_table = DeltaTable.forPath(spark, delta_save_path)
  delta_table.optimize().execute()
  

• 容错优化：若需要支持断点续传，可将已处理的文件路径存入临时表或本地文件，每次循环前先判断文件是否已处理，避免中断后重复写入
• 若无单文件特殊处理需求，可直接读取整个目录一次性写入Delta，性能更高：

  spark.read.parquet(parquet_root_dir).write.format("delta").save(delta_save_path)
  

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