无主键时PySpark仅追加新记录至目标分区的实现方案咨询

针对你描述的场景——每日读取按加载日期分区的Parquet源数据，写入按recordTimestamp分区的目标存储，无主键且需替换全量覆盖为增量追加，以下是两种可行方案：

一、基于现有Parquet格式的实现 #

核心逻辑是通过全字段哈希比对识别新记录，结合分区裁剪减少不必要的计算量：

1. 读取源数据并计算哈希值
   每日读取当日/source/Year=x/Month=y/Day=z的Parquet数据，对每条记录的所有字段计算唯一哈希值（推荐用SHA-256，避免冲突）。Spark示例代码：

   from pyspark.sql.functions import sha2, struct
   
   source_df = spark.read.parquet("/source/Year=2024/Month=05/Day=20")
   source_with_hash = source_df.withColumn("record_hash", sha2(struct(*source_df.columns), 256))
   

2. 裁剪目标分区并读取现有哈希
   提取源数据中recordTimestamp覆盖的年/月/日范围，只读取目标存储中对应分区的现有数据，同样计算哈希值：

   # 获取源数据recordTimestamp的分区范围
   min_ts = source_with_hash.selectExpr("date_trunc('day', min(recordTimestamp))").first()[0]
   max_ts = source_with_hash.selectExpr("date_trunc('day', max(recordTimestamp))").first()[0]
   # 读取对应分区的目标数据
   target_partitions_df = spark.read.parquet("/destination") \
       .filter(f"Year between {min_ts.year} and {max_ts.year}") \
       .filter(f"Month between {min_ts.month} and {max_ts.month}") \
       .filter(f"Day between {min_ts.day} and {max_ts.day}")
   target_hashes = target_partitions_df.withColumn("record_hash", sha2(struct(*target_partitions_df.columns), 256)) \
       .select("record_hash")
   

3. 筛选新记录
   使用左反连接，保留源数据中哈希值不存在于目标对应分区的记录：

   new_records = source_with_hash.join(target_hashes, on="record_hash", how="left_anti") \
       .drop("record_hash")  # 移除临时哈希列
   

4. 追加写入目标分区
   先从recordTimestamp提取年/月/日列，再按分区以追加模式写入：

   new_records = new_records.withColumn("Year", new_records.recordTimestamp.year) \
       .withColumn("Month", new_records.recordTimestamp.month) \
       .withColumn("Day", new_records.recordTimestamp.day)
   
   new_records.write.mode("append") \
       .partitionBy("Year", "Month", "Day") \
       .parquet("/destination")
   

二、基于Delta Lake的优化方案（推荐） #

Delta Lake提供ACID事务、自动分区管理和优化能力，更适合大规模数据的增量处理：

1. 初始化目标Delta表
   将目标存储转换为Delta表，按recordTimestamp的年/月/日分区：

   # 首次运行：转换现有目标数据为Delta表（若已有数据）
   spark.read.parquet("/destination").write.mode("overwrite") \
       .partitionBy("Year", "Month", "Day") \
       .format("delta") \
       .save("/destination")
   

2. 每日增量写入
   读取当日源数据并计算哈希值，通过Delta的MERGE语法实现仅插入新记录：

   from delta.tables import DeltaTable
   
   # 读取源数据并添加哈希、分区列
   source_df = spark.read.parquet("/source/Year=2024/Month=05/Day=20")
   source_with_hash = source_df.withColumn("record_hash", sha2(struct(*source_df.columns), 256)) \
       .withColumn("Year", source_df.recordTimestamp.year) \
       .withColumn("Month", source_df.recordTimestamp.month) \
       .withColumn("Day", source_df.recordTimestamp.day)
   
   # 加载目标Delta表
   delta_table = DeltaTable.forPath(spark, "/destination")
   
   # MERGE逻辑：哈希不匹配时插入新记录
   delta_table.alias("target") \
       .merge(
           source_with_hash.alias("source"),
           "target.record_hash = source.record_hash"
       ) \
       .whenNotMatchedInsertAll() \
       .execute()
   

   可开启自动优化和压缩，减少小文件并提升性能：

   # 全局配置自动优化
   spark.conf.set("spark.databricks.delta.autoOptimize.enabled", "true")
   spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")
   

关键注意事项 #

• 哈希冲突规避：使用SHA-256等强哈希算法，10亿级数据下冲突概率可忽略；若需更严谨，可同时计算多个哈希值作为匹配条件。
• 性能优化：始终通过recordTimestamp裁剪目标分区，避免全表扫描；在Synapse中选择适配的Spark池节点规格，利用并行处理宽表数据。
• 宽表处理：计算哈希时用struct(*df.columns)自动包含所有120+字段，无需手动列举。

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