实时计算 Flink版产品使用问题之从检查点重启任务，怎么在YAML配置文件中添加检查点的路径-阿里云开发者社区

实时计算 Flink版产品使用问题之从检查点重启任务，怎么在YAML配置文件中添加检查点的路径

2024-07-16 186

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

实时计算 Flink 版，1000CU*H 3个月

简介： 实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。

问题一：使用flink native k8s的nodeport模式，能指定暴露的port吗？

使用flink native k8s的nodeport模式，能指定暴露的port吗？

参考答案：

在Flink中，通过使用NodePort模式暴露服务，可以指定要暴露的端口。NodePort模式允许你将Flink服务从集群内部访问扩展到集群外部，通过为每个节点上的特定端口映射到一个公共访问端口来实现。

在Flink的NodePort模式下，你可以在Flink的配置文件（例如flink-conf.yaml）中指定要暴露的端口范围。具体来说，你可以设置taskmanager.network.port属性来指定TaskManager节点上用于与Flink JobManager通信的端口。同样，你还可以设置jobmanager.network.port属性来指定JobManager节点的端口。

一旦你配置了端口范围，Flink会为每个节点选择一个端口并在该端口上启动服务。然后，Flink会将每个节点的端口映射到一个公共访问的NodePort上，以便从集群外部访问Flink服务。

请注意，为了确保集群外部能够访问Flink服务，你还需要正确配置网络策略和防火墙规则，以便允许外部流量通过NodePort访问Flink服务。

使用Flink的NodePort模式时，你可以指定要暴露的端口范围，并使用该模式将Flink服务暴露给集群外部的客户端访问。

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/592487

问题二：Flink 2023 的PPT能下载吗？

Flink 2023 的PPT能下载吗？网页看太小了放大又很模糊。

参考答案：

可以登录网页版观看：https://flink-forward.org.cn/

目前只有主会场演讲直播回放和分论坛演讲回放，PPT暂时还没有放出来2023版本，不过往年的可以通过历届活动点击PPT获取

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/592484

问题三：用这两个flink run的参数提交任务到集群上，为什么在管理页面显示container数量不一样？

我用这两个flink run的参数提交同一个任务到yarn集群上，在yarn的管理页面显示启动的container数量不一样。为什么？实际应该用那个呢？bin/flink run -t yarn-per-job -ys 2 -yjm 1G -ytm 4G -p 5 -sae

bin/flink run -t yarn-per-job -p 5 -sae -Djobmanager.memory.process.size=1024mb -Dtaskmanager.memory.process.size=4096mb -Dtaskmanager.numberOfTaskSlots=2

参考答案：

楼主你好，在使用Flink提交任务到YARN集群上时，容器数量的不一致可能是参数配置不一致造成的，你提供的两个flink run命令中，有一些参数是不同的，比如-ys和-yjm参数，这些参数会影响任务运行时的资源分配和容器数量，所以请确保两次提交任务时的参数配置一致，才能得到一致的容器数量。

还有就是任务图的拓扑结构，Flink任务图的拓扑结构可能会导致不同的容器需求，如果任务的操作符数量或节点之间的依赖关系不同，那么为了满足任务的需求，启动的容器数量可能会不一致。

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/592482

问题四：Flink CDC里我用flink sql从kafka订阅canaljson插入数据，这是为什么？

Flink CDC里我用flink sql从kafka订阅canaljson插入数据，如果这个数据插入后，重启任务，然后发送canaljson无法删除，不重启的话就可以，为什么？

参考答案：

Flink CDC 从 Kafka 订阅 Canal JSON 格式的数据并将其插入到目标表中，如果在数据插入后重启 Flink 任务，发现无法正确处理 Canal JSON 中的 DELETE 操作，而不停止任务则可以正常处理，这个问题可能与 Flink SQL 的状态管理和 Checkpoint 机制有关。

Flink CDC 在处理 CDC 数据时，会依赖其状态来追踪和处理数据库的变更事件（包括 INSERT、UPDATE 和 DELETE）。当你重启任务时，如果没有正确设置 Checkpoint 或 Savepoint，Flink 任务重启后会从 Kafka 最新的偏移量开始消费，而忽略了之前已经消费并处理过的 DELETE 事件，因此会出现 DELETE 操作无法执行的情况。

要解决这个问题，可以采取以下措施：

启用 Checkpoint：确保你的 Flink 任务启用了 Checkpoint，并且设置合理的 Checkpoint 间隔。这样在任务重启时，可以从最近的 Checkpoint 恢复状态，继续处理 Kafka 中未消费完的数据。

CREATE TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  'enable.startup.mode' = 'latest-offset',  -- 或者设置为 'group-offsets' 并确保消费组一致
  'properties.checkpoint.interval' = '60000',  -- 根据实际情况设置Checkpoint间隔
  ...
);