Flink 原理 | Flink Sort-Shuffle 实现简介

 GitHub 地址 

目前，有两种批数据 shuffle 模型被现有的大规模分布式计算系统采用，分别是基于 hash 的方式以及基于 sort 的方式：

1. 基于 hash 方式的核心思路是将发送给下游不同并发消费任务的数据写到单独的文件中，这样文件本身就成了一个自然的区分不同数据分区的边界；
   
   
2. 基于 sort 方式的核心思路是先将所有分区的数据写在一起，然后通过 sort 来区分不同数据分区的边界。

我们之所以要在 Flink 中引入 sort-shuffle 的实现，一个重要的原因是 Flink 原本的基于 hash 的实现对大规模批作业不可用。这个也是被现有的其他大规模分布式计算系统所证明的：

1. 稳定性方面：对于高并发批作业，基于 hash 的实现会产生大量的文件，并且会对这些文件进行并发读写，这会消耗很多资源并对文件系统会产生较大的压力。文件系统需要维护大量的文件元数据，会产生文件句柄以及 inode 耗尽等不稳定风险。
   
   
2. 性能方面：对于高并发批作业，并发读写大量的文件意味着大量的随机 IO，并且每次 IO 实际读写的数据量可能是非常少的，这对于 IO 性能是一个巨大的挑战，在机械硬盘上，这使得数据 shuffle 很容易成为批处理作业的性能瓶颈。

1. 设计目标

在 Flink sort-shuffle 的整个实现过程中，我们把下面这些点作为主要的设计目标加以考量：

■ 1.1 减少文件数量

正如上面所讨论的，基于 hash 的实现会产生大量的文件，而减少文件的数量有利于提高稳定性和性能。Sort-Spill-Merge 的方式被分布式计算系统广泛采纳以达到这一目标，首先将数据写入内存缓冲区，当内存缓冲区填满后对数据进行排序，排序后的数据被写出到一个文件中，这样总的文件数量是：（总数据量 / 内存缓冲区大小），从而文件数量被减少。当所有数据写出完成后，将产生的文件合并成一个文件，从而进一步减少文件数量并增大每个数据分区的大小（有利于顺序读取）。

■ 1.2 打开更少的文件

■ 1.3 最大化顺序读写

■ 1.4 减少读写 IO 放大

■ 1.5 减少内存缓冲区消耗

2. 实现细节

■ 2.1 内存数据排序

public interface SortBuffer {
   */** Appends data of the specified channel to this SortBuffer. \*/*   boolean append(ByteBuffer source, int targetChannel, Buffer.DataType dataType) throws IOException;
   */** Copies data in this SortBuffer to the target MemorySegment. \*/*   BufferWithChannel copyIntoSegment(MemorySegment target);
   long numRecords();
   long numBytes();
   boolean hasRemaining();
   void finish();
   boolean isFinished();
   void release();
   boolean isReleased(); }

■ 2.2 文件存储结构

■ 2.3 读取 IO 调度

*// let data_regions as the data region list indexed from 0 to n - 1* *// let data_readers as the concurrent downstream data readers queue indexed from 0 to m - 1* for (data_region in data_regions) {   data_reader = poll_reader_of_the_smallest_file_offset(data_readers);   if (data_reader == null)     break;   reading_buffers = request_reading_buffers();   if (reading_buffers.isEmpty())     break;   read_data(data_region, data_reader, reading_buffers); }

■ 2.4 数据广播优化

■ 2.5 数据压缩

1. 稳定性

2. 性能

在 Flink 中，sort-shuffle 默认是不开启的，想要开启需要调小这个参数的配置：taskmanager.network.sort-shuffle.min-parallelism。这个参数的含义是如果数据分区的个数（一个计算任务并发需要发送数据给几个下游计算节点）低于这个值，则走 hash-shuffle 的实现，如果高于这个值则启用 sort-shuffle。实际应用时，在机械硬盘上，可以配置为 1，即使用 sort-shuffle。

还有一些后续的优化工作，包括但不限于：

1. 网络连接复用，这可以提高网络的建立的性能与稳定性，相关 Jira 包括 FLINK-22643 以及 FLINK-15455；
   
   
2. 多磁盘负载均衡，这有利于解决负载不均的问题，相关 Jira 包括 FLINK-21790 以及 FLINK-21789；
   
   
3. 实现远程数据 shuffle 服务，这有利于进一步提升批数据 shuffle 的性能与稳定性；
   
   
4. 允许用户选择磁盘类型，这可以提高易用性，用户可以根据作业的优先级选择使用 HDD 或者 SSD。

英文原文链接：

https://flink.apache.org/2021/10/26/sort-shuffle-part1.html

https://flink.apache.org/2021/10/26/sort-shuffle-part2.html

   戳我，报名 FFA 2021 大会！