（十二）Flink Datastream API 编程指南 侧输出流（side-output）

除了由DataStream操作产生的主流之外，您还可以产生任意数量的额外的端输出结果流。结果流中的数据类型不必与主流流中的数据类型匹配，不同端输出的类型也可以不同。当您想要分割数据流时，这个操作可能很有用，因为您通常必须复制数据流，然后从每个流中过滤出您不希望拥有的数据。

使用旁路输出时，首先需要定义用于标识旁路输出流的 OutputTag：

// 这需要是一个匿名的内部类，以便我们分析类型
OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

注意 OutputTag 是如何根据旁路输出流所包含的元素类型进行类型化的。

可以通过以下方法将数据发送到旁路输出：

• ProcessFunction
• KeyedProcessFunction
• CoProcessFunction
• KeyedCoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

你可以使用在上述方法中向用户暴露的 Context 参数，将数据发送到由 OutputTag 标识的旁路输出。这是从 ProcessFunction 发送数据到旁路输出的示例：

DataStream<Integer> input = ...;

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = input
  .process(new ProcessFunction<Integer, Integer>() {

      @Override
      public void processElement(
          Integer value,
          Context ctx,
          Collector<Integer> out) throws Exception {
        // 发送数据到主要的输出
        out.collect(value);

        // 发送数据到旁路输出
        ctx.output(outputTag, "sideout-" + String.valueOf(value));
      }
    });

你可以在 DataStream 运算结果上使用 getSideOutput(OutputTag) 方法获取旁路输出流。这将产生一个与旁路输出流结果类型一致的 DataStream：

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = ...;

DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);

一个demo

package com.stream.streaming.splitflow;

import com.happy.common.model.MetricEvent;
import com.happy.common.utils.KafkaConfigUtil;
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;

/**
 * @author happy
 * @Link https://github.com/zhisheng17/flink-learning/blob/master/flink-learning-examples/src/main/java/com/zhisheng/examples/streaming/sideoutput/SideOutputEvent.java
 * @create 2020-07-20 06:09
 */
public class SideOutput {
    private static final OutputTag<MetricEvent> machineTag      =   new OutputTag<MetricEvent>("machine");
    private static final OutputTag<MetricEvent> dockerTag       =   new OutputTag<MetricEvent>("docker");
    private static final OutputTag<MetricEvent> applicationTag  =   new OutputTag<MetricEvent>("application");

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

// 构建数据源
        DataStreamSource<MetricEvent> metricEventDataStreamSource = KafkaConfigUtil.buildSource(env);

        /**
         * ProcessFunction
         * KeyedProcessFunction
         * CoProcessFunction
         * ProcessWindowFunction
         * ProcessAllWindowFunction
         * 这里不只是ProcessFunction可以实现该sideOutput，上面的函数同样可以实现
         */
        SingleOutputStreamOperator<MetricEvent> sideOutputResult = metricEventDataStreamSource.process(new ProcessFunction<MetricEvent, MetricEvent>() {
            @Override
            public void processElement(MetricEvent metricEvent, Context context, Collector<MetricEvent> collector) throws Exception {
                String s = metricEvent.getTags().get("type");
                switch (s) {
                    case "machine":
                        context.output(machineTag, metricEvent);
                    case "docker":
                        context.output(dockerTag, metricEvent);
                    case "application":
                        context.output(applicationTag, metricEvent);
                    default:
                        collector.collect(metricEvent);
                }
            }
        });

        DataStream<MetricEvent> docker      = sideOutputResult.getSideOutput(dockerTag);
        DataStream<MetricEvent> application = sideOutputResult.getSideOutput(applicationTag);
        DataStream<MetricEvent> machine     = sideOutputResult.getSideOutput(machineTag);

        docker.print();
        application.print();
        machine.print();

        env.execute("SideOutput App start");
    }
}