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
}
}
这篇好文章是转载于:学新通技术网
- 版权申明: 本站部分内容来自互联网,仅供学习及演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,请提供相关证据及您的身份证明,我们将在收到邮件后48小时内删除。
- 本站站名: 学新通技术网
- 本文地址: /boutique/detail/tanhgchegf
系列文章
更多
同类精品
更多
-
photoshop保存的图片太大微信发不了怎么办
PHP中文网 06-15 -
Android 11 保存文件到外部存储,并分享文件
Luke 10-12 -
《学习通》视频自动暂停处理方法
HelloWorld317 07-05 -
word里面弄一个表格后上面的标题会跑到下面怎么办
PHP中文网 06-20 -
photoshop扩展功能面板显示灰色怎么办
PHP中文网 06-14 -
微信公众号没有声音提示怎么办
PHP中文网 03-31 -
excel下划线不显示怎么办
PHP中文网 06-23 -
excel打印预览压线压字怎么办
PHP中文网 06-22 -
怎样阻止微信小程序自动打开
PHP中文网 06-13 -
TikTok加速器哪个好免费的TK加速器推荐
TK小达人 10-01