【翻譯】The Broadcast State Pattern（廣播狀態）

本文翻譯自官網：The Broadcast State Pattern

• Provided APIs （提供的api）
  • BroadcastProcessFunction and KeyedBroadcastProcessFunction
• Important Considerations （注意事項）

使用State描述運算符狀態，該運算符狀態在恢復時均勻分佈在運算符的並行任務中，或者聯合使用，整個狀態用於初始化已恢復的並行任務。

第三種支持的運營商狀態是廣播狀態。 引入廣播狀態是爲了支持這樣的用例，其中來自一個流的一些數據須要被廣播到全部下游任務，其中它被本地存儲並用於處理另外一個流上的全部傳入元素。 做爲廣播狀態能夠做爲天然擬合出現的示例，能夠想象包含一組規則的低吞吐量流，咱們但願針對來自另外一個流的全部元素進行評估。 考慮到上述類型的用例，廣播狀態與其餘運營商狀態的不一樣之處在於：

　　1. 它是map格式

　　2. 它只適用於特定的操做：有一個 廣播流和一個非廣播流

　　3. 這樣的算子能夠具備不一樣名稱的多個廣播狀態。（ such an operator can have multiple broadcast states with different names. ）

Provided APIs

爲了描述提供的API，咱們將在展現其完整功能以前先舉一個示例。做爲咱們的運行示例，咱們將使用具備不一樣顏色和形狀的對象流，而且咱們想要找到跟隨特定圖案的相同顏色的對象組合，例如矩形後跟三角形。咱們假設這組有趣的模式隨着時間的推移而發展。

在此示例中，第一個流將包含Item類型的元素，其中包含Color和Shape屬性。 另外一個流將包含規則。

從Items流開始，咱們只須要經過Color作keyBy，由於咱們須要相同顏色的對。 這將確保相同顏色的元素最終在同一臺物理機器上。

// key the shapes by color
KeyedStream<Item, Color> colorPartitionedStream = shapeStream
                        .keyBy(new KeySelector<Shape, Color>(){...});

繼續執行規則，包含它們的流應該被廣播到全部下游任務，而且這些任務應該在本地存儲它們，以便它們能夠針對全部傳入的項目對它們進行評估。 下面的片斷將 1）廣播規則流和 2）使用提供的MapStateDescriptor，它將建立存儲規則的廣播狀態。

// a map descriptor to store the name of the rule (string) and the rule itself.
MapStateDescriptor<String, Rule> ruleStateDescriptor = new MapStateDescriptor<>(
            "RulesBroadcastState",
            BasicTypeInfo.STRING_TYPE_INFO,
            TypeInformation.of(new TypeHint<Rule>() {}));
        
// broadcast the rules and create the broadcast state
BroadcastStream<Rule> ruleBroadcastStream = ruleStream
                        .broadcast(ruleStateDescriptor);

最後，爲了根據Item流中的傳入元素評估規則，咱們須要：

　　1. connect兩個流

　　2. 指定咱們的匹配檢測邏輯

將流（鍵控或非鍵控）與BroadcastStream鏈接能夠經過在非廣播流上調用connect（），並將BroadcastStream做爲參數來完成。 這將返回一個BroadcastConnectedStream，咱們可使用特殊類型的CoProcessFunction調用process（）。 該函數將包含咱們的匹配邏輯。 函數的確切類型取決於非廣播流的類型：

　　* 若是它（非廣播流）是鍵控的，方法是 KeyedBroadcastProcessFunction

　　* 若是它（非廣播流）是非鍵控的，方法是BroadcastProcessFunction

鑑於咱們的非廣播流是鍵控的，如下代碼段包含以上調用：

注意：應在非廣播流上調用connect，並將Broadcast Stream做爲參數。

DataStream<Match> output = colorPartitionedStream
                 .connect(ruleBroadcastStream)
                 .process(
                     
                     // type arguments in our KeyedBroadcastProcessFunction represent: 
                     //   1. the key of the keyed stream
                     //   2. the type of elements in the non-broadcast side
                     //   3. the type of elements in the broadcast side
                     //   4. the type of the result, here a string
                     
                     new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
                         // my matching logic
                     }
                 )

BroadcastProcessFunction and KeyedBroadcastProcessFunction

 與CoProcessFunction的狀況同樣，這些函數有兩種實現方法; processBroadcastElement（）負責處理廣播流中的傳入元素和processElement（）用於非廣播流。 這些方法的完整簽名以下：

// non-keyed
public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {

    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;

    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
}
// keyed
public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {

    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;

    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;

    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
}

首先要注意的是，兩個函數都須要執行processBroadcastElement（）方法來處理廣播端的元素，而processElement（）則處理非廣播端的元素。

這兩種方法在提供的上下文中有所不一樣。 非廣播側具備ReadOnlyContext，而廣播側具備Context。

這兩個上下文（ctx在如下枚舉中）：

 　　1. 容許訪問廣播狀態：ctx.getBroadcastState（MapStateDescriptor <K，V> stateDescriptor）

 　　2. 容許查詢元素的時間戳：ctx.timestamp（）

　　 3. 獲取當前水印：ctx.currentWatermark（）

　　 4. 獲取當前處理時間：ctx.currentProcessingTime（）

　　 5. 向側邊流發出元素：ctx.output（OutputTag <X> outputTag，X value）

getBroadcastState（）中的stateDescriptor應該與上面的.broadcast（ruleStateDescriptor）中的stateDescriptor相同。

不一樣之處在於每一個流對廣播狀態的訪問類型。 廣播方對其(廣播狀態)具備讀寫訪問權限，而非廣播方具備只讀訪問權（所以命名：thus the names）。 緣由是在Flink中沒有跨任務通訊。 所以，爲了保證廣播狀態中的內容在咱們的運算符的全部並行實例中是相同的，咱們只對廣播端提供讀寫訪問，廣播端在全部任務中看到相同的元素，而且咱們須要對每一個任務進行計算。 這一側的傳入元素在全部任務中都是相同的。 忽略此規則會破壞狀態的一致性保證，從而致使不一致且一般難以調試的結果。

注意：`processBroadcast（）`中實現的邏輯必須在全部並行實例中具備相同的肯定性行爲！

最後，因爲KeyedBroadcastProcessFunction在鍵控流上運行，它暴露了一些BroadcastProcessFunction不可用的功能。 那是：

　　1. processElement（）方法中的ReadOnlyContext能夠訪問Flink的底層計時器服務，該服務容許註冊事件和/或處理時間計時器。 當一個計時器觸發時，onTimer（）（如上所示）被一個OnTimerContext調用，它暴露了與ReadOnlyContext相同的功能和

　　　　* 詢問觸發的計時器是事件仍是處理時間的能力

　　　　* 查詢與計時器關聯的key

　　2. processBroadcastElement（）方法中的Context包含方法applyToKeyedState（StateDescriptor <S，VS> stateDescriptor，KeyedStateFunction <KS，S> function）。 這容許註冊KeyedStateFunction以應用於與提供的stateDescriptor相關聯的全部鍵的全部狀態。

注意：只能在`KeyedBroadcastProcessFunction`的`processElement（）`中註冊定時器。 在`processBroadcastElement（）`方法中是不可能的，由於沒有與廣播元素相關聯的鍵。

回到咱們的原始示例，咱們的KeyedBroadcastProcessFunction可能以下所示：

new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {

    // store partial matches, i.e. first elements of the pair waiting for their second element
    // we keep a list as we may have many first elements waiting
    private final MapStateDescriptor<String, List<Item>> mapStateDesc =
        new MapStateDescriptor<>(
            "items",
            BasicTypeInfo.STRING_TYPE_INFO,
            new ListTypeInfo<>(Item.class));

    // identical to our ruleStateDescriptor above
    private final MapStateDescriptor<String, Rule> ruleStateDescriptor = 
        new MapStateDescriptor<>(
            "RulesBroadcastState",
            BasicTypeInfo.STRING_TYPE_INFO,
            TypeInformation.of(new TypeHint<Rule>() {}));

    @Override
    public void processBroadcastElement(Rule value,
                                        Context ctx,
                                        Collector<String> out) throws Exception {
        ctx.getBroadcastState(ruleStateDescriptor).put(value.name, value);
    }

    @Override
    public void processElement(Item value,
                               ReadOnlyContext ctx,
                               Collector<String> out) throws Exception {

        final MapState<String, List<Item>> state = getRuntimeContext().getMapState(mapStateDesc);
        final Shape shape = value.getShape();
    
        for (Map.Entry<String, Rule> entry :
                ctx.getBroadcastState(ruleStateDescriptor).immutableEntries()) {
            final String ruleName = entry.getKey();
            final Rule rule = entry.getValue();
    
            List<Item> stored = state.get(ruleName);
            if (stored == null) {
                stored = new ArrayList<>();
            }
    
            if (shape == rule.second && !stored.isEmpty()) {
                for (Item i : stored) {
                    out.collect("MATCH: " + i + " - " + value);
                }
                stored.clear();
            }
    
            // there is no else{} to cover if rule.first == rule.second
            if (shape.equals(rule.first)) {
                stored.add(value);
            }
    
            if (stored.isEmpty()) {
                state.remove(ruleName);
            } else {
                state.put(ruleName, stored);
            }
        }
    }
}

Important Considerations

 在描述提供的API以後，本節重點介紹使用廣播狀態時要記住的重要事項。 這些是：

* There is no cross-task communication:  如前所述，這就是爲何只有（Keyed）-BroadcastProcessFunction的廣播端能夠修改廣播狀態的內容。 此外，用戶必須確保全部任務以相同的方式爲每一個傳入元素修改廣播狀態的內容。 不然，不一樣的任務可能具備不一樣的內容，從而致使不一致的結果。

* Order of events in Broadcast State may differ across tasks: 雖然廣播流的元素保證全部元素將（最終）轉到全部下游任務，但元素能夠以不一樣的順序到達每一個任務。 所以，每一個傳入元素的狀態更新毫不取決於傳入事件的順序。

* All tasks checkpoint their broadcast state: 雖然檢查點發生時全部任務在廣播狀態中都具備相同的元素（檢查點barriers 不會覆蓋元素），但全部任務都會檢查其廣播狀態，而不只僅是其中一個。 這是一個設計決策，以免在恢復期間從同一文件中讀取全部任務（從而避免熱點），儘管它的代價是將檢查點狀態的大小增長p（=並行度）。 Flink保證在恢復/從新縮放時不會出現重複數據，也不會丟失數據。 在具備相同或更小並行度的恢復的狀況下，每一個任務讀取其檢查點狀態。 在按比例放大時，每一個任務都讀取其本身的狀態，其他任務（p_new-p_old）以循環方式讀取先前任務的檢查點。

* No RocksDB state backend: 廣播狀態在運行時保留在內存中，而且應該相應地進行內存配置。 這適用於全部算子的state。

etl實例： 基於Broadcast 狀態的Flink Etl Demo