Flink侧输出流及案例

ゝ一纸荒年。 2023-10-06 23:14 14阅读 0赞

flink可以起到分流的其实有很多种方式，比如

keyBy

filter

不过，这些流的数据类型也都相同

不如process function的side outputs功能可以产生多条流，并且这些流的数据类型可以不一样。使用起来更加灵活。

以下以案例的方式介绍

### 简单案例 ###

//数据源streamDate 是jsonObject样式的string
    //数据源中有个mn字段，有奇数偶数，案例以此作为分流
    object SideOutputTest {
        
      def main(args: Array[String]): Unit = {
        
          //省略环境和source
    val mnEven: OutputTag[String] = new OutputTag[String]("mn_even")
    
    streamDate
          .process(new EvenMN(mnEven))
          .getSideOutput(mnEven)
          .print()
     env.execute("SideOutput_job")
      }
    }
    
    //自定义方法，来判断mn是否为偶数
    class EvenMN(myOutputStream: OutputTag[String]) extends ProcessFunction[String, String] {
        
      override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {
        
        val js: JSONObject = JSON.parseObject(value)
        if (js.getInteger("mn") % 2 == 0) {
        
          ctx.output(myOutputStream, value)
        } else {
        
          out.collect(value)
        }
      }
    
    }

### 应用升级 ###

写完之后发现实用性不强，于是 我又改了需求

由于我的业务多个sink是糅合在一起的，因此我想把各种输出放在一个JSONArray里，然后在根据JSONArray里的JSONObject的key来分流到各个sink里。

//源数据是JSONArray，且随机有2或者3个JSONObject

筛选JSONObject中带有“zwh” key的JSONObject，

如此可以完成业务中多个数据在一起的问题

代码如下

object SideOutputTest {
        
      def main(args: Array[String]): Unit = {
        
          //省略环境和source
    val random: Random = new Random()
    
        val allDate: DataStream[JSONArray] = streamDate
          .map(JSON.parseObject(_))
          .map(date => {
        
    
            val jSONArray: JSONArray = new JSONArray()
            jSONArray.add(date)
            jSONArray.add(getJSON())
            if (random.nextInt(2)%2==0){
        
              jSONArray.add(getJSON2())
            }
            jSONArray
          })
    
    
        val zwhDate: OutputTag[String] = new OutputTag[String]("zwh")
        val noZWH: DataStream[String] = allDate.process(new ProcessFunction[JSONArray, String] {
        
          override def processElement(value: JSONArray, ctx: ProcessFunction[JSONArray, String]#Context, out: Collector[String]): Unit = {
        
            value.forEach(x => {
        
              val js: JSONObject = JSON.parseObject(x.toString)
              if (js.containsKey("zwh")) {
        
                out.collect(js.toString())
              } else {
        
                ctx.output(zwhDate, js.toString())
              }
            })
          }
        })
        //包含zwh字段的
        noZWH.print("是ZWH")
        //不包含zwh字段的
        noZWH.getSideOutput(zwhDate).print("非ZWH字段的")
    
      }
        
         def getJSON():JSONObject={
        
        val r: Random = new Random()
        val js: JSONObject = new JSONObject()
        js.put("testKey","lalalla")
        js.put("zwh","帅")
        js.put("num",r.nextInt(5))
        js
      }
    
      def getJSON2():JSONObject={
        
        val r: Random = new Random()
        val js: JSONObject = new JSONObject()
        js.put("kk","aaa")
        js.put("zwh","帅")
        js.put("num",r.nextInt(5))
        js
      }
        
    }