如何在Palantir Foundry中通过增量转换保留唯一键集合

要在Foundry的Python Transform里节省计算资源，核心思路是利用增量处理特性维护持久化的唯一键集合，每次只处理新增数据中未出现过的键，完全不用全量读取输入数据集做去重。具体实现方案如下：

核心逻辑 #

1. 开启增量Transform，仅处理输入数据集的新增批次
2. 读取已保存的历史唯一键集合
3. 对新增输入数据提取键，通过左反连接过滤掉已存在的键，再对新增数据内部去重
4. 将新增的唯一键追加到结果数据集，避免重复计算

完整代码示例 #

from transforms.api import transform, Input, Output, incremental, IncrementalInput, IncrementalMode

@incremental(inputs=[IncrementalInput("input_dataset", mode=IncrementalMode.APPEND)])
@transform(
    output=Output("/path/to/your/unique_keys_output"),
    input_dataset=Input("/path/to/your/source_dataset")
)
def incremental_unique_keys(ctx, input_dataset, output):
    # 读取历史唯一键（首次运行时无历史数据）
    previous_keys = output.dataframe() if output.exists() else None

    # 提取本次增量输入的键
    new_input_df = input_dataset.dataframe()
    new_keys = new_input_df.select("composite_key")

    # 筛选新增的唯一键
    if previous_keys is not None:
        # 左反连接：仅保留新增数据中未在历史集合出现过的键，再去重新增数据内部重复
        new_unique_keys = new_keys.join(previous_keys, on="composite_key", how="left_anti").dropDuplicates()
    else:
        # 首次运行直接对全量输入去重
        new_unique_keys = new_keys.dropDuplicates()

    # 仅当有新增键时才追加写入，避免空写入浪费资源
    if new_unique_keys.count() > 0:
        output.write_dataframe(new_unique_keys, mode="append")

关键细节说明 #

• 增量模式配置：通过@incremental装饰器指定输入为APPEND模式，确保每次只处理上次运行后新增的数据
• 左反连接效率：Spark的左反连接比先全量去重再对比的方式高效得多，因为它直接在分布式引擎层面做过滤，无需加载全量历史数据到内存
• 避免空写入：判断new_unique_keys的行数，只有存在新增键时才写入，减少不必要的存储操作

优化建议 #

• 若composite_key由多个字段组成，需将所有字段加入select和join的on参数中
• 对高基数的键可做哈希处理（如sha2(concat_ws('|', col1, col2), 256)），减少存储和连接的开销（注意处理哈希冲突风险）
• 给输出数据集按键的前缀分区，提升后续读取和连接的性能

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