Flink CDC里Oracle cdc latest 同步模式 如何找丢失数据？

在使用 Flink CDC 进行 Oracle 数据同步时，如果采用 latest 同步模式，并且在通过 flink stop 停止作业后从 Savepoint 恢复时发现丢失了一小段数据，可能是由于以下原因导致的。以下是问题分析及解决方案：

1. 问题分析

• latest 同步模式的特点
  在 latest 模式下，Flink CDC 会从 Oracle 的最新 Binlog 位置开始读取数据，而不会回溯历史数据。这意味着，如果作业停止后没有正确记录 Binlog 的位置，恢复时可能会跳过部分未处理的数据。

• Savepoint 的局限性
  Savepoint 是 Flink 提供的一种状态快照机制，主要用于保存作业的状态信息（如算子状态、Kafka 偏移量等）。然而，对于 CDC 场景，Savepoint 并不直接记录外部系统的 Binlog 位置。如果在恢复时未能正确对齐 Binlog 位置，可能会导致数据丢失。

• 可能的原因

  • 停止作业时未正确触发 Binlog 位置的记录。
  • 恢复时未从正确的 Binlog 位置开始读取。
  • Oracle 的 Binlog 被清理或覆盖，导致无法重新读取丢失的数据。

2. 解决方案

步骤 1：确认 Savepoint 和 Binlog 状态

• 检查 Savepoint 的生成方式
  确认 Savepoint 是否是通过 STOP_WITH_SAVEPOINT 方式生成的。如果是用户手动生成的 Savepoint（USER_REQUEST），可能存在状态不一致的风险。

• 检查 Oracle Binlog 的保留时间
  确保 Oracle 的 Binlog 在作业停止期间未被清理或覆盖。如果 Binlog 已被清理，则无法通过 Savepoint 恢复丢失的数据。

步骤 2：调整同步模式

• 切换到 initial 或 all 模式
  如果允许重新同步历史数据，可以将同步模式从 latest 切换为 initial 或 all，以确保从 Oracle 的最早 Binlog 位置开始读取数据。
  注意：此操作可能会导致重复数据的产生，需要下游系统具备去重能力。

步骤 3：结合 Tag 功能进行恢复

• 启用 Paimon 的 Tag 功能
  如果使用了 Paimon 作为存储层，可以通过 Tag 功能长期保留某个快照的状态。具体操作如下：
  1. 在表参数中开启自动创建 Tag 的功能：

     ALTER TABLE your_table SET ('sink.savepoint.auto-tag' = 'true');
     

  2. 找到与 Savepoint 对应的 Tag 名称，格式为 savepoint-${savepointID}。
  3. 使用以下命令将数据回退到指定的 Tag：

     CALL `<catalog-name>`.sys.rollback_to('<database-name>.<table-name>', 'savepoint-${savepointID}');
     

  4. 从 Savepoint 恢复作业。

步骤 4：手动对齐 Binlog 位置

• 指定 Binlog 启动位置
  在无状态启动时，可以通过配置 YAML 文件中的 scan.startup.timestamp-millis 参数，指定从某个时间戳开始读取 Binlog 数据。例如：

  source:
  type: oracle-cdc
  name: Oracle Source
  hostname: localhost
  port: 1521
  username: username
  password: password
  tables: schema.table
  scan.startup.timestamp-millis: 1672531200000  # 指定时间戳
  

  注意：需要确保指定的时间戳早于数据丢失的时间点。

步骤 5：验证数据完整性

• 检查下游数据
  在恢复作业后，对比上下游数据，确保数据完整性和一致性。如果发现仍有数据丢失，可以尝试手动补录缺失的数据。

3. 预防措施

• 定期创建 Savepoint 和 Tag
  定期触发 Savepoint 并结合 Paimon 的 Tag 功能，确保在作业异常停止时能够快速恢复到正确的状态。

• 延长 Oracle Binlog 的保留时间
  根据业务需求，适当延长 Oracle 的 Binlog 保留时间，避免因 Binlog 清理导致数据无法恢复。

• 监控作业状态
  使用 Flink 的监控工具，实时跟踪作业的运行状态和 Binlog 读取进度，及时发现潜在问题。

4. 总结

通过上述步骤，您可以有效解决 Flink CDC 在 latest 模式下从 Savepoint 恢复时丢失数据的问题。建议结合 Paimon 的 Tag 功能和 Oracle 的 Binlog 管理策略，确保数据的完整性和一致性。