Flink使用(一) Streaming API处理有界流

目前，Flink版本支持，批流处理使用一套API完成。

即，使用DataStream API既能处理流数据，又能处理批数据（有界流）。

方式一：在程序中指定运行模式（不推荐，调试可以）：

        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

方式二：

提交任务时，使用bin/flink/run -Dexecution.runtime-mode=BATCH 指定BATCH方式运行程序。

我理解是将最终的结果进行输出，而不是获取一条记录后就输出一条结果。

例子数据源word.txt,只有三行数据

hello word
hello flink
hello java

package org.galaxy.foundation.core.courses;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

 
public class Stream10 {

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

        DataStreamSource lineDataStreamSource = env.readTextFile("/tmp/foundation-core/words.txt");
       SingleOutputStreamOperator> wordAndOneTuple=  lineDataStreamSource.flatMap((String line, Collector> out) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING,Types.LONG));


          KeyedStream, String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0);

        SingleOutputStreamOperator> sum = wordAndOneKeyedStream.sum(1);
        sum.print();
        env.execute();
    }

}

打印结果：

3> (hello,1)
2> (java,1)
7> (flink,1)
6> (word,1)
3> (hello,2)
3> (hello,3)

可以看出，每来一条记录，即产生一条输出。

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

重新运行，打印结果：

6> (word,1)
2> (java,1)
7> (flink,1)
3> (hello,3)

可以看出，将最终的汇总结果，一次完整的打印出来。