[ https://issues.apache.org/jira/browse/FLINK-18960?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
xiaohang.li updated FLINK-18960: -------------------------------- Affects Version/s: (was: 1.11.1) (was: 1.11.0) > flink sideoutput union > ---------------------- > > Key: FLINK-18960 > URL: https://issues.apache.org/jira/browse/FLINK-18960 > Project: Flink > Issue Type: Bug > Components: API / DataStream > Affects Versions: 1.10.1 > Reporter: xiaohang.li > Priority: Minor > > flink sideoutput > union操作时数据出现问题。从主流分出来的侧输出流进行union操作时,显示输出的是以最后一个union的数据流结果*union的次数 > > val side = new OutputTag[String]("side") > val side2 = new OutputTag[String]("side2") > val side3 = new OutputTag[String]("side3") > val ds = env.socketTextStream("master",9001) > val res = ds.process(new ProcessFunction[String,String] { > override def processElement(value: String, ctx: ProcessFunction[String, > String]#Context, out: Collector[String]): Unit = { > if(value.contains("hello")) > { ctx.output(side,value) } > else if(value.contains("world")) > { ctx.output(side2,value) } > else if(value.contains("flink")) > { ctx.output(side3,value) } > out.collect(value) > } > }) > val res1 = res.getSideOutput(side) > val res2 = res.getSideOutput(side2) > val res3 = res.getSideOutput(side3) > println( "====>"+res1.getClass) > println( "====>"+res2.getClass) > res1.print("res1") > res2.print("res2") > res3.print("res3") > res2.union(res1).union(res3).print("all") > > > > 在socket端口分别输入 > hello > world > flink > > idea显示数据如下 > res1> hello > res2> world > res3> flink > all> flink > all> flink > all> flink > > 可见在all输出流显示的是最后一个union的侧输出流*union的次数,实际显示应为 > all>flink > 如果分别再对每个侧输出流进行map操作后再执行上面的代码即显示正确的数据输出,但是这一步操作应该是非必须的 -- This message was sent by Atlassian Jira (v8.3.4#803005)