在Flink使用場景中，常常需要獲取外部配置或規(guī)則數(shù)據(jù)后進(jìn)行計算或規(guī)則匹配，同時當(dāng)外部數(shù)據(jù)發(fā)生變化時，F(xiàn)link中程序也要保持更新，有以下幾種方式可以實現(xiàn)

在算子的open()方法中讀取MySQL或其他存儲介質(zhì)，獲取全量維表信息比如在算子RichMapFunction的open()方法中獲取全部數(shù)據(jù)，然后在算子中進(jìn)行使用，這種方法的缺點是如果外部數(shù)據(jù)更新了Flink是沒法知道的，這就需要在開啟一個定時任務(wù)定時從MySQL中獲取最新的數(shù)據(jù)。

數(shù)據(jù)不需要存儲，僅需要用到外部數(shù)據(jù)的時候去進(jìn)行查詢，可以保證查詢到的數(shù)據(jù)是最新的，但是對于吞吐量較高的場景，可能與外部（比如MySQL）交互就變成了 Flink任務(wù)的瓶頸，雖然可以設(shè)置為異步I/O的形式進(jìn)行交互優(yōu)化，但優(yōu)化程度一般有限。

需要設(shè)置過期時間或者定時更新數(shù)據(jù)，當(dāng)數(shù)據(jù)到達(dá)過期時間后從新從外部獲取，或者定時從外部撈取數(shù)據(jù)進(jìn)行更新，不能在外部數(shù)據(jù)發(fā)生變動時，及時更新到Flink程序中。

預(yù)加載和本地緩存難以應(yīng)對當(dāng)外部數(shù)據(jù)發(fā)生變化時，數(shù)據(jù)實時在Flink中保持更新。

類似于全局性共享的數(shù)據(jù)，詳見官方文檔

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

廣播變量創(chuàng)建后，它可以運行在集群中的任何function上，而不需要多次傳遞給集群節(jié)點，可以直接在內(nèi)存中拿數(shù)據(jù)，避免了大量的shuffle，導(dǎo)致集群性能下降。我們可以把一個dataset或者不變的緩存對象（例如maplist集合對象等）數(shù)據(jù)集廣播出去，然后不同的任務(wù)在節(jié)點上都能夠獲取到，并在每個節(jié)點上只會存在一份，而不是在每個并發(fā)線程中存在。

如果不使用broadcast，則在每個節(jié)點中的每個任務(wù)中都需要拷貝一份dataset數(shù)據(jù)集，比較浪費內(nèi)存(也就是一個節(jié)點中可能會存在多份dataset數(shù)據(jù))。廣播變量，可以借助下圖輔助理解。

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

將廣播的數(shù)據(jù)作為一個整體或?qū)ο髲V播，比如從MySQL中一次獲取全部數(shù)據(jù)，然后廣播出去，因為數(shù)據(jù)在MySql中，如果MySql中某條記錄發(fā)生變動，F(xiàn)link的souce是沒法知道，也不會廣播。所以只能在souce中定時從MySql中獲取全部數(shù)據(jù)，然后廣播更新。

示例數(shù)據(jù)格式：

kafka源流數(shù)據(jù)，只有itemid，沒有ip和port
{"host":"orcl", "itemid":"7875", "value":1}
{"host":"orcl2", "itemid":"7876", "value":2}
規(guī)則數(shù)據(jù)集在MySql，itemid關(guān)聯(lián)ip和port
itemid  ip              port
7875 192.168.199.105 1521
7876 192.168.199.106 1526

自定義MySql source, 定時 從Mysql獲取全部數(shù)據(jù)

@Override
public void run(SourceContext>> ctx) {
try {
 while (isRunning) {
  HashMap> output = new HashMap<>();
  ResultSet resultSet = preparedStatement.executeQuery();
  //每隔60s獲取全部外部數(shù)據(jù)集
  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);
 }
}

廣播代碼實現(xiàn)：

public void processElement(Map value, ReadOnlyContext ctx, Collector> out) throws Exception {
   //從廣播中獲取全量數(shù)據(jù)
  ReadOnlyBroadcastState>>      broadcastState = ctx.getBroadcastState(ruleStateDescriptor);
 //獲取全部規(guī)則數(shù)據(jù)進(jìn)行匹配

Map>  itemrules=  broadcastState.get(null);
  //規(guī)則數(shù)據(jù)集為空跳過
    if(itemrules==null) {
     return;
    }
    //事件流中的itemid
    Object itemidObj = value.get("itemid"); // value kafka流中數(shù)據(jù)獲取itemid
if (itemidObj == null) {
 return;
}
    Tuple2 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> value, Context ctx, Collector> out) throws Exception {
   //數(shù)據(jù)全部更新
BroadcastState>> broadcastState = ctx
  .getBroadcastState(ruleStateDescriptor);
//每次更新全部規(guī)則數(shù)據(jù)
broadcastState.put(null, value);
    System.out.println("規(guī)則全部更新成功，更新item規(guī)則：" + value);
}

執(zhí)行結(jié)果：

//itemid=7875關(guān)聯(lián)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 休眠結(jié)束后將會更新數(shù)據(jù)
規(guī)則更新成功，更新item規(guī)則：{7875=(192.168.199.105,1521), 7876=(192.168.199.106,1526)}
//itemid=7875關(guān)聯(lián)ip=192.168.199.105
{host=orcl, itemid=7875, value=1, ip=192.168.199.105, port=1521}

這種方式和預(yù)加載很像，都是通過定時任務(wù)加載全部數(shù)據(jù)，只不過是方法的位置不同，一個是在自定義source中設(shè)置休眠時間，另外一個是在算子的open方法中設(shè)置定時任務(wù)，廣播變量的方式同樣無法做到數(shù)據(jù)修改后實時更新。

當(dāng)數(shù)據(jù)來源于kafka時，F(xiàn)link消費kafka獲取流，將流數(shù)據(jù)存儲在廣播狀態(tài)中，稱之為廣播流，不同于廣播變量一次獲取全部數(shù)據(jù)，廣播流是kafka新增一條記錄就將這條記錄存儲到廣播中，那廣播流如何實現(xiàn)外部數(shù)據(jù)的新增和更新？

kafka源流數(shù)據(jù)，只有itemid，沒有ip和port
{"host":"orcl", "itemid":"7875", "value":1}
{"host":"orcl2", "itemid":"7876", "value":2}
規(guī)則數(shù)據(jù)集在kafka，itemid關(guān)聯(lián)ip和port
{"itemid":"7875","ip":"192.168.199.104","port":1521}
{"itemid":"7876","ip":"192.168.199.106","port":1526}

2.1 外部數(shù)據(jù)新增和修改記錄

// 廣播狀態(tài)底層結(jié)構(gòu)是Map結(jié)構(gòu)
//kafka中的數(shù)據(jù),flink消費后存儲到廣播狀態(tài),在廣播狀態(tài)中以itemid為key進(jìn)行存儲
{"itemid":"7875","ip":"192.168.199.104","port":1521}
{"itemid":"7876","ip":"192.168.199.106","port":1526}
//新增  往kafka寫入新記錄(key不相同),flink會持續(xù)消費kafka并將數(shù)據(jù)通過Map的put()方法存入廣播狀態(tài)
//修改  往kafka寫入新記錄(key相同)，put()方法覆蓋之前的這條記錄以達(dá)到更新的目的
//比如需要更新itemid的ip和port值,要求往kafka中寫入一條新數(shù)據(jù),比如更新itemid 7875的ip和port
{"itemid":"7875","ip":"192.168.199.105","port":1525}

2.2 刪除記錄

//kafka中的數(shù)據(jù),flink消費后存儲到廣播狀態(tài),以itemid為key進(jìn)行存儲
{"itemid":"7875","ip":"192.168.199.104","port":1521}
{"itemid":"7876","ip":"192.168.199.106","port":1526}
//如果需要刪除某條記錄，往kafka中寫入帶有key的數(shù)據(jù)和刪除標(biāo)記即可
//比如刪除itemid為7875的記錄,要求往kafka中寫入一條新數(shù)據(jù),程序刪除廣播中itemid7875的記錄
{"itemid":"7875","isRemove":true}

由于消費kafka流是實時的，kafka的新記錄會實時進(jìn)行消費，根據(jù)新記錄的內(nèi)容對廣播數(shù)據(jù)實時的進(jìn)行新增，修改或刪除

同時由于kafka中的數(shù)據(jù)是不可變的，當(dāng)程序需要重啟時，只需從頭消費kafka即可，由于具有冪等性，最終的廣播數(shù)據(jù)是不會變的。

示例代碼

//Flink消費外部kafka規(guī)則數(shù)據(jù)作為流
FlinkKafkaConsumer ruleKafkaConsumer = new FlinkKafkaConsumer("topic",new ItemRuleEntiyPojoSchema(),properties);
DataStream ruleStream = env.addSource(ruleKafkaConsumer);

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

執(zhí)行結(jié)果

//  所有廣播規(guī)則數(shù)據(jù):
7875={itemid=7875, ip=192.168.199.104, port=1521}
7876={itemid=7876, ip=192.168.199.106, port=1526}
//itemid=7875關(guān)聯(lián)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}
 獲取到新的廣播規(guī)則:{itemid=7875, ip=192.168.199.105, port=1521}
 //itemid=7875關(guān)聯(lián)ip=192.168.199.105
{host=orcl,itemid=7875, value=1, ip=192.168.199.105, port=1521},7875)

通過kafka廣播流的方式最終實現(xiàn)了Flink與外部數(shù)據(jù)交互的實時更新，不僅是kafka，還有MQ，甚至文件格式都可以作為廣播流，廣播流要求數(shù)據(jù)不能從內(nèi)部更改（無法作為流消息被實時消費),只能通過新增的方式進(jìn)行修改和刪除(新增記錄中key相同的表示覆蓋修改，帶key和刪除標(biāo)記的表示刪除)

相比表(mysql,oracle)只是結(jié)果的呈現(xiàn)，日志(kafka或其它隊列)是一種帶有時間維度（或先后順序）信息的存儲，可以說表是二維的，日志是三維的，通過日志可以復(fù)原每個時間點的表，但是表不能還原日志。

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

END