flink sideoutput

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

public class SideOutputExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 定义一个侧输出标签
        final OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

        DataStream<String> input = env.fromElements("apple", "banana", "cherry", "date");

        DataStream<String> mainOutput = input.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                if (value.startsWith("a")) {
                    // 将以 'a' 开头的元素发送到侧输出
                    ctx.output(outputTag, value);
                } else {
                    // 其他元素发送到主输出
                    out.collect(value);
                }
            }
        });

        // 获取侧输出流
        DataStream<String> sideOutput = mainOutput.getSideOutput(outputTag);

        // 打印主输出和侧输出
        mainOutput.print();
        sideOutput.print();

        env.execute("Side Output Example");
    }
}