Flink侧输出流及案例
Posted Z-hhhhh
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Flink侧输出流及案例相关的知识,希望对你有一定的参考价值。
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
以上是关于Flink侧输出流及案例的主要内容,如果未能解决你的问题,请参考以下文章