数据管道中的幂等性

数据工程的可靠性在很大程度上依赖于从故障中恢复的能力。当一个分布式数据摄取作业因网络分区或竞价实例抢占而失败时，标准的操作响应是重启管道。如果仅仅重新运行一个作业导致重复记录或数据状态损坏，那么该管道是脆弱的，并且需要对每个错误进行人工干预。

为构建具有弹性的系统，工程师们追求幂等性。在数据管道的背景下，如果一个操作多次执行产生与执行一次相同的结果，那么该操作是幂等的。这个特性保证了如果一个作业在处理了90%的数据后崩溃，随后的重试不会重复那90%的数据，而是会确保最终状态正确反映源数据。

从数学角度定义幂等性

形式上，如果函数 $f$ 满足以下条件，则它是幂等的：

$f(f(x)) = f(x)$

在数据工程中，设 $S_0$ 是您的数据湖的初始状态，$f$ 是您的摄取作业。运行该作业会将状态转换为 $S_1$。如果作业第二次运行（可能是由 Airflow 或 Dagster 等编排工具自动运行），状态应保持 $S_1$，而不是改变为包含重复行的新状态 $S_2$。

$最终状态 = 摄取(摄取(初始状态)) = 摄取(初始状态)$

实现这种行为需要特定的设计模式，因为许多分布式文件写入器的默认行为仅仅是将新文件附加到目标目录。

仅追加的风险

考虑一个简单的批处理作业，它从 API 读取每日日志并将其写入对象存储。如果作业逻辑严格遵循“读取并追加”，那么故障将产生严重的一致性问题。

1. 作业开始读取100个文件。
2. 它处理并将50个文件写入目标存储桶。
3. 作业崩溃。
4. 调度器重试该作业。
5. 作业读取所有100个文件并将100个文件写入目标。

数据湖现在包含150个文件：50个来自失败运行，100个来自成功重试。这50个重叠文件产生重复数据，下游查询必须将其过滤掉，这会浪费计算资源并可能导致分析结果偏差。

策略1：插入覆盖

对于批处理分区数据，实现幂等性的最常见模式是插入覆盖。这种方法完全替换数据的特定分区，而不是追加到它们。

当管道为一个特定的逻辑日期（例如 2023-10-27）运行时，它不应该仅仅写入数据；它应该声明自己是该日期的权威数据源。逻辑遵循以下步骤：

1. 处理目标分区中的输入数据。
2. 将新数据写入临时位置或暂存区。
3. 原子地将现有分区指针交换为新数据。

如果作业运行两次，第二次运行只会覆盖第一次运行生成的分区（或来自失败运行的部分数据）。最终结果始终是该分区的一组单一且完整的数据。

Apache Iceberg 和 Delta Lake 等表格式在元数据层面处理这种原子性。它们写入新的数据文件，然后提交一个事务日志条目，其中指出：“分区 X 的有效文件现在是这些新文件；忽略旧文件。”此操作在对象存储上是安全的，因为它避免了列出和删除物理文件的缓慢和最终一致性特点。

故障恢复期间，仅追加策略与幂等覆盖策略之间的数据一致性比较。

策略2：合并（更新插入）

对于流式管道或无法清晰映射到基于时间的分区的数据集（例如缓慢变化维度表），插入覆盖通常过于繁重。你不能只为了更新一个地址而覆盖整个客户表。

在这些情况下，通过合并操作（也称为更新插入）来实现幂等性。这要求每条记录都有一个唯一的主键。逻辑会检查目标中是否已存在具有传入键的记录：

• 匹配： 更新现有记录。
• 不匹配： 插入新记录。

如果管道处理同一批消息两次，MERGE 操作会检测到键已存在。它会用相同的值更新它们，导致数据状态没有净变化（假设在此期间源数据没有改变）。

SQL 语法通常如下所示：

MERGE INTO target_table t
USING source_updates s
ON t.id = s.id
WHEN MATCHED THEN
  UPDATE SET t.status = s.status, t.updated_at = s.updated_at
WHEN NOT MATCHED THEN
  INSERT (id, status, updated_at) VALUES (s.id, s.status, s.updated_at);

数据管道中的确定性

幂等性很大程度上依赖于确定性。如果相同的输入总是产生相同的输出，那么管道就是确定性的。一个破坏幂等性的常见错误是在转换逻辑中依赖系统时间或随机值。

考虑一个添加 ingestion_timestamp 列的管道。

• 不良实践： 在转换中使用 CURRENT_TIMESTAMP() 或 NOW()。每次重新运行作业时，此时间戳都会改变。如果您将此时间戳用于下游增量处理（水印），重试会创建数据的一个新“版本”，这可能会使下游消费者感到困惑。
• 最佳实践： 将逻辑执行日期（数据所指的日期）从编排工具传递给作业。如果您正在处理 2023-11-01 的数据，附加到数据的时间戳应反映该逻辑日期或源事件时间，而不是运行作业的服务器的实际时间。

检查点（Checkpointing）的作用

在流处理系统中（如 Spark Structured Streaming 或 Flink），幂等性通过检查点和预写日志来维护。引擎记录上次成功处理记录的偏移量。

当流崩溃并重启时：

1. 它读取检查点以找到最后提交的偏移量（例如，偏移量4500）。
2. 它从源（如Kafka）请求数据，从偏移量4501开始。

然而，存在一个不易发现的特殊情况，称为“输出提交问题”。如果引擎处理了记录 4501-4600 并将其写入存储，但在更新检查点之前崩溃了，系统会认为它停止在4500。它将重新处理 4501-4600。

为了防止此处出现重复，目标端（存储层）必须智能地处理这些重写操作。现代对象存储连接器运用表格式（Delta、Iceberg）的事务机制，以确保即使数据被重写，事务日志也能协调版本，或者文件命名约定可以防止冲突。

模式总结

摄取类型	幂等模式	技术要求
每日批处理	分区覆盖	数据必须按时间/日期分区。
CDC / 可变数据	合并（更新插入）	记录必须有一个可靠的主键。
流式（追加）	仅一次语义	引擎支持检查点；目标端支持事务性提交。

通过认真应用这些模式，您可以将管道的稳定性与基础设施的稳定性解耦。一个可以随时安全重新运行的管道能够减轻数据团队的运维负担，并确保对从数据湖中获得的分析结果的信任。