Flink常用算子

map
DataStream → DataStreamflatMap
DataStream → DataStreamfliter
DataStream → DataStreamkeyBy
DataStream → KeyedStream
对数据进行分流reduce
KeyedStream/WindowedStream/AllWindowedStream → DataStream
用于keyBy或者window/windowAll之后window
KeyedStream → WindowedStream
用于keyBy之后windowAll
DataStream → AllWindowedStream
不用于keyBy之后，此算子并行度始终为1apply
WindowedStream/AllWindowedStream → DataStreamunion
DataStream* → DataStream
合并相同类型的流join
DataStream,DataStream → DataStream
比较两条流中的元素，如果相等输出，否则不进行输出。

dataStream.join(otherStream)
    .where().equalTo()
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new JoinFunction () {...});

Interval Join
KeyedStream,KeyedStream → DataStream

// this will join the two streams so that
// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2
keyedStream.intervalJoin(otherKeyedStream)
    .between(Time.milliseconds(-2), Time.milliseconds(2)) // lower and upper bound
    .upperBoundExclusive(true) // optional
    .lowerBoundExclusive(true) // optional
    .process(new IntervalJoinFunction() {...});

CoGroup
DataStream,DataStream → DataStream
比较两条流中的元素，如果相等则放在一起输出，否则分开输出。重点是group。

dataStream.coGroup(otherStream)
    .where(0).equalTo(1)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new CoGroupFunction () {...});

Connect
DataStream,DataStream → ConnectedStream
“连接”两条数据流，并保留他们的类型(类型可以不一样)。连接允许两个流之间共享状态。

DataStream someStream = //...
DataStream otherStream = //...

ConnectedStreams connectedStreams = someStream.connect(otherStream);

CoMap, CoFlatMap
ConnectedStream → DataStream
专门针对ConnectedStream流的算子

connectedStreams.map(new CoMapFunction() {
    @Override
    public Boolean map1(Integer value) {
        return true;
    }

    @Override
    public Boolean map2(String value) {
        return false;
    }
});
connectedStreams.flatMap(new CoFlatMapFunction() {

   @Override
   public void flatMap1(Integer value, Collector out) {
       out.collect(value.toString());
   }

   @Override
   public void flatMap2(String value, Collector out) {
       for (String word: value.split(" ")) {
         out.collect(word);
       }
   }
});

Iterate
DataStream → IterativeStream → ConnectedStream
一个流被分为两部分，一部分持续不断循环输出，另一部分正常输出。

IterativeStream iteration = initialStream.iterate();
DataStream iterationBody = iteration.map ();
DataStream feedback = iterationBody.filter(new FilterFunction(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value > 0;
    }
});
iteration.closeWith(feedback);
DataStream output = iterationBody.filter(new FilterFunction(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value <= 0;
    }
});