如何为每个任务管理器（或每个节点）运行一个源？答案

【问题标题】：How to run one source per task manager (or per node)?如何为每个任务管理器（或每个节点）运行一个源？
【发布时间】：2016-03-21 16:52:15
【问题描述】：

我已经实现了打开固定 UDP 端口并监听它的源代码。所以，我想每个任务管理器只运行一个源（在我的情况下，我每个节点运行一个任务管理器），因为过度会抛出 java.net.BindException: Address already in use 异常。

我在测试 Apache Flink 的 HA 时注意到了这个问题。当我关闭一个任务管理器时，Apache Flick 开始尝试在一个节点上运行具有相同端口的两个源。

那么，如何为每个任务管理器（或每个集群节点）运行一个源？

【问题讨论】：

您是否尝试过将并行度设置为您拥有的节点数？
如何动态实现（基于任务管理器或节点的数量）？
来自文档：“可以通过在 ExecutionEnvironment 上调用 setParallelism(int parallelism) 或将 -p 传递给 Flink 命令行前端来覆盖整个作业的默认并行度。它可以通过在运算符上调用 setParallelism(int parallelism) 来覆盖单个转换。有关并行性的更多信息，请参阅编程指南。"
当然，我知道。但是如何动态改变并行度（基于任务管理器或节点的数量）？

【解决方案1】：

目前无法动态强制在每个TaskManager 上运行一种任务。通过将槽数设置为1，可以避免多个源任务被调度到同一台机器上。但是，如果您丢失了一台机器并且没有备用的TaskManager，那么您将没有足够的插槽来重新启动作业。

或者，您可以编写源代码，使其更具弹性。例如，如果源无法绑定到指定端口，您可以简单地停止源。鉴于没有其他程序可以绑定到该端口，那么您就知道已经有另一个源任务正在使用该端口的数据。

【讨论】：

我在第一个解决方案中看到了一个问题：源可以在任何任务管理器/节点上启动，但我只能将数据发送到集群的一个节点（活动/主节点）。所以，第二个解决方案更有趣，但我有一个问题：1）你能解释一下“停止源”是什么意思吗？如何停止源？ 2）据我了解，AF 将尝试重新启动停止的源。对？因此，在这种情况下，AF 将周期性地重新启动流程。对吗？
只需离开run函数。这将完成任务。所以你可以尝试打开一个套接字，如果失败了，那么你就离开run方法。
更新：我注意到这种解决方案存在问题。它阻止创建检查点：“检查点触发任务来源：RSyslog (2/2) 目前未执行。中止检查点。”其中“RSyslog (2/2)”源处于 FINISHED 状态。
这是当前 Flink 实现的一个缺点。将来，我们将增强 CheckpointCoordinator 使其可以处理已关闭/已完成的操作员。