胡杨河市网站建设_网站建设公司_悬停效果_seo优化

本文重点

我们可以通过处理函数来在“主流”上分叉出的“支流”，也就是侧输出流，而侧输出流还可以和主流类型不一样。

API

OutputTag<String> outputTag = new OutputTag<String>("side-output") {}; DataStream<Integer> stream = env.addSource(...); SingleOutputStreamOperator<Long> longStream = stream.process(new ProcessFunction<Integer, Long>() { @Override public void processElement( Integer value, Context ctx, Collector<Integer> out) throws Exception { // 转换成 Long，输出到主流中 out.collect(Long.valueOf(value)); // 转换成 String，输出到侧输出流中 ctx.output(outputTag, "side-output: " + String.valueOf(value)); } });

只要在处理函数的.processElement()或者.onTimer()方法中，调用上下文的.output()方法就可以了。output()方法需要传入两个参数：

第一个参数OutputTag，用来标识侧输出流的id和类型，如上代码所示，表示侧输出流的id是side-output，类型是String。