本文共 1970 字，大约阅读时间需要 6 分钟。
Flink 是一个流处理框架，拥有强大的状态管理功能，其中 CheckPoint 和 SavePoint 是两个核心机制。它们能够有效地帮助用户管理流处理程序的状态，实现状态点的保存与恢复，从而提升系统的可用性和容错能力。本文将详细介绍这两个机制的工作原理及其应用方法。
在实际使用 Flink 前，需要先启动 Flink 应用程序。以下是一个典型的启动命令示例：
./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024
其中，参数 -c 用于指定主类路径，-port 用于指定端口号。完整命令如下：
./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 -c streaming.SoetWindowWordCountJavaCheckPoint -port 9010
此外，若需要运行带依赖的 JAR 文件，请指定路径：
/usr/local/install/testJar/FlinkExample-1.0-SNAPSHOT-jar-with-dependencies.jar
默认情况下，Flink 会保留最近成功生成的一个 CheckPoint。如果程序失败，Flink 会自动从最近的 CheckPoint 恢复。但有时，我们可能需要保留多个 CheckPoint，以便在需要时选择特定的恢复点。要实现这一功能，需在配置文件中设置如下参数：
state.checkpoints.dir: hdfs://namenode:9000/flink/checkpoints
这样，CheckPoint 会被存储在指定的 HDFS 目录下。查看存储的 CheckPoint 可执行以下命令：
hdfs dfs -ls hdfs://namenode:9000/flink/checkpoints
如果程序在运行过程中出现故障，我们可以通过指定特定的 CheckPoint 路径来恢复。恢复命令如下：
bin/flink run -s hdfs://namenode:9000/flink/checkpoints/467e17d2cc343e6c56255d222bae3421/chk-56/_metadata flink-job.jar
运行后，程序会从指定的 CheckPoint 继续执行。需要注意的是，程序在恢复后会继续生成新的 CheckPoint，确保数据处理不中断。
SavePoint 功能的核心在于实现程序的无中断升级。通过 SavePoint，可以保存数据源的偏移量、算子状态等信息，从而在程序升级或重新启动时，能够从指定的点继续处理。SavePoint 提供了两种触发方式：定期触发和手动触发。
在实际应用中，SavePoint 的使用场景包括：
要手动触发 SavePoint，可执行以下命令：
bin/flink savepoint jobId [targetDirectory] [-yid yarnAppId]
此外，当程序需要取消运行时，可以通过以下命令恢复到最近的 SavePoint：
bin/flink cancel -s [targetDirectory] jobId [-yid yarnAppId]
flink-conf.yaml 中添加如下配置：state.savepoints.dir: hdfs://namenode:9000/flink/savepoints
手动指定算子 ID：为了确保程序在升级后能够正确恢复，建议手动为每个算子指定唯一的 ID。
依赖管理：确保程序依赖的版本与 Flink 版本兼容，避免因依赖冲突导致程序故障。
通过合理配置和使用 CheckPoint 与 SavePoint，用户可以显著提升 Flink 程序的可用性和稳定性。在实际应用中，建议根据具体需求选择合适的恢复点，并定期备份重要数据。
转载地址：http://zmefk.baihongyu.com/