Flink 原理与实现：再谈反压

古城微笑少年丶 2021-12-14 05:49 369阅读 0赞

扫码关注公众号免费阅读全文：**冰山烈焰的黑板报**  
![在这里插入图片描述][20200906232656299.jpg_pic_center]

[Flink 原理与实现：如何处理反压问题][Flink] 这一篇文章中我们讲了 Flink 的反压机制。本文我将更加详细的介绍先后采用的两种反压机制：

*  基于 TCP 的反压（< 1.5）
 *  基于信用的反压（≥ 1.5）

# 1. 逻辑视图 #

Flink 网络栈是 flink-runtime 的核心组件，所有来自 TaskManager 的工作单元（子任务）都通过它来互相连接。你的流式传输数据流都要经过网络栈，所以它对 Flink 作业的性能表现（包括吞吐量和延迟指标）至关重要。与通过 Akka 使用 RPC 的 TaskManager 和 JobManager 之间的协调通道相比，TaskManager 之间的网络栈依赖的是更底层的，基于 Netty 的 API。下面是它的逻辑视图：  
![Flink 网络栈逻辑视图][Flink 1]  
它抽象了以下三种不同的配置：

*  Subtask output type (ResultPartitionType)：
    
     *  pipelined (bounded or unbounded)：一旦生成数据就一条一条向下游发送，要么作为有界流，要么作为无界流。
     *  blocking：只有当所有的数据都产出之后，才会向下游发送。
 *  Scheduling type：
    
     *  all at once (eager)：同时部署 Job 所有的 subtask（对于流式应用）。
     *  next stage on first output (lazy)：一旦生产者有输出数据的时候，就部署下游的 task。
     *  next stage on complete output：当任一或全部生产者生成了完整的数据集，就部署下游的 task。
 *  Transport：
    
     *  high throughput：Flink 采用缓存一批数据在 network buffer 中，并一起发送它们的方式，代替逐条发送数据的方式。以此降低单条数据的成本，提升吞吐量。
     *  low latency via buffer timeout：通过降低发送未攒满 buffer 的数据，牺牲一定的吞吐以换区更低的延时。

现在我们详细说一下 output 和 scheduling types。首先，我们需要知道 subtask 的 output 和 scheduling types 是密地交织在一起，只有这两种类型的特定组合才是有效的。

Pipelined result partitions 是流式输出，需要一个正在工作的 subtask 发送数据。下游的 task 可以在结果产出之前或第一次输出时调度。批处理作业生成有界的结果，而流作业生成无界的结果。

批作业也可以采用阻塞模式，取决于使用的 operator 和 connection 模式。在这种情况下，必须先生成完整的结果，然后才能调度接收任务。这样一来，批处理作业的效率会更高，需要的资源更少。

下表总结了所有有效组合：

<table> 
 <thead> 
  <tr> 
   <th>输出类型</th> 
   <th>调度类型</th> 
   <th>应用到…</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>pipelined, unbounded</td> 
   <td>pipelined, unbounded</td> 
   <td>Streaming jobs</td> 
  </tr> 
  <tr> 
   <td>pipelined, unbounded</td> 
   <td>next stage on first output</td> 
   <td>n/a （目前 Flink 尚未使用）</td> 
  </tr> 
  <tr> 
   <td>pipelined, bounded</td> 
   <td>all at once</td> 
   <td>n/a</td> 
  </tr> 
  <tr> 
   <td>pipelined, bounded</td> 
   <td>next stage on first output</td> 
   <td>Batch jobs</td> 
  </tr> 
  <tr> 
   <td>blocking</td> 
   <td>next stage on complete output</td> 
   <td>Batch jobs</td> 
  </tr> 
 </tbody> 
</table>

# 2. TCP 流量控制 #

在详细介绍基于 TCP 的反压之前，我们需要先了解一下 TCP 流量控制（Flow Control，即流控）。TCP 的首部如下：  
![TCP 首部][TCP]  
这里仅介绍四个比较重要的概念：

1.  **Sequence Number**：是包的序号，用来解决网络包乱序（reordering）问题。
2.  **Acknowledgement Number**：就是ACK——用于确认收到，用来解决不丢包的问题。
3.  **Window**：又叫 **Advertised-Window**，也就是著名的滑动窗口（Sliding Window），用于解决流控的。
4.  **TCP Flag** ：也就是包的类型，主要是用于操控TCP的状态机的

假设现在有个 Sender 其发送窗口初始大小为 3，Receiver 其接收窗口固定大小为 5。通过 TCP 流控可以看到它的消息发送过程如下图：  
![TCP 流控图][TCP 1]  
可以看到，当 User Consumer 的消费速度赶不上 Sender 的发送速度时，Receiver 的窗口被数据填满，此时会根据 TCP 首部的 Advertised-Window 通知 Sender 降低消息发送速度，从而达到流量控制的目的。

# 3. 基于 TCP 的反压（< 1.5） #

[Flink 原理与实现：如何处理反压问题][Flink] 这篇文章已有介绍，这里不做赘述，只补充一点。TaskManager 内的 subtask 造成反压，也会影响同 TaskManager 内的其他 subtask。如下图 subtask B.4 过载对数据链路造成反压，并阻止 subtask B.3 接收和处理数据，即使它仍然有可用容量。  
![基于 TCP 的反压][TCP 2]

# 4. 基于信用的反压（≥ 1.5） #

基于信用的反压（Credit-based Flow Control）可以确保正在传输的数据在接收端都有能力处理。这是在 Flink 现有机制上的自然扩展。现在每个远程 InputChannel 都有一组独占的缓存区（exclusive buffer），而不是使用共享的本地缓存区。相应地，本地缓存区被称为浮动缓存区（float buffer），因为它们是浮动的，并可用于每个 InputChannel 。

接收方将缓冲区的可用性，称为发送方的信用（1 buffer = 1 credit）。每个 ResultSubpartition 都会跟踪自己的通道信用值（channel credit）。如果有 credit 可用，那么buffer 仅转发到较低的网络栈，每发送一个 buffer，credit 就会减一。除了 buffer 之外，我们还会发送当前 backlog 的大小，从而指定在它 Subpartition 中队列中等待的 buffer 数量。接收端通过使用它，来请求适量的 float buffer，以便更快地处理 backlog。接收端会尝试获取和 backlog 的大小一样多的 float buffer，但有时事与愿违，它可能获取一些 float buffer，也可能一点都获取不到。该接收端将使用检索到的缓冲区，并侦听进一步可用的缓冲区。  
![基于信用的反压][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAzNzY3ODg_size_16_color_FFFFFF_t_70]  
Credit-based Flow Control 使用 taskmanager.network.memory.buffers-per-channel （强制的）指定独占缓存区的大小，使用 taskmanager.network.memory.floating-buffers-per-gate（可选的）指定本地缓存区的大小，通过这两个参数实现没有流控时相同缓存的限制。这两个参数的默认值可以使流控时理论上的最大吞吐量至少和非流控时的一样高。你可能需要根据网络带宽和实际的传输往返时间进行调节。

# 4. 总结 #

通过 Credit-based Flow Control，接收端和发送端之间缓存的数据更少了，我们可以更早的遇到反压。其实，缓存更多的数据也没太大的作用，如果你想继续使用流控，可以使用 taskmanager.network.memory.floating-buffers-per-gate 增加 float buffer 的数量。以下是 Credit-based Flow Control 的优缺点。

<table> 
 <thead> 
  <tr> 
   <th>优点</th> 
   <th>缺点</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>在多路复用连接中更好的利用数据倾斜的资源</td> 
   <td>额外的 credit-announce 信息</td> 
  </tr> 
  <tr> 
   <td>改进 checkpoint 对齐</td> 
   <td>额外的 backlog-announce 信息（附带在缓存消息中，几乎没有开销）</td> 
  </tr> 
  <tr> 
   <td>更少的内存使用（网络层的数据更少）</td> 
   <td>潜在的往返延迟</td> 
  </tr> 
 </tbody> 
</table>

[20200906232656299.jpg_pic_center]: /images/20211213/c3273da91d2f45c0849fa37de5fb63a4.png
[Flink]: https://blog.csdn.net/u010376788/article/details/92086752
[Flink 1]: /images/20211213/7bfaf5f67ab04f02b74aa98746cad5a1.png
[TCP]: /images/20211213/1587e346af334f14bf6df195d4edc3f5.png
[TCP 1]: /images/20211213/aebc1a8bf3414b8497b36696ebee26b1.png
[TCP 2]: /images/20211213/0bae8685221343aea259d2d7016a748b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTAzNzY3ODg_size_16_color_FFFFFF_t_70]: /images/20211213/897ec0b03ba148088d94c284e65744ac.png