具有自定义源的Flink工作流并行性

Question

我在Flink中构建了一个工作流，它由一个自定义源、一系列映射/平台映射和一个接收器组成。

自定义源的run()方法迭代存储在文件夹中的文件，并通过上下文的collects ()方法收集每个文件的名称和内容(我有一个自定义对象，该对象将此信息存储在两个字段中)。

然后，我有一系列映射/平面图来转换这些对象，然后使用自定义接收器将这些对象打印成文件。Flink的Web中生成的执行图如下：

​

​

我有一个集群或2个工作人员设置为每个有6个槽(他们都有6个核心，也)。我将并行设置为12，从执行图中我看到源的并行性为1，而工作流的其余部分则为12。

当我运行工作流(在专用文件夹中有大约15K文件)时，我使用htop监视员工的资源。所有的核心达到100%的利用率在大部分时间，但大约每30分钟左右，8-10的核心成为闲置约2-3分钟。

我的问题如下：

1. 我知道源代码运行时具有并行性1，我认为在从本地存储读取时这是正常的(我的文件位于每个工作人员的同一个目录中，因为我不知道将选择哪个工作人员来执行源代码)。这真的很正常吗？你能解释一下为什么会这样吗？

1. 执行我的工作流程的其余部分具有并行性12，这看起来是正确的，因为通过检查任务管理器的日志，我从所有的插槽(例如，.... [Flat Map -> Map -> Map -> Sink: Unnamed (**3/12**)] INFO ....、.... [Flat Map -> Map -> Map -> Sink: Unnamed (**5/12**)] INFO ....等)中得到了打印。但是，我不明白的是，如果一个插槽正在执行源角色，而我在集群中有12个插槽，那么其余的工作流是如何由12个时隙执行的？是否有一个时隙同时代表工作流其余部分的源和一个实例？如果是，这一特定时隙的资源是如何分配的？是否有人可以解释在这个工作流中正在执行的步骤？例如(这可能是错误的)：

• 插槽1读取文件并将它们转发给可用的插槽(2至12)
• 插槽1将一个文件转发给自己并停止读取，直到完成任务时
• ，插槽1读取更多的文件并将其转发到可用的

插槽。

我相信我上面描述的是错误的，但我举了一个例子来更好地解释我的问题。

1. 为什么我每30分钟就有一次空闲状态，每30分钟(或多或少)持续3分钟？

Answer 1

1. --单一使用者设置--将管道的总吞吐量限制在唯一一个使用者的性能上。此外，它还向所有插槽引入了沉重的洗牌--在本例中，由使用者读取的所有数据也会在这个使用者插槽上序列化，这是额外的CPU负载。相反，拥有与映射/平面映射并行相同的使用者并行性将允许链接源->映射操作并避免洗牌。默认情况下，
2. 允许子任务共享插槽，即使它们是不同任务的子任务，只要它们来自同一任务。其结果是，一个槽可以容纳整个工作流水线。因此，在您的情况下，插槽1有使用者和map /平面映射任务，而其他插槽只有map /平面映射任务。有关更多细节，请参见这里：https://ci.apache.org/projects/flink/flink-docs-release-1.10/concepts/runtime.html#task-slots-and-resources。另外，您还可以查看Web上每个子任务的实例。
3. 是否启用了检查点？如果是，如果是30分钟，那么这很可能是状态被快照时的间隔。

Answer 2

为了回答有关并行化您的读取的具体问题，我将执行以下操作.

RichSourceFunction.

• In

• 通过扩展open()方法实现自定义源，调用getRuntimeContext().getNumberOfParallelSubtasks()获取总并行性，调用getRuntimeContext().getIndexOfThisSubtask()获取子任务的索引initialized.

• In your run()方法，当您在文件上迭代时，获取每个文件名的hashCode()，模块化总的并行性。如果这等于子任务的索引，则处理它.

通过这种方式，您可以将工作分散到12个子任务上，而不需要子任务尝试处理相同的文件。