Apache Flink：SpiltStream与侧输出

Question

从这个问题中，我了解到Apache Flink中的SplitStream现在已被弃用，建议改用side输出。有人能举一个侧面输出如何替代splitStream的例子吗？

e、 g.如何修改下面的代码段以使用侧输出？

        DataStream mainDataStream = some definition
        SplitStream<some-type> splitStream = mainDataStream.select("some-string")

Answer 1

而不是

java prettyprint-override">SplitStream<Integer> split = someDataStream.split(new OutputSelector<Integer>() {

    @Override
    public Iterable<String> select(Integer value) {
        List<String> output = new ArrayList<String>();
        if (value % 2 == 0) {
            output.add("even");
        }
        else {
            output.add("odd");
        }
        return output;
    }
});

DataStream<Integer> evens = split.select("even");
DataStream<Integer> odds = split.select("odd");

你现在可以

final OutputTag<Integer> evenTag = new OutputTag<String>("even"){};
final OutputTag<Integer> oddTag = new OutputTag<String>("odd"){};

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

        @Override
        public void processElement(
          Integer value,
          Context ctx,
          Collector<Integer> out) throws Exception {

          if (value % 2 == 0) {
              ctx.output(evenTag, value);
          } else {
              ctx.output(oddTag, value);
          }
   });

DataStream<Integer> evens = mainDataStream.getSideOutput(evenTag);
DataStream<Integer> odds = mainDataStream.getSideOutput(oddTag);

请注意，与拆分流不同，侧输出流可以是不同类型的。