flink資料廣播場景總結

資料集廣播，主要分為廣播變數，廣播維表(資料集)兩種，一種為變數，一種為常量(抽象的說法)；

一.資料廣播背景

對於小變數，小資料集，需要和大資料集，大流進行聯合計算的時候，往往把小資料集廣播出去，整體直接和大資料集(流)的分散式最小粒度資料進行計算，最後把計算結果合併，這樣效率更高，省去分散式節點之間的資料傳輸及二次計算。

例如：在Flink使用場景中，外部的配置檔案或計算規則及維表等進行預載入，並定期更新，流式計算中廣播小變數等場景。

資料集的廣播，主要有以下幾種方式可以實現

1.預載入

在運算元的open()方法中讀取MySQL或其他儲存介質，獲取全量維表資訊比如在運算元RichMapFunction的open()方法中獲取全部資料，然後在運算元中進行使用，這種方法的缺點是如果外部資料更新了Flink是沒法知道的，這就需要在開啟一個定時任務定時從MySQL中獲取最新的資料。

2.外部查詢

資料不需要儲存，僅需要用到外部資料的時候去進行查詢，可以保證查詢到的資料是最新的，但是對於吞吐量較高的場景，可能與外部（比如MySQL）互動就變成了 Flink任務的瓶頸，雖然可以設定為非同步I/O的形式進行互動優化，但優化程度一般有限。

3.本地快取

需要設定過期時間或者定時更新資料，當資料到達過期時間後從新從外部獲取，或者定時從外部撈取資料進行更新，不能在外部資料發生變動時，及時更新到Flink程式中。

預載入和本地快取難以應對當外部資料發生變化時，資料實時在Flink中保持更新。

二.什麼是廣播

類似於全域性性共享的資料，詳見官方文件

https://flink.apache.org/2019/06/26/broadcast-state.html

https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/stream/state/broadcast_state.html

廣播的優勢

廣播變數建立後，它可以執行在叢集中的任何function上，而不需要多次傳遞給叢集節點，可以直接在記憶體中拿資料，避免了大量的shuffle，導致叢集效能下降。我們可以把一個dataset或者不變的快取物件（例如maplist集合物件等）資料集廣播出去，然後不同的任務在節點上都能夠獲取到，並在每個節點上只會存在一份，而不是在每個併發執行緒中存在。

如果不使用broadcast，則在每個節點中的每個任務中都需要拷貝一份dataset資料集，比較浪費記憶體(也就是一個節點中可能會存在多份dataset資料)。廣播變數，可以藉助下圖輔助理解。

三.廣播的使用

根據廣播使用場景將廣播的型別分為廣播變數和廣播流(其實廣播原理是一樣的)。

1.廣播變數

將廣播的資料作為一個整體或物件廣播，比如從MySQL中一次獲取全部資料，然後廣播出去，因為資料在MySql中，如果MySql中某條記錄發生變動，Flink的souce是沒法知道，也不會廣播。所以只能在souce中定時從MySql中獲取全部資料，然後廣播更新。

示例資料格式：

kafka源流資料，只有itemid，沒有ip和port
{"host":"orcl", "itemid":"7875", "value":1}
{"host":"orcl2", "itemid":"7876", "value":2}
規則資料集在MySql，itemid關聯ip和port
itemid ip port
7875 192.168.199.105 1521
7876 192.168.199.106 1526

自定義MySql source, 定時從Mysql獲取全部資料

@Override
public void run(SourceContext<HashMap<String, Tuple2<String, Integer>>> ctx) {
try {
while (isRunning) {
  HashMap<String, Tuple2<String, Integer>> output = new HashMap<>();
  ResultSet resultSet = preparedStatement.executeQuery();
  //每隔60s獲取全部外部資料集
  while (resultSet.next()) {
String itemid = resultSet.getString("itemid");
String ip = resultSet.getString("ip");
int port = resultSet.getInt("port");
output.put(itemid, new Tuple2<>(ip, port));
  }
  ctx.collect(output);
  Thread.sleep(1000 * 60);
  }
} catch (Exception ex) {
  log.error("從Mysql獲取配置異常...", ex);
}
}

廣播程式碼實現：

public void processElement(Map<String,Object> value, ReadOnlyContext ctx, Collector<Map<String,Object>> out) throws Exception {
//從廣播中獲取全量資料
ReadOnlyBroadcastState<Void, Map<String, Tuple2<String, Integer>>> broadcastState = ctx.getBroadcastState(ruleStateDescriptor);
//獲取全部規則資料進行匹配

Map<String, Tuple2<String, Integer>> itemrules= broadcastState.get(null);
  //規則資料集為空跳過
    if(itemrules==null) {
     return;
    }
    //事件流中的itemid
    Object itemidObj = value.get("itemid"); // value kafka流中資料獲取itemid
if (itemidObj == null) {
return;
}
    Tuple2<String, Integer> itemruld = itemrules.get(itemidObj.toString());
    if(itemruld!=null){
     //匹配成功增加ip,port欄位
     value.put("ip", itemruld.f0);
     value.put("port", itemruld.f1);
        out.collect(value);
    }
}
@Override
public void processBroadcastElement(HashMap<String, Tuple2<String, Integer>> value, Context ctx, Collector<Map<String,Object>> out) throws Exception {
   //資料全部更新
BroadcastState<Void, Map<String, Tuple2<String, Integer>>> broadcastState = ctx
  .getBroadcastState(ruleStateDescriptor);
//每次更新全部規則資料
broadcastState.put(null, value);
    System.out.println("規則全部更新成功，更新item規則：" + value);
}

執行結果：

//itemid=7875關聯ip=192.168.199.104
{host=orcl, itemid=7875, value=1, ip=192.168.199.104, port=1521}
//手動將mysql中的ip=192.168.199.104改為ip=192.168.199.105，在source 休眠結束後將會更新資料
規則更新成功，更新item規則：{7875=(192.168.199.105,1521), 7876=(192.168.199.106,1526)}
//itemid=7875關聯ip=192.168.199.105
{host=orcl, itemid=7875, value=1, ip=192.168.199.105, port=1521}

這種方式和預載入很像，都是通過定時任務載入全部資料，只不過是方法的位置不同，一個是在自定義source中設定休眠時間，另外一個是在運算元的open方法中設定定時任務，廣播變數的方式同樣無法做到資料修改後實時更新。

2.廣播流

當資料來源於kafka時，Flink消費kafka獲取流，將流資料儲存在廣播狀態中，稱之為廣播流，不同於廣播變數一次獲取全部資料，廣播流是kafka新增一條記錄就將這條記錄儲存到廣播中，那廣播流如何實現外部資料的新增和更新？

kafka源流資料，只有itemid，沒有ip和port
{"host":"orcl", "itemid":"7875", "value":1}
{"host":"orcl2", "itemid":"7876", "value":2}
規則資料集在kafka，itemid關聯ip和port
{"itemid":"7875","ip":"192.168.199.104","port":1521}
{"itemid":"7876","ip":"192.168.199.106","port":1526}

2.1 外部資料新增和修改記錄

// 廣播狀態底層結構是Map結構
//kafka中的資料,flink消費後儲存到廣播狀態,在廣播狀態中以itemid為key進行儲存
{"itemid":"7875","ip":"192.168.199.104","port":1521}
{"itemid":"7876","ip":"192.168.199.106","port":1526}
//新增往kafka寫入新記錄(key不相同),flink會持續消費kafka並將資料通過Map的put()方法存入廣播狀態
//修改往kafka寫入新記錄(key相同)，put()方法覆蓋之前的這條記錄以達到更新的目的
//比如需要更新itemid的ip和port值,要求往kafka中寫入一條新資料,比如更新itemid 7875的ip和port
{"itemid":"7875","ip":"192.168.199.105","port":1525}

2.2 刪除記錄

//kafka中的資料,flink消費後儲存到廣播狀態,以itemid為key進行儲存
{"itemid":"7875","ip":"192.168.199.104","port":1521}
{"itemid":"7876","ip":"192.168.199.106","port":1526}
//如果需要刪除某條記錄，往kafka中寫入帶有key的資料和刪除標記即可
//比如刪除itemid為7875的記錄,要求往kafka中寫入一條新資料,程式刪除廣播中itemid7875的記錄
{"itemid":"7875","isRemove":true}

由於消費kafka流是實時的，kafka的新記錄會實時進行消費，根據新記錄的內容對廣播資料實時的進行新增，修改或刪除

同時由於kafka中的資料是不可變的，當程式需要重啟時，只需從頭消費kafka即可，由於具有冪等性，最終的廣播資料是不會變的。

示例程式碼

//Flink消費外部kafka規則資料作為流
FlinkKafkaConsumer<ItemRuleEntiy> ruleKafkaConsumer = new FlinkKafkaConsumer<ItemRuleEntiy>("topic",new ItemRuleEntiyPojoSchema(),properties);
DataStream<ItemRuleEntiy> ruleStream = env.addSource(ruleKafkaConsumer);

//廣播方法
@Override
public void processBroadcastElement(ItemRuleEntiy value,
BroadcastProcessFunction<Map<String, Object>, ItemRuleEntiy, Map<String, Object>>.Context ctx,
Collector<Map<String, Object>> out) throws Exception {
BroadcastState<String, ItemRuleEntiy> broadcastState = ctx.getBroadcastState(ruleStateDescriptor);
if (StringUtils.isNoneBlank(value.getItemid())) {
System.out.println("獲取到新的廣播規則:" + value);
//相比廣播變數，這裡每次只存一條規則，相同key則覆蓋修改
broadcastState.put(value.getItemid(), value); // 存放資料到廣播
}
}
@Override
public void processElement(Map<String, Object> value,
BroadcastProcessFunction<Map<String, Object>, ItemRuleEntiy, Map<String, Object>>.ReadOnlyContext ctx,
Collector<Map<String, Object>> out) throws Exception {
ReadOnlyBroadcastState<String, ItemRuleEntiy> broadcastState = ctx.getBroadcastState(ruleStateDescriptor);
Object itemidObj = value.get("itemid"); // 源kafka流中資料獲取itemid
if (itemidObj == null) {
return;
}
// 根據item從廣播資料中查詢規則,能查到,則增加ip,port欄位
ItemRuleEntiy itemRule = broadcastState.get(itemidObj.toString());
if (itemRule != null) { // 從廣播中撈取到資料時
value.put("ip", itemRule.getIp());
value.put("port", itemRule.getPort());
out.collect(value);
}
}

執行結果

// 所有廣播規則資料:
7875={itemid=7875, ip=192.168.199.104, port=1521}
7876={itemid=7876, ip=192.168.199.106, port=1526}
//itemid=7875關聯ip=192.168.199.104
({host=orcl, itemid=7875,value=1, ip=192.168.199.104, port=1521},7875)
//kafka寫入{itemid=7875, ip=192.168.199.105, port=1521}
獲取到新的廣播規則:{itemid=7875, ip=192.168.199.105, port=1521}
//itemid=7875關聯ip=192.168.199.105
{host=orcl,itemid=7875, value=1, ip=192.168.199.105, port=1521},7875)

四.總結

通過kafka廣播流的方式最終實現了Flink與外部資料互動的實時更新，不僅是kafka，還有MQ，甚至檔案格式都可以作為廣播流，廣播流要求資料不能從內部更改（無法作為流訊息被實時消費),只能通過新增的方式進行修改和刪除(新增記錄中key相同的表示覆蓋修改，帶key和刪除標記的表示刪除)

相比表(mysql,oracle)只是結果的呈現，日誌(kafka或其它佇列)是一種帶有時間維度（或先後順序）資訊的儲存，可以說表是二維的，日誌是三維的，通過日誌可以復原每個時間點的表，但是表不能還原日誌。

廣播作為一種流（流明顯帶有時間特性)，所以當不帶時間維度的表作為流時，是沒法形成真正意義上的流，只能通過定時獲取表的全部資料作為偽流，流中每個時間點的資料也只能是全量表資料，同時定時也就沒法做到實時獲取。只有帶時間維度的日誌作為流時，才能做到實時獲取，而且每次只獲取最新的一條記錄即可，不用每次獲取全部資料。

flink資料廣播場景總結