Flink定期保存数据，failover后从上次成功的保存点处恢复，并提供Exactly-Once的投递保障机制

  1. // 恢复保存点
  2. public boolean restoreSavepoint(
  3.     String savepointPointer,
  4.     boolean allowNonRestored,
  5.     Map<JobVertexID, ExecutionJobVertex> tasks,
  6.     ClassLoader userClassLoader) throws Exception {
  8.     Preconditions.checkNotNull(savepointPointer, "The savepoint path cannot be null.");
  10.     LOG.info("Starting job {} from savepoint {} ({})",
  11.              job, savepointPointer, (allowNonRestored ? "allowing non restored state" : ""));
  12.     // 从指定目录获取hdfs的地址
  13.     final CompletedCheckpointStorageLocation checkpointLocation = checkpointStorage.resolveCheckpoint(savepointPointer);
  15.     // 1、加载 metadata 信息
  16.     // 2、生成operator 到 task 的映射
  17.     // 3、检查 并行度
  18.     // 4、转换这次的 savepoint为checkpoint，以便失败后恢复
  19.     CompletedCheckpoint savepoint = Checkpoints.loadAndValidateCheckpoint(
  20.         job, tasks, checkpointLocation, userClassLoader, allowNonRestored);
  22.     // 将要恢复的checkpoint信息写到zk，并异步删除旧的checkpoint
  23.     completedCheckpointStore.addCheckpoint(savepoint);
  25.     // 重置checkpoint 计数器
  26.     long nextCheckpointId = savepoint.getCheckpointID() + 1;
  27.     checkpointIdCounter.setCount(nextCheckpointId);
  29.     LOG.info("Reset the checkpoint ID of job {} to {}.", job, nextCheckpointId);
  30.     // 从最近一次 Checkpoint 处恢复 State
  31.     // 获取OperatorState，分配state
  32.     return restoreLatestCheckpointedStateInternal(new HashSet<>(tasks.values()), true, true, allowNonRestored);
  33. }

总结

  • 首先客户端提供 Checkpoint 或 Savepoint 的目录
  • JM 从给定的目录中找到 _metadata 文件（Checkpoint 的元数据文件）
  • JM 解析元数据文件，做一些校验，将信息写入到 zk 中，然后准备从这一次 Checkpoint 中恢复任务
  • JM 拿到所有算子对应的 State，给各个 subtask 分配 StateHandle（状态文件句柄）
  • TM 启动时，也就是 StreamTask 的初始化阶段会创建 KeyedStateBackend 和 OperatorStateBackend
  • 创建过程中就会根据 JM 分配给自己的 StateHandle 从 dfs 上恢复 State