为什么flinkcdc读取mysql数据作为广播流然后使用手动savepoint方式进行？

在使用 Flink CDC 读取 MySQL 数据并将其作为广播流时，如果通过手动 Savepoint 方式进行 Checkpoint，并在作业重启后发现广播流的数据丢失，这通常与广播流的特性以及 Savepoint 的工作机制有关。以下是详细的原因分析和解决方案：

原因分析

1. 广播流的状态管理机制

   • 广播流（Broadcast Stream）是 Flink 中一种特殊的流类型，其数据会被广播到下游的所有并行子任务中。广播流的状态存储在 BroadcastState 中。
   • BroadcastState 是一种非持久化的状态，默认情况下不会被 Savepoint 或 Checkpoint 持久化保存。因此，当作业通过 Savepoint 重启时，广播流中的数据不会自动恢复。

2. Savepoint 的作用范围

   • Savepoint 是用户手动触发的一种快照机制，主要用于作业升级或配置变更。它会保存作业的状态（如 Keyed State 和 Operator State），但不包括 BroadcastState。
   • 因此，即使您通过 Savepoint 重启作业，广播流中的数据也不会从 Savepoint 中恢复。

3. 广播流数据的来源

   • 如果广播流的数据来源于 MySQL CDC，那么这些数据在全量阶段和增量阶段的行为可能不同：
     • 全量阶段：MySQL CDC 通过 JDBC 查询全量数据并广播。
     • 增量阶段：MySQL CDC 通过 Binlog 读取增量数据并广播。
   • 在作业重启后，广播流需要重新从源头（MySQL CDC）读取数据，但由于广播流的状态未被持久化，之前的数据无法恢复。

解决方案

为了确保广播流的数据在作业重启后仍然可用，可以采取以下措施：

1. 使用外部存储保存广播流数据

• 将广播流的数据写入外部存储系统（如 Hologres、Kafka 或其他持久化存储），并在作业重启时从外部存储重新加载数据。
• 示例流程：
  1. 在广播流的处理逻辑中，将数据写入外部存储。
  2. 在作业启动时，从外部存储读取数据并重新初始化广播流。
• 这种方式可以确保广播流的数据在作业重启后仍然完整。

2. 自定义 BroadcastState 的持久化逻辑

• 如果必须依赖 BroadcastState，可以通过自定义逻辑实现其持久化：
  1. 在 Checkpoint 或 Savepoint 触发时，将 BroadcastState 的内容写入外部存储。
  2. 在作业重启时，从外部存储读取数据并重新填充 BroadcastState。

• 示例代码：

  // 自定义 BroadcastState 的持久化逻辑
  public class CustomBroadcastFunction extends BroadcastProcessFunction<InputType, BroadcastType, OutputType> {
      private transient MapStateDescriptor<String, BroadcastType> broadcastStateDescriptor;
  
      @Override
      public void open(Configuration parameters) {
          broadcastStateDescriptor = new MapStateDescriptor<>("broadcastState", Types.STRING, Types.POJO(BroadcastType.class));
      }
  
      @Override
      public void processBroadcastElement(BroadcastType value, Context ctx, Collector<OutputType> out) throws Exception {
          // 将广播数据写入 BroadcastState
          ctx.getBroadcastState(broadcastStateDescriptor).put(value.getKey(), value);
      }
  
      @Override
      public void snapshotState(FunctionSnapshotContext context) throws Exception {
          // 在 Checkpoint/Savepoint 时将 BroadcastState 写入外部存储
          Map<String, BroadcastType> stateMap = new HashMap<>();
          for (Map.Entry<String, BroadcastType> entry : getRuntimeContext().getBroadcastState(broadcastStateDescriptor).entries()) {
              stateMap.put(entry.getKey(), entry.getValue());
          }
          // 将 stateMap 写入外部存储（如 HDFS、Kafka 等）
      }
  
      @Override
      public void initializeState(FunctionInitializationContext context) throws Exception {
          // 在作业启动时从外部存储加载 BroadcastState
          Map<String, BroadcastType> stateMap = loadFromExternalStorage();
          for (Map.Entry<String, BroadcastType> entry : stateMap.entrySet()) {
              getRuntimeContext().getBroadcastState(broadcastStateDescriptor).put(entry.getKey(), entry.getValue());
          }
      }
  }
  

3. 避免依赖广播流的全量数据

• 如果广播流的数据仅用于增量处理，可以考虑在作业重启后直接从增量阶段开始消费，而无需恢复全量数据。
• 通过配置 scan.startup.mode 参数，控制 MySQL CDC 的启动模式。例如：

  'scan.startup.mode' = 'latest-offset'
  

  这样可以避免全量数据的重复加载。

重要提醒

• 广播流的状态默认不会被 Savepoint 持久化，这是 Flink 的设计限制。如果您的业务逻辑强依赖广播流的历史数据，请务必通过外部存储或自定义逻辑实现数据的持久化。
• 合理设置 Checkpoint 和 Savepoint 的间隔时间，以减少数据丢失的风险。同时，确保外部存储系统的高可用性和一致性。

通过上述方法，您可以有效解决 Flink CDC 读取 MySQL 数据作为广播流时，因 Savepoint 导致数据丢失的问题。