深入了解 Flink 的网络协议栈 _网络协议栈

作者：Nico Kruber
出处：https://ververica.cn/developers/flink-network-protocol
创作不易，欢迎转载，但必须在文章开头保留此段声明，否则保留追究法律责任的权利。

Flink 的网络协议栈是组成 flink-runtime 模块的核心组件之一，是每个 Flink 作业的核心。它连接所有 TaskManager 的各个子任务(subtask)，因此，对于 Flink 作业的性能包括吞吐与延迟都至关重要。与 TaskManager 和 JobManager 之间通过基于 Akka 的 RPC 通信的控制通道不同，TaskManager 之间的网络协议栈依赖于更加底层的 Netty API 。
本文将首先介绍 Flink 暴露给流算子(Stream operator)的高层抽象，然后详细介绍 Flink 网络协议栈的物理实现和各种优化、优化的效果以及 Flink 在吞吐量和延迟之间的权衡。
1. 逻辑视图Flink 的网络协议栈为彼此通信的子任务提供以下逻辑视图，例如在 A 通过 keyBy() 操作进行数据 Shuffle ：

文章插图

这一过程建立在以下三种基本概念抽象的基础上：
▼ 子任务输出类型（ResultPartitionType）：

Pipelined（有限的或无限的）：一旦产生数据就可以持续向下游发送有限数据流或无限数据流。
Blocking：仅在生成完整结果后向下游发送数据。

▼ 调度策略：

同时调度所有任务(Eager)：同时部署作业的所有子任务（用于流作业）。
上游产生第一条记录部署下游(Lazy)：一旦任何生产者生成任何输出，就立即部署下游任务。
上游产生完整数据部署下游：当任何或所有生产者生成完整数据后，部署下游任务。

▼ 数据传输：

高吞吐：Flink 不是一个一个地发送每条记录，而是将若干记录缓冲到其网络缓冲区中并一次性发送它们。这降低了每条记录的发送成本因此提高了吞吐量。
低延迟：当网络缓冲区超过一定的时间未被填满时会触发超时发送，通过减小超时时间，可以通过牺牲一定的吞吐来获取更低的延迟。

我们将在下面深入Flink网络协议栈的物理实现时看到关于吞吐延迟的优化。对于这一部分，让我们详细说明输出类型与调度策略。首先，需要知道的是子任务的输出类型和调度策略是紧密关联的，只有两者的一些特定组合才是有效的。
Pipelined结果是流式输出，需要目标subtask正在运行以便接收数据。因此需要在上游task产生数据之前或者产生第一条数据的时候调度下游目标task运行。批处理作业生成有界结果数据，而流式处理作业产生无限结果数据。
批处理作业也可能以阻塞方式产生结果，具体取决于所使用的算子和连接模式。在这种情况下，必须等待上游task先生成完整的结果，然后才能调度下游的接收task运行。这能够提高批处理作业的效率并且占用更少的资源。
下表总结了 Task 输出类型以及调度策略的有效组合：
输出类型调度策略适用于…pipelined, unbounded一次性调度所有 task流作业pipelined, unbounded上游产生第一条输出调度下游N / A’pipelined,
bounded一次性调度所有 taskN /A²pipelined,
bounded上游产生第一条输出调度下游批作业blocking上游产生完整输出调度下游批作业
注释：
[1]目前 Flink 未使用
[2]批处理 / 流计算统一完成后，可能适用于流式作业
此外，对于具有多个输入的子任务，调度以两种方式启动：当所有或者任何上游任务产生第一条数据或者产生完整数据时调度任务运行。要调整批处理作业中的输出类型和调度策略，可以参考ExecutionConfig#setExecutionMode() ——尤其是ExecutionMode，以及ExecutionConfig#setDefaultInputDependencyConstraint() 。
2. 物理数据传输为了理解物理数据连接，请回想一下，在Flink中，不同的任务可以通过slot sharing group共享相同slot 。TaskManager还可以提供多个slot，以允许将同一任务的多个子任务调度到同一个TaskManager上。
对于下图所示的示例，我们假设 2 个并发为 4 的任务部署在 2 个 TaskManager 上，每个 TaskManager 有两个 Slot 。TaskManager 1 执行子任务 A.1，A.2，B.1 和 B.2，TaskManager 2 执行子任务 A.3，A.4，B.3 和 B.4 。在 A 和 B 之间是 Shuffle 连接类型，比如来自于 A 的 keyBy() 操作，在每个 TaskManager 上会有 2×4 个逻辑连接，其中一些是本地的，另一些是远程的：

文章插图

不同任务（远程）之间的每个网络连接将在 Flink 的网络堆栈中获得自己的 TCP 通道。但是，如果同一任务的不同子任务被调度到同一个 TaskManager，则它们与同一个 TaskManager 的网络连接将多路复用并共享同一个 TCP 信道以减少资源使用。在我们的例子中，这适用于 A.1→B.3，A.1→B.4，以及 A.2→B.3 和 A.2→B.4，如下图所示：