一，Flink快速上手

1.依赖配置

1.1 pom文件

<properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <flink.version>1.13.0</flink.version>
        <java.version>1.8</java.version>
    <!--需要设定scala版本因为flink也引用了scala的一些东西-->
        <scala.binary.version>2.12</scala.binary.version>
    
        <slf4j.version>1.7.30</slf4j.version>
    </properties>

    <dependencies>
        <!-- 引入 Flink 相关依赖-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- 引入日志管理相关依赖-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-to-slf4j</artifactId>
            <version>2.14.0</version>
        </dependency>
    </dependencies>

1.2 日志文件

log4j.rootLogger=error, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

2.编写代码

2.1 在根目录下创建数据

2.2 书写批处理执行代码

public static void main(String[] args) throws Exception {
    // 1. 创建执行环境
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    // 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
    DataSource<String> lineDs = env.readTextFile("input/word.txt");
    // 3. 转换数据格式
    FlatMapOperator<String, Tuple2<String, Long>> wordAndOne =
        lineDs.flatMap((String line, Collector<Tuple2<String, Long>> out) ->
                       {
                           String[] words = line.split(" ");
                           for (String word : words) {
                               out.collect(Tuple2.of(word, 1L));
                           }
                       });
    // 4.防止泛型擦除
    FlatMapOperator<String, Tuple2<String, Long>> returns =
        wordAndOne.returns(Types.TUPLE(Types.STRING, Types.LONG));
    // 5. 按照 word 进行分组
    UnsortedGrouping<Tuple2<String, Long>> wordAndOneUg = wordAndOne.groupBy(0);
    // 6. 分组内聚合统计
    AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUg.sum(1);
    // 7. 打印结果
    sum.print();
    // 结果
    // (flink,1)
    // (world,1)
    // (hello,3)
    // (java,1)
}

代码说明和注意事项：

① Flink 在执行应用程序前应该获取执行环境对象，也就是运行时上下文环境。

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

② Flink 同时提供了 Java 和 Scala 两种语言的 API，有些类在两套 API 中名称是一样的。所以在引入包时，如果有 Java 和 Scala 两种选择，要注意选用 Java 的包。

③ 直接调用执行环境的 readTextFile 方法，可以从文件中读取数据。

④ 我们的目标是将每个单词对应的个数统计出来，所以调用 flatmap 方法可以对一行文字进行分词转换。将文件中每一行文字拆分成单词后，要转换成(word,count)形式的二元组，初始 count 都为 1。returns 方法指定的返回数据类型 Tuple2，就是 Flink 自带的二元组数据类型。

⑤ 在分组时调用了 groupBy 方法，它不能使用分组选择器，只能采用位置索引或属性名称进行分组。

需要注意的是，这种代码的实现方式，是基于 DataSet API 的，也就是我们对数据的处理转换，是看作数据集来进行操作的。事实上 Flink 本身是流批统一的处理架构，批量的数据集本质上也是流，没有必要用两套不同的 API 来实现。所以从 Flink 1.12 开始，官方推荐的做法是直接使用 DataStream API，在提交任务时通过将执行模式设为 BATCH 来进行批处理：

$ bin/flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar

这样，DataSet API 就已经处于“软弃用”（soft deprecated）的状态，在实际应用中我们只要维护一套 DataStream API 就可以了。这里只是为了方便大家理解，我们依然用 DataSet API做了批处理的实现。

2.3 书写流处理执行代码(有界)

public static void main(String[] args) throws Exception {
    // 1. 创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 2. 读取文件
    DataStreamSource<String> lineDss = env.readTextFile("input/word.txt");
    // 3. 转换数据格式
    SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne =
        lineDss.flatMap((String line, Collector<String> out) ->
        {
            Arrays.stream(line.split(" ")).forEach(out::collect);
        }).returns(Types.STRING).map(word -> Tuple2.of(word, 1L)).returns(Types.TUPLE(Types.STRING, Types.LONG));
    // 4. 分组
    KeyedStream<Tuple2<String, Long>, String> wordAndOneKs = wordAndOne.keyBy(t -> t.f0);
    // 5. 求和
    SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKs.sum(1);
    // 6. 打印
    result.print();
    // 7. 执行
    env.execute();
}

① 主要观察与批处理程序 BatchWordCount 的不同：

② 创建执行环境的不同，流处理程序使用的是 StreamExecutionEnvironment。

③ 每一步处理转换之后，得到的数据对象类型不同。

④ 分组操作调用的是 keyBy 方法，可以传入一个匿名函数作为键选择器（KeySelector），指定当前分组的 key 是什么。

⑤ 代码末尾需要调用 env 的 execute 方法，开始执行任务。

• 输出结果

3> (java,1)
9> (world,1)
5> (hello,1)
5> (hello,2)
13> (flink,1)
5> (hello,3)

我们可以看到，这与批处理的结果是完全不同的。批处理针对每个单词，只会输出一个最终的统计个数；而在流处理的打印结果中，“hello”这个单词每出现一次，都会有一个频次统计数据输出。这就是流处理的特点，数据逐个处理，每来一条数据就会处理输出一次。我们通过打印结果，可以清晰地看到单词“hello”数量增长的过程。

看到这里大家可能又会有新的疑惑：我们读取文件，第一行应该是“hello flink”，怎么这里输出的第一个单词是“world”呢？每个输出的结果二元组，前面都有一个数字，这又是什么呢？

我们可以先做个简单的解释。Flink 是一个分布式处理引擎，所以我们的程序应该也是分布式运行的。在开发环境里，会通过多线程来模拟 Flink 集群运行。所以这里结果前的数字，其实就指示了本地执行的不同线程，对应着 Flink 运行时不同的并行资源。这样第一个乱序的问题也就解决了：既然是并行执行，不同线程的输出结果，自然也就无法保持输入的顺序了。另外需要说明，这里显示的编号为 1~13，是由于运行电脑的 CPU 的核心数来决定的，我自己的是16核的，所以默认模拟的并行线程有 16 个。这段代码不同的运行环境，得到的结果会是不同的。关于 Flink 程序并行执行的数量，可以通过设定“并行度”（Parallelism）来进行配置，我们会在后续详细讲解这些内容。