在Databricks中用PySpark实现Delta表Incremental Append写入及相关差异咨询

1. 如何在Databricks环境中使用PySpark执行Incremental Append操作，并将数据写入Delta表 #

Incremental Append指仅将新增数据源记录追加到目标Delta表，不修改已有数据，操作流程如下：

• 筛选增量数据：通过时间戳、自增ID或CDC标记获取本次需追加的新数据。例如读取近24小时的新增数据：

  # 假设源表含create_time字段标记数据生成时间
  incremental_df = spark.read.table("source_table") \
    .filter("create_time >= current_timestamp() - interval 24 hours")
  

• 执行Append写入：使用Delta格式写入器，指定mode("append")完成追加：

  # 写入到指定路径的Delta表
  incremental_df.write.format("delta") \
    .mode("append") \
    .save("/path/to/delta_table")
  
  # 或写入到Databricks元数据管理的表
  incremental_df.write.format("delta") \
    .mode("append") \
    .saveAsTable("database.target_delta_table")
  

• 注意事项：若目标表有分区，可通过partitionBy字段优化写入性能；需保证Exactly-Once语义时，可开启option("txnAppId", "unique_app_id")和option("txnVersion", str(version))。

2. Incremental Append与Incremental Upsert之间的区别 #

两者核心差异在于对已有数据的处理逻辑：

• Incremental Append：仅追加增量数据，不检查或修改已有记录。即使增量数据存在与目标表主键重复的记录，也会直接插入，导致数据重复。适合日志、事件流等无需去重或更新的场景，操作逻辑简单，性能开销低。
• Incremental Upsert（Merge）：同时处理新增与已有数据——对增量中与目标表主键匹配的记录执行更新，对不匹配的记录执行插入。需通过Delta Lake的mergeInto API实现，确保数据唯一性，适合用户信息、交易记录等需更新状态的场景。示例代码片段：

  from delta.tables import DeltaTable
  
  delta_table = DeltaTable.forPath(spark, "/path/to/delta_table")
  incremental_df = spark.read.table("source_increment")
  
  delta_table.alias("target") \
    .merge(incremental_df.alias("source"), "target.id = source.id") \
    .whenMatchedUpdate(set={"status": "source.status", "update_time": "current_timestamp()"}) \
    .whenNotMatchedInsert(values={"id": "source.id", "status": "source.status", "create_time": "current_timestamp()"}) \
    .execute()
  

3. Incremental Append与Incremental Insert的区别 #

这两个术语常因语境混淆，核心区别如下：

• Incremental Append：无条件追加所有增量数据，无论目标表是否存在重复记录。本质是将增量数据直接追加到表尾，不做任何去重或条件判断，是最基础的增量写入方式。
• Incremental Insert（条件插入）：仅将增量数据中不存在于目标表的记录插入，避免重复。通常需先通过主键或唯一键筛选出增量中未在目标表出现的记录，再执行插入。示例实现：

  # 获取目标表已存在的主键集合
  existing_ids = spark.read.table("target_delta_table").select("id")
  # 筛选增量中未存在的记录
  new_records_df = incremental_df.join(existing_ids, on="id", how="left_anti")
  # 执行插入
  new_records_df.write.format("delta").mode("append").saveAsTable("target_delta_table")
  

  注：部分场景中“Incremental Insert”可能被当作Append的同义词，但严格来说，它特指带去重逻辑的增量插入操作。

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