官方文檔-關(guān)于分流

最好大家還是看官方文檔，我只是當(dāng)一個搬運工

一、背景

在一些業(yè)務(wù)場景中，一個流中可能有多種類型的數(shù)據(jù)，比如訂單：有線上訂單，有線下訂單。當(dāng)需要將不同類型的數(shù)據(jù)進(jìn)行分別處理，比如 寫入到不同的數(shù)據(jù)表或者join 不同的其他流時，這個時候使用分流就比較合適。

二、官方常用的幾種方法

• ProcessFunction
• KeyedProcessFunction
• CoProcessFunction
• KeyedCoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

三、示范

本文只詳細(xì)介紹最常用 process 分流 ，道理都是相通的

直接上代碼偽碼，大家主要要理解，而不是直接復(fù)制代碼

//這是訂單source，最原始的流
val orderSource =  這是你構(gòu)建source 的方法

  //創(chuàng)建線上訂單 tag
    val onlineOrderTag = new OutputTag[JSONObject]("onlineOrder")
   //創(chuàng)建線下訂單 tag
    val offlineOrderTag = new OutputTag[JSONObject]("offlineOrder")

   // 這個sideOutStream 就是分流之后的流對象
   val sideOutStream = orderSource
      .filter(new PaymentFilter)  // 這里是一個過濾邏輯，如果你沒有可以不過濾
       // 這個process 就是分流的操作了
      .process(new ProcessFunction[String, JSONObject] {
        override def processElement(orderString: String, ctx: ProcessFunction[String, JSONObject]#Context, out: Collector[JSONObject]): Unit = {
          val outOrder = JSON.parseObject(orderString)
        
          //通過收銀員信息判斷是否屬于線下訂單
          if (!outOrder.containsKey("cashier_id") || StringUtils.isBlank(outOrder.getString("cashier_id"))) {
            ctx.output(onlineOrderTag, outOrder)
          } else {
            ctx.output(offlineOrderTag, outOrder)
          }
        }
      }
      )


val onlineStream = sideOutStream.getSideOutput(onlineOrderTag)

val offlineStream = sideOutStream.getSideOutput(offlineOrderTag)

// 流已經(jīng)分好了，后面是sink 還是 去干其他的，就看你的業(yè)務(wù)邏輯了
onlineStream.addSink()

offlineStream.addSink()

生產(chǎn)實踐

下圖是真實生產(chǎn)的一個DAG圖

內(nèi)部使用了分流， join ，自定義剔除器 等滿足業(yè)務(wù)需求

后面會更新 join 和 自定義剔除器 trigger 等 實戰(zhàn)場景，感興趣的朋友可以加個關(guān)注喲

實際運用DAG