一篇聊聊Go錯(cuò)誤封裝機(jī)制
%w 是用于錯(cuò)誤包裝（Error Wrapping）的格式化動(dòng)詞。它是用于 fmt.Errorf 和 fmt.Sprintf 函數(shù)中的一個(gè)特殊格式化動(dòng)詞，用于將一個(gè)錯(cuò)誤（或其他可打印的值）包裝在一個(gè)新的錯(cuò)誤中。使
flink任務(wù)在執(zhí)行過(guò)程中，一個(gè)流（stream）包含一個(gè)或多個(gè)分區(qū)（Stream partition）。TaskManager中的一個(gè)slot的subtask就是一個(gè)stream partition（流分區(qū)），一個(gè)Job的流（stream）分布在多個(gè)不同的Slot上執(zhí)行。每一個(gè)算子可以包含一個(gè)或多個(gè)子任務(wù)（subtask），這些subtask執(zhí)行在不同的分區(qū)中，本質(zhì)是在不同的線程、不同的物理機(jī)或不同的容器中彼此互不依賴地執(zhí)行。
本地線程內(nèi)的流數(shù)據(jù)傳輸（同一個(gè)SubTask中）：同一個(gè)SubTask內(nèi)的兩個(gè)Operator（屬于同一個(gè)OperatorChain）之間的數(shù)據(jù)傳輸是方法調(diào)用，即上游算子處理完數(shù)據(jù)后，直接調(diào)用下游算子的processElement方法。
本地線程間的流數(shù)據(jù)傳輸（同一個(gè)TaskManager的不同SubTask中）：即同一個(gè)TaskManager（JVM進(jìn)程）中的不同Task（線程，本質(zhì)上是SubTask）的算子之間的數(shù)據(jù)傳輸，通過(guò)本地內(nèi)存進(jìn)行數(shù)據(jù)傳遞，存在數(shù)據(jù)序列化和反序列過(guò)程。
跨網(wǎng)絡(luò)的流數(shù)據(jù)傳輸（不同TaskManager的SubTask中）：采用Netty框架，通過(guò)Socket傳遞，也存在數(shù)據(jù)序列化和反序列過(guò)程。
flink中的重分區(qū)算子定義上下游subtask之間數(shù)據(jù)傳遞的方式，SubTask之間進(jìn)行數(shù)據(jù)傳遞模式有兩種，一種是one-to-one（forwarding）模式，另一種是redistributing的模式。
圖片
flink中的重分區(qū)算子除了keyBy以外，還有broadcast、rebalance、shuffle、rescale、global、partitionCustom等多種算子，它們的分區(qū)方式各不相同。需要注意的是，這些算子中除了keyBy能將DataStream轉(zhuǎn)化為KeyedStream外，其它重分區(qū)算子均不會(huì)改變Stream的類型。
數(shù)據(jù)在算子之間流動(dòng)需要依靠分區(qū)策略（分區(qū)器），F(xiàn)link目前內(nèi)置了以下幾種分區(qū)策略和自定義分區(qū)策略。已實(shí)現(xiàn)的分區(qū)策略對(duì)應(yīng)的API為：
圖片
自定義分區(qū)策略的API為CustomPartitionerWrapper。
各個(gè)API的繼承關(guān)系如下圖所示：
圖片
ChannelSelector是分區(qū)策略的頂層接口，其決定了記錄應(yīng)該寫(xiě)入哪個(gè)邏輯通道，通道可理解為下游算子的某個(gè)實(shí)例，或下游并行算子的某個(gè)子任務(wù)。該接口的定義源碼如下：
圖片
抽象類StreamPartitioner實(shí)現(xiàn)了ChannelSelector接口，是一個(gè)用于流程序的特殊的ChannelSelector，其中定義了一些通用的分區(qū)策略方法。Flink中的所有分區(qū)策略（分區(qū)器）都繼承了StreamPartitioner類，并且實(shí)現(xiàn)了各自獨(dú)有的分區(qū)規(guī)則。
該分區(qū)策略位于Blink的Table API的org.apache.flink.table.runtime.partitioner包中，是一種針對(duì)BinaryRowData的哈希分區(qū)器。BinaryRowData是RowData的實(shí)現(xiàn)，可以顯著減少Java對(duì)象的序列化／反序列化。RowData用于表示結(jié)構(gòu)化數(shù)據(jù)類型，運(yùn)行時(shí)通過(guò)Table API或SQL管道傳遞的所有頂級(jí)記錄都是RowData的實(shí)例。關(guān)于BinaryHashPartitioner，我們這里不做過(guò)多講解。
廣播分區(qū)策略將上游數(shù)據(jù)記錄輸出到下游算子的每個(gè)并行實(shí)例中，即下游每個(gè)分區(qū)都會(huì)有上游的所有數(shù)據(jù)。使用DataStream的broadcast()方法即可設(shè)置該DataStream向下游發(fā)送數(shù)據(jù)時(shí)使用廣播分區(qū)策略。
來(lái)一段代碼演示下：
/** * 微信公眾號(hào)：老周聊架構(gòu) */public class PartitionerTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(3); DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5, 6); //1.分區(qū)策略前的操作 //輸出dataStream每個(gè)元素及所屬的子任務(wù)編號(hào) dataStream.map(new RichMapFunction<Integer, Object>() { @Override public Object map(Integer value) throws Exception { System.out.println(String.format("元素值: %s, 分區(qū)策略前，子任務(wù)編號(hào): %s", value, getRuntimeContext().getIndexOfThisSubtask())); return value; } }); //2.設(shè)置分區(qū)策略 //設(shè)置DataStream向下游發(fā)送數(shù)據(jù)時(shí)使用的策略 DataStream<Integer> dataStreamAfter = dataStream.broadcast(); //3.分區(qū)策略后的操作 dataStreamAfter.map(new RichMapFunction<Integer, Object>() { @Override public Object map(Integer value) throws Exception { System.out.println(String.format("元素值: %s, 分區(qū)策略后，子任務(wù)編號(hào): %s", value, getRuntimeContext().getIndexOfThisSubtask())); return value; } }).print(); env.execute("PartitionerTest Job"); }}直接IDEA控制臺(tái)輸出：
圖片
從輸出結(jié)果可以看出，數(shù)據(jù)共分為3個(gè)分區(qū)（編號(hào)為0、1、2）。執(zhí)行分區(qū)策略前，每個(gè)元素所屬的分區(qū)：
圖片
執(zhí)行分區(qū)策略后，每個(gè)元素所屬的分區(qū)如下：
圖片
對(duì)比表發(fā)現(xiàn)，廣播分區(qū)策略將上游每個(gè)元素分別發(fā)送到了下游算子的所有分區(qū)，這種策略會(huì)把數(shù)據(jù)復(fù)制多份，向下游算子的每個(gè)分區(qū)發(fā)送一份。
圖片
我們把上面的任務(wù)提交到Flink，同樣也可以看出前面分區(qū)前每個(gè)子任務(wù)兩條數(shù)據(jù)，分區(qū)后每個(gè)子任務(wù)六條數(shù)據(jù)。
圖片
圖片
轉(zhuǎn)發(fā)分區(qū)策略只將元素轉(zhuǎn)發(fā)給本地運(yùn)行的下游算子的實(shí)例，即將元素發(fā)送到與當(dāng)前算子實(shí)例在同一個(gè)TaskManager的下游算子實(shí)例，而不需要進(jìn)行網(wǎng)絡(luò)傳輸。要求上下游算子并行度一樣，這樣上下游算子可以同屬一個(gè)子任務(wù)。
這里把上面的代碼調(diào)整下：
dataStream.forward()
IDEA控制臺(tái)輸出：
圖片
從輸出結(jié)果可以看出，數(shù)據(jù)共分為3個(gè)分區(qū)（編號(hào)為0、1、2）。執(zhí)行分區(qū)策略前，每個(gè)元素所屬的分區(qū)：
圖片
執(zhí)行分區(qū)策略后，每個(gè)元素所屬的分區(qū)如下：
圖片
對(duì)比發(fā)現(xiàn)，轉(zhuǎn)發(fā)分區(qū)策略將上游同一個(gè)分區(qū)的元素發(fā)送到了下游同一個(gè)分區(qū)中。使用數(shù)據(jù)流圖表示如下圖：
圖片
在上下游的算子沒(méi)有指定分區(qū)策略的情況下，如果上下游的算子并行度一致，則默認(rèn)使用ForwardPartitioner，否則使用RebalancePartitioner。在StreamGraph類的源碼中可以看到該規(guī)則：
圖片
對(duì)于ForwardPartitioner，必須保證上下游算子并行度一致，否則會(huì)拋出異常。
圖片
全局分區(qū)策略將上游所有元素發(fā)送到下游子任務(wù)編號(hào)等于0的分區(qū)算子實(shí)例上（下游第一個(gè)實(shí)例）。
這里把上面的代碼調(diào)整下：
IDEA控制臺(tái)輸出：
圖片
分區(qū)前：
圖片
分區(qū)后：
圖片
全局分區(qū)策略將上游所有分區(qū)中的所有元素發(fā)送到了下游編號(hào)為0的分區(qū)中：
圖片
Key分區(qū)策略根據(jù)元素Key的Hash值輸出到下游算子指定的實(shí)例。keyBy()算子底層正是使用的該分區(qū)策略，底層最終會(huì)調(diào)用KeyGroupStreamPartitioner的selectChannel()方法，計(jì)算每個(gè)Key對(duì)應(yīng)的通道索引（通道編號(hào)，可理解為分區(qū)編號(hào)），根據(jù)通道索引將Key發(fā)送到下游相應(yīng)的分區(qū)中。selectChannel()方法源碼如下：
圖片
圖片
總的來(lái)說(shuō)，F(xiàn)link底層計(jì)算通道索引（分區(qū)編號(hào)）的流程如下：
平衡分區(qū)策略使用循環(huán)遍歷下游分區(qū)的方式，將上游元素均勻分配給下游算子的每個(gè)實(shí)例。每個(gè)下游算子的實(shí)例都具有相等的負(fù)載。當(dāng)數(shù)據(jù)流中的元素存在數(shù)據(jù)傾斜時(shí)，使用該策略對(duì)性能有很大的提升。
這里把上面的代碼調(diào)整下：
dataStream.setParallelism(2);
dataStreamAfter.setParallelism(3);
dataStream.rebalance()
IDEA控制臺(tái)輸出：
圖片
分區(qū)前：
圖片
分區(qū)后：
圖片
平衡分區(qū)策略將上游所有元素均勻發(fā)送到了下游算子的所有分區(qū)：
圖片
重新調(diào)節(jié)分區(qū)策略基于上下游算子的并行度，將元素以循環(huán)的方式輸出到下游算子的每個(gè)實(shí)例。類似于平衡分區(qū)策略，但又與平衡分區(qū)策略不同。
上游算子將元素發(fā)送到下游哪一個(gè)算子實(shí)例，取決于上游和下游算子的并行度。例如，如果上游算子的并行度為2，而下游算子的并行度為4，那么一個(gè)上游算子實(shí)例將把元素均勻分配給兩個(gè)下游算子實(shí)例，而另一個(gè)上游算子實(shí)例將把元素均勻分配給另外兩個(gè)下游算子實(shí)例。相反，如果下游算子的并行度為2，而上游算子的并行度為4，那么兩個(gè)上游算子實(shí)例將分配給一個(gè)下游算子實(shí)例，而另外兩個(gè)上游算子實(shí)例將分配給另一個(gè)下游算子實(shí)例。
假設(shè)上游算子并行度為2，分區(qū)編號(hào)為A和B，下游算子并行度為4，分區(qū)編號(hào)為1、2、3、4，那么A將把數(shù)據(jù)循環(huán)發(fā)送給1和2，B則把數(shù)據(jù)循環(huán)發(fā)送給3和4。假設(shè)上游算子并行度為4，編號(hào)為A、B、C、D，下游算子并行度為2，編號(hào)為1、2，那么A和B把數(shù)據(jù)發(fā)送給1，C和D則把數(shù)據(jù)發(fā)送給2。
這里把上面的代碼調(diào)整下：
同時(shí)將第一個(gè)map算子的并行度設(shè)置為2，第二個(gè)map算子的并行度設(shè)置為4。
IDEA控制臺(tái)輸出：
圖片
分區(qū)前：
圖片
分區(qū)后：
圖片
圖片
接下來(lái)改變map算子的并行度，將第一個(gè)map算子的并行度設(shè)置為4，第二個(gè)map算子的并行度設(shè)置為2。
圖片
如果想將元素均勻地輸出到下游算子的每個(gè)實(shí)例，以實(shí)現(xiàn)負(fù)載均衡，同時(shí)又不希望使用平衡分區(qū)策略的全局負(fù)載均衡，則可以使用重新調(diào)節(jié)分區(qū)策略。該策略會(huì)盡可能避免數(shù)據(jù)在網(wǎng)絡(luò)間傳輸，而能否避免還取決于TaskManager的Task Slot數(shù)量、上下游算子的并行度等。
隨機(jī)分區(qū)策略將上游算子元素輸出到下游算子的隨機(jī)實(shí)例中。元素會(huì)被均勻分配到下游算子的每個(gè)實(shí)例。這種策略可以實(shí)現(xiàn)計(jì)算任務(wù)的負(fù)載均衡。
這里把上面的代碼調(diào)整下：
dataStream.shuffle()
這里就不做過(guò)多演示了。我們下面來(lái)看下自定義分區(qū)策略。
自定義分區(qū)策略的API為CustomPartitionerWrapper。該策略允許開(kāi)發(fā)者自定義規(guī)則將上游算子元素發(fā)送到下游指定的算子實(shí)例中。
新建分區(qū)器類MyCustomPartitioner并實(shí)現(xiàn)接口Partitioner（Object表示分區(qū)Key的數(shù)據(jù)類型），實(shí)現(xiàn)其中未實(shí)現(xiàn)的方法partition()，在該方法中添加相應(yīng)的分區(qū)邏輯。
/** * 自定義分區(qū)策略 * 微信公眾號(hào)：老周聊架構(gòu) */public class MyCustomPartitioner implements Partitioner { @Override public int partition(Object key, int numPartitions) { if (key.equals("chinese")) { return 0; } else if (key.equals("math")) { return 1; } else { return 2; } }}上述代碼通過(guò)partition()方法取得分區(qū)編號(hào)，將Key值等于chinese的元素分配到編號(hào)為0的分區(qū)，將Key值等于math的元素分配到編號(hào)為1的分區(qū)，其余元素分配到編號(hào)為2的分區(qū)。
調(diào)用DataStream的partitionCustom()方法傳入自定義分區(qū)器類MyCustomPartitioner的實(shí)例，可以對(duì)DataStream按照自定義規(guī)則進(jìn)行重新分區(qū)，代碼如下：
/** * 自定義分區(qū)策略 * 微信公眾號(hào)：老周聊架構(gòu) */public class CustomPartitionerTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(3); DataStream<String> dataStream = env.fromElements("chinese，98", "math，88", "english，96"); //1.分區(qū)策略前的操作 //輸出dataStream每個(gè)元素及所屬的子任務(wù)編號(hào) SingleOutputStreamOperator<Map<String, Integer>> dataStreamBefore = dataStream.map(new RichMapFunction<String, Map<String, Integer>>() { @Override public Map<String, Integer> map(String value) throws Exception { System.out.println(String.format("元素值: %s, 分區(qū)策略前，子任務(wù)編號(hào): %s", value, getRuntimeContext().getIndexOfThisSubtask())); Map<String, Integer> map = new HashMap<>(); map.put(value.split("，")[0], Integer.parseInt(value.split("，")[1])); return map; } }).setParallelism(2); //2.設(shè)置分區(qū)策略 //設(shè)置DataStream向下游發(fā)送數(shù)據(jù)時(shí)使用的策略 DataStream<Map<String, Integer>> dataStreamAfter = dataStreamBefore.partitionCustom(new MyCustomPartitioner(), value -> value); //3.分區(qū)策略后的操作 dataStreamAfter.map(new RichMapFunction<Map<String, Integer>, Map<String, Integer>>() { @Override public Map<String, Integer> map(Map<String, Integer> value) throws Exception { System.out.println(String.format("元素值: %s, 分區(qū)策略后，子任務(wù)編號(hào): %s", value, getRuntimeContext().getIndexOfThisSubtask())); return value; } }).setParallelism(3).print(); env.execute("CustomPartitionerTest Job"); }}分區(qū)前：
圖片
分區(qū)后：
圖片
自定義分區(qū)策略將上游所有元素按照自定義的規(guī)則發(fā)送到了下游的3個(gè)分區(qū)中。
把任務(wù)給到Flink上去跑，發(fā)現(xiàn)：
圖片
這是因?yàn)榉盒筒脸旅娴腄ataStream泛型需要指定類型，不能
圖片
小知識(shí)：
在編譯之后程序會(huì)采取去泛型化的措施。也就是說(shuō)Java中的泛型，只在編譯階段有效。在編譯過(guò)程中，正確檢驗(yàn)泛型結(jié)果后，在運(yùn)行時(shí)會(huì)將泛型的相關(guān)信息擦除，編譯器只會(huì)在對(duì)象進(jìn)入JVM和離開(kāi)JVM的邊界處添加類型檢查和轉(zhuǎn)換的方法，泛型的信息不會(huì)進(jìn)入到運(yùn)行時(shí)階段，這就是所謂的Java類型擦除。
類型加好以后，再跑一下任務(wù)，會(huì)出現(xiàn)任務(wù)成功。
圖片
圖片
本文鏈接：http://www.www897cc.com/showinfo-26-75362-0.html聊聊Flink：Flink的分區(qū)機(jī)制
聲明：本網(wǎng)頁(yè)內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問(wèn)題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請(qǐng)聯(lián)系：2376512515@qq.com 津ICP備18001702號(hào)
津公網(wǎng)安備 12010102000574號(hào)