Flink 原理 | Flink Sort-Shuffle 實(shí)現(xiàn)簡(jiǎn)介

 GitHub 地址 

目前，有兩種批數(shù)據(jù) shuffle 模型被現(xiàn)有的大規(guī)模分布式計(jì)算系統(tǒng)采用，分別是基于 hash 的方式以及基于 sort 的方式：

1. 基于 hash 方式的核心思路是將發(fā)送給下游不同并發(fā)消費(fèi)任務(wù)的數(shù)據(jù)寫(xiě)到單獨(dú)的文件中，這樣文件本身就成了一個(gè)自然的區(qū)分不同數(shù)據(jù)分區(qū)的邊界；
   
   
2. 基于 sort 方式的核心思路是先將所有分區(qū)的數(shù)據(jù)寫(xiě)在一起，然后通過(guò) sort 來(lái)區(qū)分不同數(shù)據(jù)分區(qū)的邊界。

我們之所以要在 Flink 中引入 sort-shuffle 的實(shí)現(xiàn)，一個(gè)重要的原因是 Flink 原本的基于 hash 的實(shí)現(xiàn)對(duì)大規(guī)模批作業(yè)不可用。這個(gè)也是被現(xiàn)有的其他大規(guī)模分布式計(jì)算系統(tǒng)所證明的：

1. 穩(wěn)定性方面：對(duì)于高并發(fā)批作業(yè)，基于 hash 的實(shí)現(xiàn)會(huì)產(chǎn)生大量的文件，并且會(huì)對(duì)這些文件進(jìn)行并發(fā)讀寫(xiě)，這會(huì)消耗很多資源并對(duì)文件系統(tǒng)會(huì)產(chǎn)生較大的壓力。文件系統(tǒng)需要維護(hù)大量的文件元數(shù)據(jù)，會(huì)產(chǎn)生文件句柄以及 inode 耗盡等不穩(wěn)定風(fēng)險(xiǎn)。
   
   
2. 性能方面：對(duì)于高并發(fā)批作業(yè)，并發(fā)讀寫(xiě)大量的文件意味著大量的隨機(jī) IO，并且每次 IO 實(shí)際讀寫(xiě)的數(shù)據(jù)量可能是非常少的，這對(duì)于 IO 性能是一個(gè)巨大的挑戰(zhàn)，在機(jī)械硬盤(pán)上，這使得數(shù)據(jù) shuffle 很容易成為批處理作業(yè)的性能瓶頸。

1. 設(shè)計(jì)目標(biāo)

在 Flink sort-shuffle 的整個(gè)實(shí)現(xiàn)過(guò)程中，我們把下面這些點(diǎn)作為主要的設(shè)計(jì)目標(biāo)加以考量：

■ 1.1 減少文件數(shù)量

正如上面所討論的，基于 hash 的實(shí)現(xiàn)會(huì)產(chǎn)生大量的文件，而減少文件的數(shù)量有利于提高穩(wěn)定性和性能。Sort-Spill-Merge 的方式被分布式計(jì)算系統(tǒng)廣泛采納以達(dá)到這一目標(biāo)，首先將數(shù)據(jù)寫(xiě)入內(nèi)存緩沖區(qū)，當(dāng)內(nèi)存緩沖區(qū)填滿(mǎn)后對(duì)數(shù)據(jù)進(jìn)行排序，排序后的數(shù)據(jù)被寫(xiě)出到一個(gè)文件中，這樣總的文件數(shù)量是：（總數(shù)據(jù)量 / 內(nèi)存緩沖區(qū)大?。瑥亩募?shù)量被減少。當(dāng)所有數(shù)據(jù)寫(xiě)出完成后，將產(chǎn)生的文件合并成一個(gè)文件，從而進(jìn)一步減少文件數(shù)量并增大每個(gè)數(shù)據(jù)分區(qū)的大?。ㄓ欣陧樞蜃x?。?。

■ 1.2 打開(kāi)更少的文件

■ 1.3 最大化順序讀寫(xiě)

■ 1.4 減少讀寫(xiě) IO 放大

■ 1.5 減少內(nèi)存緩沖區(qū)消耗

2. 實(shí)現(xiàn)細(xì)節(jié)

■ 2.1 內(nèi)存數(shù)據(jù)排序

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 文件存儲(chǔ)結(jié)構(gòu)

■ 2.3 讀取 IO 調(diào)度

*// 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 數(shù)據(jù)廣播優(yōu)化

■ 2.5 數(shù)據(jù)壓縮

1. 穩(wěn)定性

2. 性能

在 Flink 中，sort-shuffle 默認(rèn)是不開(kāi)啟的，想要開(kāi)啟需要調(diào)小這個(gè)參數(shù)的配置：taskmanager.network.sort-shuffle.min-parallelism。這個(gè)參數(shù)的含義是如果數(shù)據(jù)分區(qū)的個(gè)數(shù)（一個(gè)計(jì)算任務(wù)并發(fā)需要發(fā)送數(shù)據(jù)給幾個(gè)下游計(jì)算節(jié)點(diǎn)）低于這個(gè)值，則走 hash-shuffle 的實(shí)現(xiàn)，如果高于這個(gè)值則啟用 sort-shuffle。實(shí)際應(yīng)用時(shí)，在機(jī)械硬盤(pán)上，可以配置為 1，即使用 sort-shuffle。

還有一些后續(xù)的優(yōu)化工作，包括但不限于：

1. 網(wǎng)絡(luò)連接復(fù)用，這可以提高網(wǎng)絡(luò)的建立的性能與穩(wěn)定性，相關(guān) Jira 包括 FLINK-22643 以及 FLINK-15455；
   
   
2. 多磁盤(pán)負(fù)載均衡，這有利于解決負(fù)載不均的問(wèn)題，相關(guān) Jira 包括 FLINK-21790 以及 FLINK-21789；
   
   
3. 實(shí)現(xiàn)遠(yuǎn)程數(shù)據(jù) shuffle 服務(wù)，這有利于進(jìn)一步提升批數(shù)據(jù) shuffle 的性能與穩(wěn)定性；
   
   
4. 允許用戶(hù)選擇磁盤(pán)類(lèi)型，這可以提高易用性，用戶(hù)可以根據(jù)作業(yè)的優(yōu)先級(jí)選擇使用 HDD 或者 SSD。

英文原文鏈接：

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

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

   戳我，報(bào)名 FFA 2021 大會(huì)！