2026/3/1 15:26:53
网站建设
项目流程
网站建设经验材料,阿里云nas做网站,做搜狗网站优化,做的视频发到哪个网站好本文重点
我们可以通过处理函数来在“主流”上分叉出的“支流”,也就是侧输出流,而侧输出流还可以和主流类型不一样。
API
OutputTagString outputTag = new OutputTagString(side-output) {};
DataStreamInteger stream = env.addSo…本文重点我们可以通过处理函数来在“主流”上分叉出的“支流”,也就是侧输出流,而侧输出流还可以和主流类型不一样。APIOutputTagString outputTag = new OutputTagString("side-output") {}; DataStreamInteger stream = env.addSource(...); SingleOutputStreamOperatorLong longStream = stream.process(new ProcessFunctionInteger, Long() { @Override public void processElement( Integer value, Context ctx, CollectorInteger 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。