说明

在 flink1.12.x 中，推荐使用 DataStream ，即可实现 批处理，也可实现 流处理

官方文档：
https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/datastream_api.html

java代码

package top.malaoshi;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class TestDataStream {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置模式，默认为流模式
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);//批模式
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);//流模式
//        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        //数据源
        DataStream<String> ds = env.fromElements(
                "hadoop hdfs",
                "hadoop yarn",
                "hadoop flink",
                "flink yarn");

        DataStream<Tuple2<String, Integer>> dataStream = ds
                .flatMap(new Splitter())//拆分、并记为1，每个元素是 (单词,1)
                .keyBy(value -> value.f0)//流处理中的分组，根据第一个元素（即：单词）进行分组
                .sum(1);//按照第二个元素聚合

        dataStream.print();
        //启动并等待程序结束，可传入任务名
        env.execute("Window WordCount");
    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        /**
         *
         * @param sentence 表示每一行数据
         * @param out  返回的数据
         */
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}

设置运行模式

通过代码可以设置以下模式：

批处理模式
流处理模式
自动模式

提示：代码设置模式，可通过 flink命令 设置运行模式

设置批处理模式

取消下面注释：

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

注释掉下面：

//env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
//env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

执行结果如下：

7> (flink,2)
8> (hadoop,3)
2> (yarn,2)

与使用 DataSet 结果类似

设置流处理模式

取消下面注释：

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

注释掉下面：

//env.setRuntimeMode(RuntimeExecutionMode.BATCH);
//env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

执行结果如下：

5> (hdfs,1)
2> (yarn,1)
2> (yarn,2)
8> (hadoop,1)
8> (hadoop,2)
8> (hadoop,3)
7> (flink,1)
7> (flink,2)

读取一个数据，处理一个数据，所以会出现重复单词，且该单词的 数量是累加的

设置自动模式

取消下面注释：

env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

注释掉下面：

//env.setRuntimeMode(RuntimeExecutionMode.BATCH);
//env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

根据数据源自行判断，在 env.fromElements() 中，识别为 批处理

不设置模式

如果不设置模式，就认为是 流处理

原文出处：/show_1IX1z8zbwub0.html