CDP集群中Delta-Spark实现CDF类功能及合并去重问题咨询

1. CDP平台替代Databricks CDF的行级版本控制方案 #

由于Databricks CDF是其专属增强功能，开源Delta Lake及CDP原生环境不支持，可选择以下方案实现行级版本控制：

• Cloudera Hive ACID表：CDP原生支持Hive ACID v2表，支持行级INSERT/UPDATE/DELETE操作，自带版本历史记录。可通过SHOW HISTORY <table_name>查看版本迭代，用SELECT * FROM <table_name> VERSION AS OF <version_id>或时间戳回溯历史数据，完全适配CDP生态。
• Apache Iceberg表格式：CDP已集成Apache Iceberg，其快照（Snapshot）机制天然支持版本控制。每次数据修改都会生成新快照，可通过SELECT * FROM <table_name> AS OF SNAPSHOT <snapshot_id>查询历史版本，还能通过对比前后快照提取变更数据，满足CDC场景的版本追溯需求。
• 自定义版本追踪逻辑：基于现有Parquet/Delta表，新增version、update_timestamp、operation_type（INSERT/UPDATE/DELETE）字段。合并CDC数据时，既更新主表的当前记录，也将旧版本数据写入单独的历史表；或在主表保留所有版本记录，通过is_latest标记区分当前有效数据，实现版本追溯。

2. 解决delta-spark合并操作重复未变更记录的问题 #

针对delta-spark v0.0.6版本合并时产生的存储冗余问题，可通过以下方式优化：

• 精准过滤变更行后再合并：避免全量CDC数据直接合并，先对比CDC数据与基础表，仅保留真正有变更的记录（包括新增行和字段修改的行），再执行合并操作：

# 获取基础表当前数据
base_df = delta_table.toDF()
# 筛选CDC中新增或字段有变化的记录
changed_cdc_df = day2_df.alias("cdc") \
    .join(base_df.alias("base"), on="pk_1", how="left") \
    .filter(
        base_df["pk_1"].isNull() | 
        (day2_df["col1"] != base_df["col1"]) | 
        (day2_df["col2"] != base_df["col2"])  # 按需扩展所有需要对比的字段
    ) \
    .select("cdc.*")

# 仅合并变更记录
delta_table.alias('base').merge(changed_cdc_df.alias('update'), 'base.pk_1 = update.pk_1') \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

• 升级delta-spark版本：v0.0.6是极早期版本，缺乏合并操作的优化逻辑。升级至与Spark版本兼容的稳定版（如Spark 3.x搭配delta-spark 1.2.x或2.x），新版本Delta Lake会自动优化合并操作，仅生成变更行的新数据文件，不会重复存储未变更记录。
• 避免无差别更新：不要使用whenMatchedUpdateAll()无条件更新，而是添加字段变更判断条件，仅当字段确实不同时才执行更新：

delta_table.alias('base').merge(day2_df.alias('update'), 'base.pk_1 = update.pk_1') \
    .whenMatched(
        "base.col1 != update.col1 OR base.col2 != update.col2"  # 按需扩展字段
    ).updateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

补充说明：Databricks CDF的_change_data文件夹及table_changes函数属于Databricks私有扩展，开源Delta Lake及CDP集成版本均不支持，因此无法通过配置启用，需依赖上述替代方案实现行级版本控制。

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