Flink 侧流输出完整示例

对数据流进行标记，在下游通过 getSideOutput 取已经标记过的数据流处理。

package com.yuchaoshui.flink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class StreamingWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        
        OutputTag<String> oddTag = new OutputTag<String>("odd"){};
        OutputTag<String> evenTag = new OutputTag<String>("even"){};
        OutputTag<String> notNumberTag = new OutputTag<String>("notNumber"){};

        SingleOutputStreamOperator<String> mainStream = lines.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    int i = Integer.parseInt(value);
                    if (i % 2 == 0) {
                        ctx.output(evenTag, value);
                    } else {
                        ctx.output(oddTag, value);
                    }
                } catch (NumberFormatException e) {
                    ctx.output(notNumberTag, value);
                }

                // 主流也可以输出数据
                out.collect(value);
            }
        });
        DataStream<String> streamOdd = mainStream.getSideOutput(oddTag);
        DataStream<String> streamEven = mainStream.getSideOutput(evenTag);
        DataStream<String> streamNotNumber = mainStream.getSideOutput(notNumberTag);

        streamOdd.print("streamOdd");
        streamEven.print("streamEven");
        streamNotNumber.print("streamNotNumber");
        
        mainStream.print("mainStream");

        env.execute();
    }
}