代码之家 › 专栏 › 技术社区 › igx

如何在Flink中更新作业,而不通过UI失去状态以及一些后端状态问题

flink-streaming apache-flink

igx · 技术社区 · 6 年前

我有以下代码:

val stateUri = "file:///tmp/"

new RocksDBStateBackend(stateUri, true)

1 回复 | 直到 6 年前

Till Rohrmann 6 年前

为了升级您的作业,您首先应该通过 bin/flink savepoint <JOB_ID> <TARGET_DIRECTORY> . 或者,您也可以使用保存点取消作业,该保存点创建保存点并停止作业。 bin/flink cancel --withSavepoint <TARGET_DIRECTORY> <JOB_ID> . 两个CLI调用都将返回到创建的保存点的路径,该保存点应存储在 TARGET_DIRECTORY . 要从此保存点恢复,应将此路径输入 保存点路径 用户界面中的字段或通过提交作业 bin/flink run --fromSavepoint <SAVEPOINT_PATH> <JAR> .
不, stateUri 仅是状态后端存储检查点的基本路径。状态后端将创建一个子目录,其ID为作业,在该目录下存储检查点。因此,检查点的路径通常看起来像 stateUri/JOB_ID/chk-1 哪里 JOB_ID 是UUID(例如) 0ba86fd9d1b29d90796e4a7d27f9b2f9 )第一个检查站。
为了扩展一个作业,您应该获取一个保存点,取消该作业,然后重新提交从保存点恢复的作业,并增加并行性(例如 bin/flink run --fromSavepoint <SAVEPOINT_PATH> --parallelism 10 <JAR> )
每个作业都有一个唯一的作业ID。因此,您可以在下面找到两个子目录 斯塔内利 这是不同的作业ID。检查点将分别存储在每个作业的每个子目录中。

推荐文章

Ellen Spertus · 使用Scala访问HBase的推荐方法

6 年前

skrshn · JAVA尝试在Flink中创建Kafka使用者时出现lang.NoClassDefFoundError(CheckpointedRestoring)

6 年前

Daniel · 目录中的Apache Flink流文件

6 年前

Christos Hadjinikolis · Flink:执行管道java时出错。util。同时发生的TimeoutException:期货在[10000毫秒]后超时

6 年前

Mustafa · Apache Flink:自定义InputFormat仅以1的并行度运行

6 年前

Soheil Pourbafrani · Apache Flink-org的类文件。阿帕奇。弗林克。流式处理。api。斯卡拉。找不到数据流

7 年前

Jicaar · 一个卡夫卡制作人可以被多个接收器安全使用吗?

7 年前

Christos Hadjinikolis · Flink JobExecutionException:akka。客户超时

7 年前

Christos Hadjinikolis · Flink IOException:网络缓冲区数量不足

7 年前

Brutal_JL · flink ReduceFunction的默认触发器是什么?

7 年前