[大数据]Flink随笔

一、Flink简介

1.1 初识Flink

1.1.1 序言

? Flink起源于Stratosphere项目，Stratosphere是在2010~2014年由3所地处柏林的大学和欧洲的一些其他的大学共同进行的研究项目，2014年4月Stratosphere的代码被复制并捐赠给了Apache软件基金会，参加这个孵化项目的初始成员是Stratosphere系统的核心开发人员，2014年12月，Flink一跃成为Apache软件基金会的顶级项目。
? 在德语中，Flink一词表示快速和灵巧，项目采用一只松鼠的彩色图案作为logo，这不仅是因为松鼠具有快速和灵巧的特点，还因为柏林的松鼠有一种迷人的红棕色，而Flink的松鼠logo拥有可爱的尾巴，尾巴的颜色与Apache软件基金会的logo颜色相呼应，也就是说，这是一只Apache风格的松鼠。

Flink项目的理念是：“Apache Flink是为分布式、高性能、随时可用以及准确的流处理应用程序打造的开源流处理框架”。
Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。

重点去理解：

① ：有界和无界（当前的计算是流处理还是批处理）

② ：有状态（当前的设计是否被操作，状态保存到累机器中）

1.1.2 计算引擎的发展史

计算引擎迄今为止，一共经历了4代 ：

1.2 Flink的重要特点

1.2.1 事件驱动型(Event-driven)

事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。比较典型的就是以kafka为代表的消息队列几乎都是事件驱动型应用。(Flink的计算也是事件驱动型)

• 与之不同的就是SparkStreaming微批次，如图:

• 事件驱动型：

1.2.2 流与批的世界观

批处理的特点是有界、持久、大量，非常适合需要访问全套记录才能完成的计算工作，一般用于离线统计。
流处理的特点是无界、实时, 无需针对整个数据集执行操作，而是对通过系统传输的每个数据项执行操作，一般用于实时统计。
在spark的世界观中，一切都是由批次组成的，离线数据是一个大批次，而实时数据是由一个一个无限的小批次组成的。
而在flink的世界观中，一切都是由流组成的，离线数据是有界限的流，实时数据是一个没有界限的流，这就是所谓的有界流和无界流。

1. 无界数据流：
   无界数据流有一个开始但是没有结束，它们不会在生成时终止并提供数据，必须连续处理无界流，也就是说必须在获取后立即处理event。对于无界数据流我们无法等待所有数据都到达，因为输入是无界的，并且在任何时间点都不会完成。处理无界数据通常要求以特定顺序（例如事件发生的顺序）获取event，以便能够推断结果完整性。
2. 有界数据流：
   有界数据流有明确定义的开始和结束，可以在执行任何计算之前通过获取所有数据来处理有界流，处理有界流不需要有序获取，因为可以始终对有界数据集进行排序，有界流的处理也称为批处理。

1.2.3 分层API

? 1 : 最底层级的抽象仅仅提供了有状态流，它将通过过程函数（Process Function）被嵌入到DataStream API中。底层过程函数（Process Function） 与 DataStream API 相集成，使其可以对某些特定的操作进行底层的抽象，它允许用户可以自由地处理来自一个或多个数据流的事件，并使用一致的容错的状态。除此之外，用户可以注册事件时间并处理时间回调，从而使程序可以处理复杂的计算。
? 2 : 实际上，大多数应用并不需要上述的底层抽象，而是针对核心API（Core APIs） 进行编程，比如DataStream API（有界或无界流数据）以及DataSet API（有界数据集）。这些API为数据处理提供了通用的构建模块，比如由用户定义的多种形式的转换（transformations），连接（joins），聚合（aggregations），窗口操作（windows）等等。DataSet API 为有界数据集提供了额外的支持，例如循环与迭代。这些API处理的数据类型以类（classes）的形式由各自的编程语言所表示。
? 3 : Table API 是以表为中心的声明式编程，其中表可能会动态变化（在表达流数据时）。Table API遵循（扩展的）关系模型：表有二维数据结构（schema）（类似于关系数据库中的表），同时API提供可比较的操作，例如select、project、join、group-by、aggregate等。Table API程序声明式地定义了什么逻辑操作应该执行，而不是准确地确定这些操作代码的看上去如何。
? 4 : 尽管Table API可以通过多种类型的用户自定义函数（UDF）进行扩展，其仍不如核心API更具表达能力，但是使用起来却更加简洁（代码量更少)。除此之外，Table API程序在执行之前会经过内置优化器进行优化。
你可以在表与 DataStream/DataSet 之间无缝切换，以允许程序将 Table API与DataStream 以及 DataSet 混合使用。
? 5 : Flink提供的最高层级的抽象是 SQL 。这一层抽象在语法与表达能力上与 Table API 类似，但是是以SQL查询表达式的形式表现程序。SQL抽象与Table API交互密切，同时SQL查询可以直接在Table API定义的表上执行。
? 目前Flink作为批处理还不是主流，不如Spark成熟，所以DataSet使用的并不是很多。Flink Table API和Flink SQL也并不完善，大多都由各大厂商自己定制。所以我们主要学习DataStream API的使用。实际上Flink作为最接近Google DataFlow模型的实现，是流批统一的观点，所以基本上使用DataStream就可以了。
? 2020年12月8日发布的最新版本1.12.0, 已经完成实现了真正的**流批一体. **写好的一套代码, 即可以处理流失数据, 也可以处理离线数据. 这个与前面版本的处理有界流的方式是不一样的, Flink专门对批处理数据做了优化处理。

1.3 Spark or Flink

Flink和Spark的技术选择上的主要出发点就是：计算模型的选择！！！！！！！！！！！！！！！！！

**1 ：**Spark 和 Flink 一开始都拥有着同一个梦想，他们都希望能够用同一个技术把流处理和批处理统一起来，但他们走了完全不一样的两条路前者是以批处理的技术为根本，并尝试在批处理之上支持流计算；后者则认为流计算技术是最基本的，在流计算的基础之上支持批处理。正因为这种架构上的不同，今后二者在能做的事情上会有一些细微的区别。比如在低延迟场景，Spark 基于微批处理的方式需要同步会有额外开销，因此无法在延迟上做到极致。在大数据处理的低延迟场景，Flink 已经有非常大的优势。

Spark无法实现的方面体现在以下方面：

• 保证数据精准一次性处理（Exactly-once）【需要开发者自己开发程序保证】
• 乱序数据，迟到数据
• 低延迟、高吞吐、准确性
• 容错性

2 ：Spark和Flink的主要差别就在于计算模型不同。Spark采用了微批处理模型，而Flink采用了基于操作符的连续流模型。因此，对Apache Spark和Apache Flink的选择实际上变成了计算模型的选择，而这种选择需要在延迟、吞吐量和可靠性等多个方面进行权衡。

**3 ：如果企业中非要技术选型从Spark和Flink这两个主流框架中选择一个来进行流数据处理，我们推荐使用Flink，**主（显而）要（易见）的原因为：

• Flink灵活的窗口

• Exactly Once

  语义保证这两个原因可以大大的解放程序员, 加快编程效率, 把本来需要程序员花大力气手动完成的工作交给框架完成，棒棒哒，点赞！！！

1.3.1 两者的区别总结：

1. 处理方式方面：

   • Flink : 持续流方式
   • Spark : 批处理（微批次方式）

2. 延迟效果：

   • Flink : 低延迟（实时）
   • Spark : 高延迟（离线）

1.3.2 常见问题：

1. Spark的计算速度为什么比MR快?

   ①：基于内存进行计算操作

   ②：提供DAG（有向无环图）

2. Flink的计算速度为什么比Spark快？

   ?

二、Flink快速上手

WordCount的基本流程：

1. 获取执行环境

2. 获取数据源

3. 处理数据 —> 转换算子

• 扁平化操作：切分，转换为二元组flatMap()，实现FlatMapFunction类。

• 进行分组（参数：下标（word,1）【下标为“0”也就是根据word进行分组】）

  (word,1)|(word,1)|(word,1)|(word,1) ===> 为一组

  • 在批处理中分组的 API 是groupBy()方法
  • 在流处理中分组的 API 是keyBy()方法

• 按照分组进行聚合操作 （参数：下标(word,1)【下标为“1”也就是对1进行聚合累加操作】）===> (word,4)

4. 输出、保存…

5. 启动：执行处理程序（批处理不需要执行这一步）execute()方法

2.1 创建maven项目

• POM文件中添加需要的依赖:

<!--properties标签的作用：统一管理jar包的版本号-->
<properties>
    <flink.version>1.12.0</flink.version>
    <java.version>1.8</java.version>
    <scala.binary.version>2.11</scala.binary.version>
    <slf4j.version>1.7.30</slf4j.version>
</properties>
<!-- 引入jar包 -->
<dependencies>
    <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.apache.flink</groupId>
        <artifactId>flink-runtime-web_${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>
<!-- 引入打包插件 -->
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.3.0</version>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

• src/main/resources添加文件:log4j.properties

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

2.2 批处理WordCount

准备数据：

hello java
hello scala
hello spark
hello flink

代码示例：

package com.heather.chapter02;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class Batch_WordCount {

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

        // 1. 获取执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        DataSource<String> fileDS = env.readTextFile("input/hello.txt");
        // 3. 处理数据 - 操作算子
        // 3.1 扁平化操作：切分、转换为元组 ===> （word,1）
        FlatMapOperator<String, Tuple2<String, Integer>> wordAndOneDS = fileDS.flatMap(new MyFlatMapFunction());
        // 3.2 进行分组（参数：下标（word,1）【下标为“0”也就是根据word进行分组】）
        // (word,1)|(word,1)|(word,1)|(word,1) ===> 为一组
        // 在流处理中分组的 API groupBy()方法
        UnsortedGrouping<Tuple2<String, Integer>> wordToGroupDS = wordAndOneDS.groupBy(0);
        // 3.3 按照分组进行聚合操作 （参数：下标(word,1)【下标为“1”也就是对1进行聚合累加操作】）
        // ===> (word,4)
        AggregateOperator<Tuple2<String, Integer>> result = wordToGroupDS.sum(1);
        // 4. 输出、保存 ....
        result.print();
        // 5. 启动：执行处理程序（批处理不需要执行这一步）
    }

    /*
        自定义一个类实现FlatMapFunction接口，接口中有两个参数：
            参数1：输入的参数的类型：—>hello     ==> String类型
            参数2：输出的参数的类型：—>(hello,1) ==> 二元祖的类型
     */
    public static class MyFlatMapFunction implements FlatMapFunction<String, Tuple2<String,Integer>> {
        /*
            flatMap()方法：
                1. 切分数据
                2. 把数据封装为元组
                3. 使用采集器向下游发送数据
         */
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            // 1. 切分数据
            String[] wordArr = value.split(" ");
            for (String word : wordArr) {
                // 2. 把数据封装为元组
                /*
                    创建二元组有两种方式：
                        1. Tuple2<T,E> tuple = new Tuple2<>(value1,value2);
                        2. Tuple2<T,E> tuple = Tuple2.of(value1,value2);
                 */
                Tuple2<String, Integer> tuple = new Tuple2<>(word, 1);
                // 3. 使用采集器向下游发送数据
                out.collect(tuple);
            }
        }
    }
}

2.3 流处理WordCount

有界流和无界流的区别：

• 有界流在读取完数据后，程序自行停止；

• 无界流的程序一单启动后是不会停止的，除非数据源停止输入数据。

2.3.1 有界流

第一种方式：原生方式

public class Bound_WordCount {

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

        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        DataStreamSource<String> fileDS = env.readTextFile("input/hello.txt");
        // 3. 处理数据 —> 转换算子
        // 3.1 扁平化操作：切分，转换为二元组
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = fileDS.flatMap(new MyFlatMapFunction());
        // 3.2 进行分组（参数：下标（word,1）【下标为“0”也就是根据word进行分组】）
        // (word,1)|(word,1)|(word,1)|(word,1) ===> 为一组
        // 在流处理中分组的 API 是keyBy()方法
        KeyedStream<Tuple2<String, Integer>, Tuple> wordToGroupDS = wordAndOneDS.keyBy(0);
        // 3.3 按照分组进行聚合操作 （参数：下标(word,1)【下标为“1”也就是对1进行聚合累加操作】）
        // ===> (word,4)
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordToGroupDS.sum(1);
        // 4. 输出、保存...
        result.print();
        // 5. 启动：执行处理程序
        env.execute();

    }

    public static class MyFlatMapFunction implements FlatMapFunction<String, Tuple2<String,Integer>>{
        /*
            flatMap()方法的作用：
                1. 切分数据
                2. 把数据转换为二元组
                3. 使用采集器向下游发送数据

        * */
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            //1. 切分数据
            String[] wordArr = value.split(" ");
            for (String word : wordArr) {
                //2. 把数据转换为二元组
                /*
                    创建二元组有两种方式：
                        1. Tuple2<T,E> tuple = new Tuple2<>(value1,value2);
                        2. Tuple2<T,E> tuple = Tuple2.of(value1,value2);
                 */
                Tuple2<String, Integer> tuple = Tuple2.of(word, 1);
                //3. 使用采集器向下游发送数据
                out.collect(tuple);
            }
        }
    }
}

第二种方式：Lambda表达式

当Lambda表达式使用 java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息

public class Lambda_Bound_WordCount {

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

        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 获取数据源
        DataStreamSource<String> fileDS = env.readTextFile("input/hello.txt");
        // 3. 处理数据 —> 转换算子
        // 3.1 扁平化操作：切分，转换为二元组
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = fileDS.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
            //1. 切分数据
            String[] wordArr = value.split(" ");
            for (String word : wordArr) {
                //2. 把数据转换为二元组
            /*
                创建二元组有两种方式：
                    1. Tuple2<T,E> tuple = new Tuple2<>(value1,value2);
                    2. Tuple2<T,E> tuple = Tuple2.of(value1,value2);
             */
                Tuple2<String, Integer> tuple = Tuple2.of(word, 1);
                //3. 使用采集器向下游发送数据
                out.collect(tuple);
            }
        })       
                //当Lambda表达式使用 java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息，有以下两种解决方案
                //.returns(new TypeHint<Tuple2<String, Integer>>() {});
                .returns(Types.TUPLE(Types.STRING, Types.LONG));
        // 3.2 进行分组（参数：下标（word,1）【下标为“0”也就是根据word进行分组】）
        // (word,1)|(word,1)|(word,1)|(word,1) ===> 为一组
        // 在流处理中分组的 API 是keyBy()方法
        KeyedStream<Tuple2<String, Integer>, Tuple> wordToGroupDS = wordAndOneDS.keyBy(0);
        // 3.3 按照分组进行聚合操作 （参数：下标(word,1)【下标为“1”也就是对1进行聚合累加操作】）
        // ===> (word,4)
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordToGroupDS.sum(1);
        // 4. 输出、保存...
        result.print();
        // 5. 启动：执行处理程序
        env.execute();
    }
}

2.3.2 无界流

public class UnBound_WordCount {

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

        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 获取数据源
        DataStreamSource<String> socketDS = env.socketTextStream("localhost", 9999);

        // 3. 处理数据
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = socketDS.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (in, out) -> {
            String[] wordArr = in.split(" ");
            for (String word : wordArr) {
                Tuple2<String, Integer> tuple = Tuple2.of(word, 1);
                out.collect(tuple);
            }
        }).returns(new TypeHint<Tuple2<String, Integer>>() {
        });

        KeyedStream<Tuple2<String, Integer>, Tuple> wordToGroupDS = wordAndOneDS.keyBy(0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordToGroupDS.sum(1);

        // 4. 输出、保存
        result.print();
        // 5. 启动程序
        env.execute();

    }
}

三、Flink部署

3.1 开发模式

Flink中的Local-cluster(本地集群)模式,主要用于测试, 学习。

3.2 local-cluster模式

Flink中的Local-cluster(本地集群)模式,主要用于测试, 学习。

3.2.1 local-cluster模式配置

local-cluster模式基本属于零配置。

1. 上传Flink的安装包flink-1.12.0-bin-scala_2.11.tgz到hadoop102

2. 解压

   tar -zxvf flink-1.12.0-bin-scala_2.11.tgz -C /opt/module

3. 进入目录/opt/module, 复制flink-local

   cd /opt/module
cp -r flink-1.12.0 flink-local

3.2.2 在local-cluster模式下运行无界的WordCount

1. 打包idea中的应用

2. 把不带依赖的jar包上传到目录/opt/module/flink-local下

3. 启动本地集群

   bin/start-cluster.sh

4. 在hadoop102中启动netcat

   nc -lk 9999
   注意: 如果没有安装netcat需要先安装:
   sudo yum install -y nc

5. 命令行提交Flink应用

   bin/flink run -m hadoop102:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar

6. 在浏览器中查看应用执行情况

   http://hadoop102:8081
   

7. 也可以在log日志查看执行结果

   cat flink-atguigu-taskexecutor-0-hadoop102.out
   

8. 也可以在WEB UI提交应用

3.3 Standalone模式

Standalone模式又叫独立集群模式。

3.3.1 Standalone模式配置

1. 复制flink-standalone

   cp -r flink-1.12.0 flink-standalone

2. 修改配置文件:flink-conf.yaml

   jobmanager.rpc.address: hadoop102

3. 修改配置文件:workers

   hadoop102 hadoop103 hadoop104

4. 分发flink-standalone到其他节点

3.3.2 Standalone模式运行无界流WorkCount

1. 启动standalone集群

   bin/start-cluster.sh

2. 命令行提交Flink应用

   bin/flink run -m hadoop102:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar

3. 查看执行情况和本地集群一致.

4. 也支持Web UI界面提交Flink应用

3.3.3 Standalone高可用(HA)

任何时候都有一个 主 JobManager 和多个备用 JobManagers，以便在主节点失败时有备用 JobManagers 来接管集群。这保证了没有单点故障，一旦备 JobManager 接管集群，作业就可以正常运行。主备 JobManager 实例之间没有明显的区别。每个 JobManager 都可以充当主备节点。

1. 修改配置文件: flink-conf.yaml

high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop102:8020/flink/standalone/ha
high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
high-availability.zookeeper.path.root: /flink-standalone
high-availability.cluster-id: /cluster_atguigu

1. 修改配置文件: masters

   hadoop102:8081 hadoop103:8081

2. 分发修改的后配置文件到其他节点

3. 在/etc/profile.d/my.sh中配置环境变量

   export HADOOP_CLASSPATH=hadoop classpath
   **注意: **

• 需要提前保证HAOOP_HOME环境变量配置成功
• 分发到其他节点

1. 首先启动dfs集群和zookeeper集群

2. 启动standalone HA集群

   bin/start-cluster.sh

3. 可以分别访问

   http://hadoop102:8081 http://hadoop103:8081

4. 在zkCli.sh中查看谁是leader

   get /flink-standalone/cluster_atguigu/leader/rest_server_lock

杀死hadoop102上的Jobmanager, 再看leader
   
   **注意: **不管是不是leader从WEB UI上看不到区别, 并且都可以与之提交应用.

3.4 Yarn模式

独立部署（Standalone）模式由Flink自身提供计算资源，无需其他框架提供资源，这种方式降低了和其他第三方资源框架的耦合性，独立性非常强。但是你也要记住，Flink主要是计算框架，而不是资源调度框架，所以本身提供的资源调度并不是它的强项，所以还是和其他专业的资源调度框架集成更靠谱，所以接下来我们来学习在强大的Yarn环境中Flink是如何使用的。（其实是因为在国内工作中，Yarn使用的非常多）
把Flink应用提交给Yarn的ResourceManager, Yarn的ResourceManager会申请容器从Yarn的NodeManager上面. Flink会创建JobManager和TaskManager在这些容器上.Flink会根据运行在JobManger上的job的需要的slot的数量动态的分配TaskManager资源

3.4.1 Yarn模式配置

1. 复制flink-yarn

   cp -r flink-1.12.0 flink-yarn

2. 配置环境变量HADOOP_CLASSPATH, 如果前面已经配置可以忽略。

   在/etc/profile.d/my.sh中配置
   export HADOOP_CLASSPATH=hadoop classpath

3.4.2 Yarn运行无界流WordCount

1. 启动hadoop集群(hdfs, yarn)

2. 运行无界流

   bin/flink run -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar

3. 在yarn的ResourceManager界面查看执行情况

3.4.3 Flink on Yarn的3种部署模式

Flink提供了yarn上运行的3模式，分别为**Session-Cluster，Application Mode**和**Per-Job-Cluster**模式。

3.4.3.1 Session-Cluster

Session-Cluster模式需要先启动Flink集群，向Yarn申请资源, 资源申请到以后,永远保持不变。以后提交任务都向这里提交。这个Flink集群会常驻在yarn集群中，除非手工停止。
在向Flink集群提交Job的时候, 如果资源被用完了,则新的Job不能正常提交.
缺点: 如果提交的作业中有长时间执行的大作业, 占用了该Flink集群的所有资源, 则后续无法提交新的job.
所以, Session-Cluster适合那些需要频繁提交的多个小Job, 并且执行时间都不长的Job.

3.4.3.2 Per-Job-Cluster

一个Job会对应一个Flink集群，每提交一个作业会根据自身的情况，都会单独向yarn申请资源，直到作业执行完成，一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager，按需接受资源申请；适合规模大长时间运行的作业。
每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成之后创建的集群也会消失。

3.4.3.3 Application Mode

Application Mode会在Yarn上启动集群, 应用jar包的main函数(用户类的main函数)将会在JobManager上执行. 只要应用程序执行结束, Flink集群会马上被关闭. 也可以手动停止集群.
与Per-Job-Cluster的区别: 就是Application Mode下, 用户的main函数式在集群中执行的
官方建议:
出于生产的需求, 我们建议使用Per-job or Application Mode,因为他们给应用提供了更好的隔离!

3.4.4 Per-Job-Cluster模式执行无界流WordCount

bin/flink run -d -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar

提交任务到Yarn的其他队列
bin/flink run -d -m yarn-application -yqu hive -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar

bin/flink run -d -t yarn-per-job -Dyarn.application.queue hive -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar

3.4.5 Session-Cluster模式执行无界流WordCount

1. 启动一个Flink-Session

   bin/yarn-session.sh -d

2. 在Session上运行Job

   bin/flink run -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
   会自动找到你的yarn-session启动的Flink集群.也可以手动指定你的yarn-session集群:
   bin/flink run -t yarn-session -Dyarn.application.id=application_XXXX_YY ./flink-prepare-1.0-SNAPSHOT.jar
   **注意: **application_XXXX_YY 指的是在yarn上启动的yarn应用

3.4.6 Application Mode模式执行无界流WordCount

? bin/flink run-application -t yarn-application -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
提交任务到Yarn的其他队列
? bin/flink run-application -t yarn-application -Dyarn.application.queue hive -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar

3.4.7 Yarn模式高可用

Yarn模式的高可用和Standalone模式的高可用原理不一样。
Standalone模式中, 同时启动多个Jobmanager, 一个为leader其他为standby的, 当leader挂了, 其他的才会有一个成为leader。
yarn的高可用是同时只启动一个Jobmanager, 当这个Jobmanager挂了之后, yarn会再次启动一个, 其实是利用的yarn的重试次数来实现的高可用.

1. 在yarn-site.xml中配置

<property>
  <name>yarn.resourcemanager.am.max-attempts</name>
  <value>4</value>
  <description>
    The maximum number of application master execution attempts.
  </description>
</property>

**注意: **配置完不要忘记分发, 和重启yarn

1. 在flink-conf.yaml中配置

yarn.application-attempts: 3
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop102:8020/flink/yarn/ha
high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
high-availability.zookeeper.path.root: /flink-yarn

1. 启动yarn-session
2. 杀死Jobmanager, 查看的他的复活情况

注意: yarn-site.xml中是它活的次数的上限, flink-conf.xml中的次数应该小于这个值。

3.5 Scala REPL

scala 交互环境。

1. local模式启动 REPL

   /opt/module/flink-local ? bin/start-scala-shell.sh local

2. yarn-session 模式启动

   先启动一个yarn-session, 然后就可以把shell跑在yarn-session上了
   bin/start-scala-shell.sh yarn

3.6 K8S & Mesos模式

Mesos是Apache下的开源分布式资源管理框架，它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署，但是在国内，依然使用着传统的Hadoop大数据框架，所以国内使用mesos框架的并不多，这里我们就不做过多讲解了。
容器化部署时目前业界很流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。这里我们也不做过多的讲解。

3.7 Windows模式

在学习时，每次都需要启动虚拟机，启动集群，这是一个比较繁琐的过程，并且会占大量的系统资源，导致系统执行变慢，不仅仅影响学习效果，也影响学习进度，Flink提供了可以在windows系统下启动本地集群的方式，这样，在不使用虚拟机的情况下，也能学习flink的基本使用，所以后续中，为了能够给同学们更加流畅的教学效果和教学体验，我们一般情况下都会采用windows系统的集群来学习flink。

Windows系统搭建Flink集群具体如下：

1. 将文件flink-1.10.0-bin-scala_2.11.tgz解压缩到无中文无空格的路径中

2. 修改conf/flink-conf.yaml文件，添加配置

   注意：行头部不允许有空格，因为yaml文件是以空格区分上下级的！！！

   taskmanager.cpu.cores: 1.8
   taskmanager.memory.task.heap.size: 2048m
   taskmanager.memory.managed.size: 2048m
   taskmanager.memory.network.fraction: 0.1
   taskmanager.memory.network.min: 64mb
   taskmanager.memory.network.max: 64mb
   

3. 执行脚本：bin/start-cluster.bat，启动两个进程

4. 在浏览器端输入：localhost:8081即可访问。

四、Flink运行架构

4.1 运行架构

https://ci.apache.org/projects/flink/flink-docs-release-1.11/fig/processes.svg(图片来源地址)

4.1.1 官网图解：

4.1.2 中文图解：

Flink运行时包含2种进程:1个JobManager和至少1个TaskManager

4.2 核心组件

4.2.0 Flink VS Spark核心组件

4.2.1 Application Master

这个进程包含3个不同的组件 : Dispatcher、ResourceManager、JobManager

4.2.1.1 Dispatcher

负责接收用户提供的作业，并且负责为这个新提交的作业启动一个新的JobManager 组件

4.2.1.2 ResourceManager

1. 负责资源的管理，在整个 Flink 集群中只有一个 ResourceManager。（管理多个JobManager）
2. 注意这个ResourceManager不是Yarn中的ResourceManager, 是Flink中内置的, 只是赶巧重名了而已。

主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger插槽是Flink中定义的处理资源单元。
当JobManager申请插槽资源时，ResourceManager会将有空闲插槽的TaskManager分配给JobManager。如果ResourceManager没有足够的插槽来满足JobManager的请求，它还可以向资源提供平台发起会话，以提供启动TaskManager进程的容器。另外，ResourceManager还负责终止空闲的TaskManager，释放计算资源。

4.2.1.2 JobManager

负责管理作业的执行，在一个Flink集群中可能有多个作业同时执行，每个作业都有自己的JobManager组件。
控制一个应用程序执行的主进程，也就是说，每个应用程序都会被一个不同的JobManager所控制执行。

JobManager会先接收到要执行的应用程序，这个应用程序会包括：作业图（JobGraph）、逻辑数据流图（logical dataflow graph）和打包了所有的类、库和其它资源的JAR包。
JobManager会把JobGraph转换成一个物理层面的数据流图，这个图被叫做“执行图”（ExecutionGraph），包含了所有可以并发执行的任务。JobManager会向资源管理器（ResourceManager）请求执行任务必要的资源，也就是任务管理器（TaskManager）上的插槽（slot）。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。
而在运行过程中，JobManager会负责所有需要中央协调的操作，比如说检查点（checkpoints）的协调。

4.2.1.2 JobMaster

JobMaster负责管理单个JobGraph的执行。多个Job可以同时运行在一个Flink集群中, 每个Job都有一个自己的JobMaster，是JobManager的物理单元。

4.2.2 TaskManager

Flink中的工作进程（JobManager）。通常在Flink中会有多个TaskManager运行，每一个TaskManager都包含了一定数量的插槽（slots）。插槽的数量限制了TaskManager能够执行的任务数量。
启动之后，TaskManager会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager就会将一个或者多个插槽提供给JobManager调用。JobManager就可以向插槽分配任务（tasks）来执行了。
在执行过程中，一个TaskManager可以跟其它运行同一应用程序的TaskManager交换数据。

4.2.3 Cluster Manager

集群管理器，比如Standalone、YARN、K8s等，就是前面我们学习的不同环境。

4.2.4 Client

提交Job的客户端，可以是运行在任何机器上（与 JobManager 环境连通即可）。提交Job后，Client可以结束进程（Streaming的任务），也可以不结束并等待结果返回。

4.3 核心概念

4.3.1 TaskManager与Slots

Flink中每一个worker(TaskManager)都是一个JVM进程，它可能会在独立的线程上执行一个Task。为了控制一个worker能接收多少个task，worker通过Task Slot来进行控制（一个worker至少有一个Task Slot）。
这里的Slot如何来理解呢？很多的文章中经常会和Spark框架进行类比，将Slot类比为Core，其实简单这么类比是可以的，可实际上，可以考虑下，当Spark申请资源后，这个Core执行任务时有可能是空闲的（空转），但是这个时候Spark并不能将这个空闲下来的Core共享给其他Job使用，所以这里的Core是Job内部共享使用的。接下来我们再回想一下，之前在Yarn Session-Cluster模式时，其实是可以并行执行多个Job的，那如果申请两个Slot，而执行Job时，只用到了一个，剩下的一个怎么办？那我们自认而然就会想到Flink可以将这个空闲的Slot给并行的其他Job，对吗？所以Flink中的Slot和Spark中的Core还是有很大区别的。
每个task slot表示TaskManager拥有资源的一个固定大小的子集。假如一个TaskManager有三个slot，那么它会将其管理的内存分成三份给各个slot。资源slot化意味着一个task将不需要跟来自其他job的task竞争被管理的内存，取而代之的是它将拥有一定数量的内存储备。需要注意的是，这里不会涉及到CPU的隔离，slot目前仅仅用来隔离task的受管理的内存。

4.3.2 Parallelism（并行度）

并行度设置的优先级：

算子(代码) > 全局(代码) > 提交参数 > 配置文件

Parallelism 和 Slots之间的关系 ：

1. slots表示此任务可以承担的最大的线程数量，是静态的。
2. prallelism表示任务运行时实际开启的线程数量，是动态的。（Parallelism <= slots）
3. 并行度的计算推理原理：

• Flink的并行度可以对算子进行设置，那么算子的子任务的数量，就是算子的并行度。
• Job的并行度 => 并行度最大的算子的并行度。
• 一个Job需要多少并行度 ？ 取决于 Job 的并行度。

在学习Spark RDD时，无论是读取内存中的数据，或读取文件数据，都会接触一个叫并行度的概念，并且在RDD的算子中也可以动态改变并行度，通过学习，咱们应该知道Spark中的并行度最终体现为分区，而分区又意味着Task。所以Spark 计算中Task的数量是可以通过并行度推算出来的。这个大家没有的问题的话，那就好办了，为什么？因为Flink的并行度的作用和Spark中并行度的作用的一样的。最后都可以表现为任务的并行执行。

虽然Spark中的并行度和Flink的并行度的原理，作用基本一致，但是由于模型选择的问题，所以使用上依然有些细微的区别：
	? Spark的并行度设置后需要调用特殊的算子（repartition）或特殊的操作（shuffle）才能进行改变，比如调用			flatMap算子后再调用repartition改变分区。
	? Flink的并行度设置可以在任何算子后使用，并且为了方便，也可以设置全局并行度
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(2);
	
但是需要注意，某些数据源数据的采集是无法改变并行度，如Socket
	? 如果Flink的一个算子的并行度为2，那么这个算子在执行时，这个算子对应的task就会拆分成2个subtask，发到不同的Slot中执行。

一个特定算子的子任务（subtask）的个数被称之为这个算子的并行度（parallelism），一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度。并行度是单指一个任务中的，而不是整个程序的。

4.3.3 算子与算子之间的关系

Stream在算子之间传输数据的形式可以是one-to-one(forwarding)的模式也可以是redistributing的模式，具体是哪一种形式，取决于算子的种类。

4.3.3.1 One-to-one

类似于spark中的窄依赖

stream（比如在source和flatMap operator之间）维护着分区以及元素的顺序。那意味着flatmap 算子的子任务看到的元素的个数以及顺序跟source 算子的子任务生产的元素的个数、顺序相同，map、fliter、flatMap等算子都是one-to-one的对应关系。

4.3.3.2 Redistributing

类似于spark中的宽依赖

stream(map()跟keyBy/window之间或者keyBy/window跟sink之间)的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。

例如，keyBy()基于**hashCode重分区（类似于Spark中Shuffle）、broadcast（广播）和rebalance（重分配）**会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似于Spark中的shuffle过程。

1. hashCode重分区

1. broadcast广播

1. rebalance

4.3.4 Task与SubTask

算子的一个并行子任务，叫做subtask。

**task ：**是由①不同算子的subtask(子任务)②根据一定的规则③合并在一起形成。

不同算子组成一个 task 的条件：详情请查看下一章节 [4.3.5 Operator Chains（任务链）](# 4.3.5 Operator Chains（任务链）)

①：并行度一致

②：两个算子之间是one-to-one的关系

Flink执行时,由于并行度的设置，可以将同一个Job不同算子的subtask（子任务）放在同一块内存中进行处理，那么这样在执行时就可以合并成一个完整的task进行处理，而不是独立的子任务，这样就减少了子任务（SubTask）之间调度和数据传递的性能损耗，避免了跨节点进行数据传输。

4.3.5 Operator Chains（任务链）

把相同并行度one to one操作，Flink将这样相连的算子链接在一起形成一个 task ，原来的算子成为里面的一部分。 每个task被一个线程执行.
将算子链接成task是非常有效的优化：它能减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。链接的行为可以在编程API中进行指定。

阻止任务链的常用方法：

使用下面的方法后，哪怕两个算子之间是one to one的关系，并且并行度一致，也不会形成任务链。

执行环境对象：disableOperatorChaining()方法

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.disableOperatorChaining();

算子：startNewChain()方法 disableChaining()方法

算子.startNewChain();
算子.disableChaining();

4.3.6 ExecutionGraph（执行图）

由Flink程序直接映射成的数据流图是StreamGraph，也被称为逻辑流图，因为它们表示的是计算逻辑的高级视图。为了执行一个流处理程序，Flink需要将逻辑流图转换为物理数据流图（也叫执行图），详细说明程序的执行方式。
Flink 中的执行图可以分成四层：StreamGraph -> JobGraph -> ExecutionGraph -> Physical Graph。

2个并发度（Source为1个并发度）的 SocketTextStreamWordCount 四层执行图的演变过程 env.socketTextStream().flatMap(…).keyBy(0).sum(1).print();

• StreamGraph：

是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。

• JobGraph：

StreamGraph经过优化后生成了 JobGraph，是提交给 JobManager 的数据结构。主要的优化为: 将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。

• ExecutionGraph：

JobManager 根据 JobGraph 生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。

• Physical Graph：

JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个TaskManager 上部署 Task 后形成的“图”，并不是一个具体的数据结构。

4.4 提交流程

下面的两张图比较笼统，而且yarn-cluster模式不准确，完整准确的提交流程请采用[4.1.2 中文图解：](# 4.1.2 中文图解：)结合下方执行流程说明作详细系统学习。

4.4.1 高级视角提交流程(通用提交流程)

我们来看看当一个应用提交执行时，Flink的各个组件是如何交互协作的：

4.4.2 yarn-cluster提交流程per-job

执行流程说明：

1. Flink任务提交后，Client向HDFS上传Flink的Jar包和配置
2. 向Yarn ResourceManager提交任务，ResourceManager分配Container资源
3. 通知对应的NodeManager启动ApplicationMaster，ApplicationMaster启动后加载Flink的Jar包和配置构建环境，然后启动JobManager
4. ApplicationMaster向ResourceManager申请资源启动TaskManager
5. ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager
6. NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager
7. TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。

五、Flink流处理核心编程

和其他所有的计算框架一样，Flink也有一些基础的开发步骤以及基础，核心的API，从开发步骤的角度来讲，主要分为四大部分

5.1 Environment

Flink Job在提交执行计算时，需要首先建立和Flink框架之间的联系，也就指的是当前的flink运行环境，只有获取了环境信息，才能将task调度到不同的taskManager执行。而这个环境对象的获取方式相对比较简单

以下的两种API都可以自动区分部署环境

// 批处理环境
ExecutionEnvironment benv = ExecutionEnvironment.getExecutionEnvironment();
// 流式数据处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

5.2 Source

Flink框架可以从不同的来源获取数据，将数据提交给框架进行处理, 我们将获取数据的来源称之为数据源**(Source)**。

5.2.1 准备工作

1. 导入注解工具依赖, 方便生产POJO类

   <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
   <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>
       <version>1.18.16</version>
       <scope>provided</scope>
   </dependency>

2. 准备一个WaterSensor类方便演示

   /**
    * 水位传感器：用于接收水位数据
    *
    * id:传感器编号
    * ts:时间戳
    * vc:水位
    */
   @Data
   @NoArgsConstructor
   @AllArgsConstructor
   public class WaterSensor {
       private String id;
       private Long ts;
       private Integer vc;
   }

5.2.2 从Java的集合中读取数据

一般情况下，可以将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用。这里的数据结构采用集合类型是比较普遍的。

public class Flink05_Source_Collection {
    public static void main(String[] args) throws Exception {
        List<WaterSensor> waterSensors = Arrays.asList(
          new WaterSensor("ws_001", 1577844001L, 45),
          new WaterSensor("ws_002", 1577844015L, 43),
          new WaterSensor("ws_003", 1577844020L, 42));

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
          .fromCollection(waterSensors)
         .print();
        env.execute();
    }
}

5.2.3 从文件读取数据

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

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
          .readTextFile("input")
          .print();

        env.execute();
    }
}

说明:

1. 参数可以是目录也可以是文件

2. 路径可以是相对路径也可以是绝对路径

3. 相对路径是从系统属性user.dir获取路径: idea下是project的根目录, standalone模式下是集群节点根目录

4. 也可以从hdfs目录下读取, 使用路径:hdfs://…, 由于Flink没有提供hadoop相关依赖, 需要pom中添加相关依赖:

   <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client</artifactId>
       <version>3.1.3</version>
       <scope>provided</scope>
   </dependency>
   

5.2.4 从Socket读取数据

[参考第二章无界流读取](# 2.3.2 无界流)

5.2.5 从Kafka读取数据

添加相应的依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>1.13.0</version>
</dependency>

参考代码

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

        // 0.Kafka相关配置
        Properties conf = new Properties(); 	
        // conf.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
        conf.setProperty("bootstrap.servers","hadoop102:9092,hadoop103:9092,hadoop104:9092");
        conf.setProperty("group.id", "Flink01_Source_Kafka");
        conf.setProperty("auto.offset.reset", "latest");

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 添加Kafka的数据源Source
       		 /*
                参数1：kafka中的主题
                参数2：序列化规则
                参数3：kafka的相关配置信息
        	*/
          DataStreamSource<String> kafkaSource = env.addSource(new FlinkKafkaConsumer<>											("sensor", new SimpleStringSchema(), conf));
        
         // 打印     
         kafkaSource.print("kafka source");

         env.execute();
    }
}

5.2.6 自定义Source

大多数情况下，前面的数据源已经能够满足需要，但是难免会存在特殊情况的场合，所以flink也提供了能自定义数据源的方式。

示例1：

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

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
          .addSource(new MySource("hadoop102", 9999))
          .print();

        env.execute();
    }

    // 自定义数据源，实现SourceFunction接口，泛型即是数据的类型
    /*
    	重写接口中的两个方法 ：
		①：run()方法就是程序启动的方法
		②：cancel()方法就是程序结束的方法
	*/
    public static class MySource implements SourceFunction<WaterSensor> {
        private String host;
        private int port;
        private volatile boolean isRunning = true;
        private Socket socket;

        public MySource(String host, int port) {
            this.host = host;
            this.port = port;
        }

        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            // 实现一个从socket读取数据的source
            socket = new Socket(host, port);
            BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
            String line = null;
            while (isRunning && (line = reader.readLine()) != null) {
                String[] split = line.split(",");
                ctx.collect(new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2])));
            }
        }

        /**
         * 大多数的source在run方法内部都会有一个while循环,
         * 当调用这个方法的时候, 应该可以让run方法中的while循环结束
         */

        @Override
        public void cancel() {
            isRunning = false;
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
/*
sensor_1,1607527992000,20
sensor_1,1607527993000,40
sensor_1,1607527994000,50
 */

示例2：

public class flink05_source_consum {

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

        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 获取数据源
        DataStreamSource<WaterSensor> ConsumSource = env.addSource(new MySource());

        // 3. 算子操作

        // 4. 输出、保存...
        ConsumSource.print();
        // 5. 启动执行
        env.execute();

    }

    private static class MySource implements SourceFunction<WaterSensor> {

        private boolean flag = true;
        Random random = new Random();
        // 启动数据源
        @Override
        public void run(SourceContext<WaterSensor> sourceContext) throws Exception {
            while(flag){
                // 准备数据源
                WaterSensor waterSensor = new WaterSensor(
                        "sensor_" + random.nextInt(10),
                        System.currentTimeMillis()+ 492,
                        random.nextInt(100));
                Thread.sleep(1000);
                // 向下游发送数据
                sourceContext.collect(waterSensor);
            }
        }

        // 释放数据源
        @Override
        public void cancel() {
            this.flag = false;
        }
    }
}

5.3 Transform

转换算子可以把一个或多个DataStream转成一个新的DataStream。程序可以把多个复杂的转换组合成复杂的数据流拓扑。类似于Spark中的转换算子。

5.3.0 算子的分类

如果算子前后是一对一关系，则算子中没有收集器，比如map算子

如果算子前后是一对多关系，则算子中有收集器，比如flatMap算子

1. 算子分类

**无状态算子 : ** 数据的输入和数据的输出是一致的，对数据没有聚合作用。

1. map

2. flatMap

3. filter

数据传输算子：

1. keyBy

**多流转换算子：**涉及到对多条流的合并、拆分等操作。

1. split
2. select
3. connect —> coMap
4. union

5.3.1 map

作用
将数据流中的数据进行转换, 形成新的数据流，消费一个元素并产出一个元素

适用场景：把符合规范的数据转换为javaBean实体类

参数
lambda表达式或MapFunction实现类

返回
DataStream → DataStream（可以转换数据的类型）

示例
得到一个新的数据流: 新的流的元素是原来流的元素的平方

1. 匿名内部类对象

package com.heather.flink.java.chapter_5.transform;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

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

        env
          .fromElements(1, 2, 3, 4, 5)
           // MapFunction()方法的参数：
           // 参数1：数据的输入的类型
           // 参数2：数据的输出的类型
          .map(new MapFunction<Integer, Integer>() {
              @Override
              public Integer map(Integer value) throws Exception {
                  return value * value;
              }
          })
          .print();

        env.execute();
    }
}

2. Lambda表达式表达式

env
  .fromElements(1, 2, 3, 4, 5)
  .map(ele -> ele * ele)
  .print();

3. 静态内部类

package com.heather.flink.java.chapter_5.transform;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

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

        env
          .fromElements(1, 2, 3, 4, 5)
          .map(new MyMapFunction())
          .print();

        env.execute();
    }

    public static class MyMapFunction implements MapFunction<Integer, Integer> {

        @Override
        public Integer map(Integer value) throws Exception {
            return value * value;
        }
    }
    
}

Rich…Function类
所有Flink函数类都有其Rich版本。它与常规函数的不同在于提供了更多的，更丰富的功能：

1. 可以获取运行环境的上下文（主要用于获取数据的状态）
2. 提供了获取连接的生命周期方法（open()和close()，各执行一次）
   • open()方法：执行的次数和并行度成**“正相关”**。
   • close()方法：执行的次数和并行度成**“正相关”，有界流会是并行度 * 2**。

例如：

RichMapFunction

package com.heather.flink.java.chapter_5.transform;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flink01_TransForm_Map_RichMapFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(5);

        env
          .fromElements(1, 2, 3, 4, 5)
          .map(new MyRichMapFunction()).setParallelism(2)
          .print();

        env.execute();
    }

    public static class MyRichMapFunction extends RichMapFunction<Integer, Integer> {
        // 默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次
        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("open ... 执行一次");
        }

        // 默认生命周期方法, 最后一个方法, 做一些清理工作, 在每个并行度上只调用一次
        @Override
        public void close() throws Exception {
            System.out.println("close ... 执行一次");
        }

        @Override
        public Integer map(Integer value) throws Exception {
            System.out.println("map ... 一个元素执行一次");
            return value * value;
        }
    }
}

说明：

1. 默认生命周期方法, 初始化方法, 在每个并行度上只会被调用一次, 而且先被调用。
2. 默认生命周期方法, 最后一个方法, 做一些清理工作, 在每个并行度上只调用一次, 而且是最后被调用。
3. getRuntimeContext()方法提供了函数的RuntimeContext的一些信息，例如函数执行的并行度，任务的名字，以及state状态. 开发人员在需要的时候自行调用获取运行时上下文对象。

5.3.2 flatMap

作用
消费一个元素并产生零个或多个元素

参数
参数1：Integer value ： 输入的数据

参数2：Collector out ： 输出的数据

返回
DataStream → DataStream（可以转换数据的类型）

示例 ： 如下 ↓

1. 匿名内部类写法

// 新的流存储每个元素的平方和3次方
env
  .fromElements(1, 2, 3, 4, 5)
   // FlatMapFunction()方法的参数：
   // 参数1：数据的输入的类型
   // 参数2：数据的输出的类型
  .flatMap(new FlatMapFunction<Integer, Integer>() {
      @Override
      public void flatMap(Integer value, Collector<Integer> out) throws Exception {
          out.collect(value * value);
          out.collect(value * value * value);
      }
  })
  .print();

2. Lambda表达式写法

env
  .fromElements(1, 2, 3, 4, 5)
  .flatMap((Integer value, Collector<Integer> out) -> {
      out.collect(value * value);
      out.collect(value * value * value);
  }).returns(Types.INT)
  .print();

说明: 在使用Lambda表达式表达式的时候, 由于泛型擦除的存在, 在运行的时候无法获取泛型的具体类型, 全部当做Object来处理, 及其低效, 所以Flink要求当参数中有泛型的时候, 必须明确指定泛型的类型.

5.3.3 filter

作用
根据指定的规则将满足条件（true）的数据保留，不满足条件(false)的数据丢弃

参数
FlatMapFunction实现类

返回
DataStream → DataStream（不可以转换数据的类型，仅仅是过滤的作用）

示例 ： 如下 ↓

1. 匿名内部类写法

// 保留偶数, 舍弃奇数
env
  .fromElements(10, 3, 5, 9, 20, 8)
  .filter(new FilterFunction<Integer>() {
      @Override
      public boolean filter(Integer value) throws Exception {
          return value % 2 == 0;
      }
  })
  .print();

2. Lambda表达式写法

env
  .fromElements(10, 3, 5, 9, 20, 8)
  .filter(value -> value % 2 == 0)
  .print();

5.3.4 keyBy

作用

1. （键控流）把流中的数据分到不同的分区中，具有相同key的元素会分到同一个分区中【任务槽】（数据倾斜的根源），不同的key也有可能在同一个分区【任务槽】中，但是逻辑上是隔离开的，即 给每个数据打上标签（属于哪个分组），并不是对并行度进行改变。

2. 通过keyBy之后得到keyedStream流，keyedStream流就可以进行各种各样的聚合算子进行操作。

   [5.3.10 简单滚动聚合算子](# 5.3.10 简单滚动聚合算子)

分区规则：在内部是使用的hash分区来实现的，根据key的hash值对key进行求模操作。

难点:

【重点】 【重点】 【重点】

KeyedStream的参数说明：

Type parameters:
– The type of the elements in the Keyed Stream. （返回值得类型）
– The type of the key in the Keyed Stream.（分组的key的类型）

KeyedStream<Tuple2<String, Integer>, Tuple> wordToGroupDS = wordAndOneDS.keyBy(0);
KeyedStream<WaterSensor, Tuple> sensorKSByFieldName = sensorDS.keyBy("id");

通过上述的两个代码可知，无论是通过 位置索引 还是 字段名称 进行分组，返回的key的类型都是Tuple（因为无法确定），这就导致后续使用key的时候还需要进行类型的转换等。

**所以：**通过 明确的指定 key 的方式， 获取到的 key就是具体的类型 => 实现 KeySelector 或 lambda

参数
Key选择器函数: interface KeySelector<IN, KEY>

? 注意: 什么值不可以作为KeySelector的Key:

• 没有覆写hashCode方法的POJO, 而是依赖Object的hashCode. 因为这样分组没有任何的意义: 每个元素都会得到一个独立无二的组. 实际情况是:可以运行, 但是分的组没有意义.
• 任何类型的数组

返回
DataStream → KeyedStream

示例：明确的指定 key 的方式【对象.get属性()】 ： 如下 ↓

1. 匿名内部类写法

public class Flink02_Transform_KeyBy {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1. 获取数据源
        DataStreamSource<String> fileDS = env.readTextFile("input/sensor_data.log");
        // 2. 操作算子
        // 2.1 map()算子把数据转换为对象
        SingleOutputStreamOperator<WaterSensor> mapDS = fileDS.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] datas = value.split(",");
                return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
            }
        });
        // 2.2 分组（下面的三种方式请观察返回值类型中的泛型）

        // 2.2.1 通过指定下标索引的方式【不推荐使用】
        // KeyedStream<WaterSensor, Tuple> keyByDS = mapDS.keyBy(0);
        // 2.2.2 通过指定 属性 的方式【不推荐使用】
        // KeyedStream<WaterSensor, Tuple> keyByDS = mapDS.keyBy("id");
        // 2.2.3 明确的指定 key 的 方式 【推荐使用】
        KeyedStream<WaterSensor, String> keybyDS = mapDS.keyBy(new MyKeyBySeletor());

        // 3. 输出、保存 ...
        keybyDS.print();
        env.execute();
    }

    private static class MyKeyBySeletor implements KeySelector<WaterSensor,String> {
        @Override
        public String getKey(WaterSensor value) throws Exception {
            return value.getId();
        }
    }
}

2. Lambda表达式写法

public class Flink02_Transform_KeyBy {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1. 获取数据源
        DataStreamSource<String> fileDS = env.readTextFile("input/sensor_data.log");
        // 2. 操作算子
        // 2.1 map()算子把数据转换为对象
        SingleOutputStreamOperator<WaterSensor> mapDS = fileDS.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] datas = value.split(",");
                return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
            }
        });
        // 2.2 分组【lambda表达式方法】
        KeyedStream<WaterSensor, String> keybyDS = mapDS.keyBy( r -> r.getId() );

        // 3. 输出、保存 ...
        keybyDS.print();
        env.execute();
    }
}

5.3.6 shuffle

作用
把流中的元素随机打乱，然后随机（底层是分区器）发送到下游，对同一个组数据, 每次只需得到的结果都不同。

参数
无

返回
DataStream → DataStream

示例 ： 如下 ↓

env
  .fromElements(10, 3, 5, 9, 20, 8)
  .shuffle()
  .print();
env.execute();

5.3.7 split和select(过时)

已经过时, 在1.12中已经被移除

作用
某些情况下，我们需要将数据流根据某些特征拆分成两个或者多个数据流，给不同数据流增加标记以便于从流中取出。
split用于给流中的每个元素添加标记. select用于根据标记取出对应的元素, 组成新的流。在新的版本中已经被新的API替代：Please use side output instead

~

参数
split参数: interface OutputSelector
select参数: 字符串

返回
split: SingleOutputStreamOperator -> SplitStream~

slect: SplitStream -> DataStream

示例 ： 如下 ↓

1. 匿名内部类写法

public class Flink03_Transform_select {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> fileDS = env.readTextFile("input/sensor_data.log");

        // 把数据转换为javaBean类
        SingleOutputStreamOperator<WaterSensor> mapDS = fileDS.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] datas = value.split(",");
                return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
            }
        });

        // 把一个流中的元素按照指定规则进行切分（其实就是给符合规则的元素贴上标签）
        SplitStream<WaterSensor> splitDS = mapDS.split(new MyOutputSeletor());

//        splitDS.select("normal").print("正常");
//        splitDS.select("warning").print("报警");
//        splitDS.select("alarm").print("警告");

        splitDS.select("balance").print("正常");
        splitDS.select("balance").print("报警");

        env.execute();

    }

    private static class MyOutputSeletor implements OutputSelector<WaterSensor> {
        /*
            定义规则：
                正常： 水位小于95
                报警： 水位大于95，小于98
                警告： 水位大于98
         */
        @Override
        public Iterable<String> select(WaterSensor value) {
            // 返回值类型是String类型的迭代器，说明，每一个元素都可以贴上多个标签进行说明
            if(value.getVc() <= 95){
                return Arrays.asList("normal","balance");
            }else if(value.getVc() < 98){
                return Arrays.asList("warning","balance");
            }else if(value.getVc() >= 98){
                return Arrays.asList("alarm");
            }else {
                return Arrays.asList("。。。。");
            }
        }
    }
}

2. Lambda表达式写法

// 奇数一个流, 偶数一个流
SplitStream<Integer> splitStream = env
  .fromElements(10, 3, 5, 9, 20, 8)
  .split(value -> value % 2 == 0
    ? Collections.singletonList("偶数")
    : Collections.singletonList("奇数"));
splitStream
  .select("偶数")
  .print("偶数");

splitStream
  .select("奇数")
  .print("奇数");
env.execute();

5.3.8 connect

作用
在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息，这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。
Flink中的connect算子可以连接两个保持他们类型的数据流，两个数据流被connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

扩展：

1. 以FIFO队列的方式进入到合流中。
2. connect算子只能合并两条数据流，但是这两条数据流的数据类型可以不一致。

参数
另外一个流

返回
DataStream[A], DataStream[B] -> ConnectedStreams[A,B]

示例 ： 如下 ↓

DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource<String> stringStream = env.fromElements("a", "b", "c");
// 把两个流连接在一起: 貌合神离
ConnectedStreams<Integer, String> cs = intStream.connect(stringStream);
cs.getFirstInput().print("first");
cs.getSecondInput().print("second");
env.execute();

注意：

1. 两个流中存储的数据类型可以不同
2. 只是机械的合并在一起, 内部仍然是分离的2个流
3. 只能2个流进行connect, 不能有第3个参与

5.3.9 union

作用
对两个或者两个以上的DataStream流进行**union（合并）**操作，产生的结果是一个包含所有DataStream流的元素的新DataStream流。

扩展：

1. 以FIFO队列的方式进入到合流中
2. 不去重
3. union算子可以合并多条流 ，但是保证每一条流的元素的类型必须一致。

示例 ： 如下 ↓

DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3, 4, 5);
DataStreamSource<Integer> stream2 = env.fromElements(10, 20, 30, 40, 50);
DataStreamSource<Integer> stream3 = env.fromElements(100, 200, 300, 400, 500);

// 把多个流union在一起成为一个流, 这些流中存储的数据类型必须一样: 水乳交融
stream1
  .union(stream2)
  .union(stream3)
  .print();

connect与 union 区别：

1. union之前两个流的类型必须是一样，connect可以不一样。
2. connect只能操作两个流，union可以操作多个。
3. connect连接后的两个流其实也是两条流，并没有真正的合并为一条流（因为流的类型不一致）具体查看源码，而union是真的把多条流合并在了一起，组成了一条流。

5.4 Operator

5.4.1 滚动聚合算子RollingAgg

1、简言之：数据来一条，聚合一条，输出一次！！！

2、在适用Operarot算子之前，必须先适用keyBy算子对数据进行分组

常见的滚动聚合算子

• sum
• min
• ma
• minBy
• maxBy

作用
KeyedStream的每一个支流做聚合。执行完成后，会将聚合的结果合成一个流返回，所以结果都是DataStream

参数
如果流中存储的是POJO或者scala的样例类, 参数使用字段名
如果流中存储的是元组, 参数就是位置(基于0…)

返回
KeyedStream -> SingleOutputStreamOperator

示例 ： 如下 ↓

• 示例1

DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4, 5);
KeyedStream<Integer, String> kbStream = stream.keyBy(ele -> ele % 2 == 0 ? "奇数" : "偶数");
kbStream.sum(0).print("sum");
kbStream.max(0).print("max");
kbStream.min(0).print("min");

• 示例2

ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 30));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

KeyedStream<WaterSensor, String> kbStream = env
  .fromCollection(waterSensors)
  .keyBy(WaterSensor::getId);

kbStream
  .sum("vc")
  .print("maxBy...");

注意: **
分组聚合后, 理论上只能取分组字段和聚合结果**, 但是Flink允许其他的字段也可以取出来, 其他字段默认情况是取的是这个组内第一个元素的字段值

• 示例3:

ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

KeyedStream<WaterSensor, String> kbStream = env
  .fromCollection(waterSensors)
  .keyBy(WaterSensor::getId);

kbStream
  .maxBy("vc", false)
  .print("max...");

env.execute();

注意: **
maxBy和minBy可以指定当出现相同值的时候,其他字段是否取第一个**. true表示取第一个, false表示取最后一个.

5.4.2 process

作用
process算子在Flink算是一个比较底层的算子，很多类型的流上都可以调用,可以从流中获取更多的信息(不仅仅数据本身)

1. 示例1: 在keyBy之前的流上使用（ProcessFunction）

env
  .fromCollection(waterSensors)
  .process(new ProcessFunction<WaterSensor, Tuple2<String, Integer>>() {
      @Override
      public void processElement(WaterSensor value,
                                 Context ctx,
                                 Collector<Tuple2<String, Integer>> out) throws Exception {
          out.collect(new Tuple2<>(value.getId(), value.getVc()));
      }
  })
  .print();

2. 示例2: 在keyBy之后的流上使用 （KeyedProcessFunction）

env
  .fromCollection(waterSensors)
  .keyBy(WaterSensor::getId)
  .process(new KeyedProcessFunction<String, WaterSensor, Tuple2<String, Integer>>() {
      @Override
      public void processElement(WaterSensor value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
          out.collect(new Tuple2<>("key是:" + ctx.getCurrentKey(), value.getVc()));
      }
  })
  .print();

3. 综合案例：

public class Flink06_Transform_process {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //DataStreamSource<String> fileDS = env.socketTextStream("localhost", 9999);

        DataStreamSource<String> fileDS = env.readTextFile("input/sensor_data.log");
        // 转换为对象
        SingleOutputStreamOperator<WaterSensor> beanDS = fileDS.map((MapFunction<String, WaterSensor>) r -> {
            String[] datas = r.split(",");
            return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
        });

        // 获取对象中的各个属性，针对于属性进行分组
        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> mapDS = beanDS.map(new MapFunction<WaterSensor, Tuple3<String, Long, Integer>>() {
            @Override
            public Tuple3<String, Long, Integer> map(WaterSensor value) throws Exception {
                return new Tuple3<>(value.getId(), value.getTs(), value.getVc());
            }
        });

        // 分组
        // 适用lambda方式
        KeyedStream<Tuple3<String, Long, Integer>, String> keyByDS = mapDS.keyBy(r -> r.f0);

        keyByDS.process(new KeyedProcessFunction<String, Tuple3<String, Long, Integer>, String>() {
            /**
             * processElement 处理数据的方法：来一条，处理一条
             * @param value 一条数据
             * @param ctx 上下文对象
             * @param out 采集器
             * @throws Exception
             */
            @Override
            public void processElement(Tuple3<String, Long, Integer> value, Context ctx, Collector<String> out) throws Exception {
                out.collect("当前key : " + ctx.getCurrentKey() +  ",当前时间 ：" + ctx.timestamp() + ",数据 ：" + value);
            }
        }).print("process方法——>");

        env.execute();
    }
}

KeyedProcessFunction类中的Context上下文对象的来源（源码追踪）：

1. KeyedProcessFunction继承自AbstractRichFunction类

public abstract class KeyedProcessFunction<K, I, O> extends AbstractRichFunction {

2. AbstractRichFunction类中定义了上下文对象

public abstract class AbstractRichFunction implements RichFunction, Serializable {

   private transient RuntimeContext runtimeContext;

所以KeyedProcessFunction类中的Context上下文对象来源于AbstractRichFunction类

5.4.3 reduce

作用
（简化规约 — 聚合）一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。
为什么还要把中间值也保存下来? 考虑流式数据的特点: 没有终点, 也就没有最终的概念了. 任何一个中间的聚合结果都是值!（也就是flink中的状态）

参数
interface ReduceFunction

返回
KeyedStream -> SingleOutputStreamOperator

示例 ： 如下 ↓

1. 匿名内部类写法

ArrayList<WaterSensor> waterSensors = new ArrayList<>();
waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

KeyedStream<WaterSensor, String> kbStream = env
  .fromCollection(waterSensors)
  .keyBy(WaterSensor::getId);

kbStream
  .reduce(new ReduceFunction<WaterSensor>() {
      /*
          参数说明：
             - 参数1 - value1：累加器，原始数据
             - 参数2 - value2：新添加的数据
       */
      @Override
      public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
          System.out.println("reducer function ...");
          return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
      }
  })
  .print("reduce...");

env.execute();

2. Lambda表达式写法

kbStream
  .reduce((value1, value2) -> {
      System.out.println("reducer function ...");
      return new WaterSensor(value1.getId(), value1.getTs(), value1.getVc() + value2.getVc());
  })
  .print("reduce...");

示例3：

public class Flink05_Transform_reduce {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> fileDS = env.socketTextStream("localhost", 9999);

        // 转换为对象
        SingleOutputStreamOperator<WaterSensor> beanDS = fileDS.map((MapFunction<String, WaterSensor>) r -> {
            String[] datas = r.split(",");
            return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
        });

        // 获取对象中的各个属性，针对于属性进行分组
        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> mapDS = beanDS.map(new MapFunction<WaterSensor, Tuple3<String, Long, Integer>>() {
            @Override
            public Tuple3<String, Long, Integer> map(WaterSensor value) throws Exception {
                return new Tuple3<>(value.getId(), value.getTs(), value.getVc());
            }
        });

        // 分组
        // 适用lambda方式
        KeyedStream<Tuple3<String, Long, Integer>, String> keyByDS = mapDS.keyBy(r -> r.f0);

        // reduce聚合方法
        /*
            1. 输入的类型要一致，输出的类型也一致（构建的新值）。
            2. 首次进入的数据，不会进入reduce（因为reduce必须满足两条数据）
            3. 保存了中间的状态（值）
         * */
        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> result = keyByDS.reduce(new ReduceFunction<Tuple3<String, Long, Integer>>() {
            @Override
            public Tuple3<String, Long, Integer> reduce(Tuple3<String, Long, Integer> value1, Tuple3<String, Long, Integer> value2) throws Exception {
                // 求水位的聚合值
                //return value1.f2 + value2.f2;
                return Tuple3.of("result", 123456789L, value1.f2 + value2.f2);
            }
        });
        result.print("result");

        env.execute();
    }
}

注意:

1. 聚合后结果的类型, 必须和原来流中元素的类型保持一致!

5.4.4 对流重新分区的几个算子

1. KeyBy

先按照key分组, 按照key的双重hash来选择后面的分区

2. shuffle

对流中的元素随机分区

3. reblance

对流中的元素平均分布到每个区.当处理倾斜数据的时候, 进行性能优化

4. rescale

同 rebalance一样, 也是平均循环的分布数据。但是要比rebalance更高效, 因为rescale不需要通过网络, 完全走的"管道"。

5.5 Sink

Sink有下沉的意思，在Flink中所谓的Sink其实可以表示为将数据存储起来的意思，也可以将范围扩大，表示将处理完的数据发送到指定的存储系统的输出操作。类似于Spark中的行动算子。

1. 之前我们一直在使用的print方法其实就是一种Sink

public DataStreamSink<T> print(String sinkIdentifier) {
   PrintSinkFunction<T> printFunction = new PrintSinkFunction<>(sinkIdentifier, false);
   return addSink(printFunction).name("Print to Std. Out");
}

2. Flink内置了一些Sink, 除此之外的Sink需要用户自定义!

   

   

5.5.1 KafkaSink

kafka常用目录技巧：

1. topic --zookeeper
2. producer --broker-list
3. consumer --bootstrap-server

• 添加Kafka Connector依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>1.12.0</version>
</dependency>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.75</version>
</dependency>

• 启动Kafka集群

  [heather@hadoop102 bin]$ kafka.sh start

• Sink到Kafka的示例代码

public class Flink07_Sink_kafka {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.readTextFile("input/sensor_data.log");

        inputDS.addSink(new FlinkKafkaProducer011<String>(
                "hadoop102:9092",
                "sensor110", new SimpleStringSchema())
        );

        // 启动执行程序
        env.execute();
    }
}

• 在linux启动一个消费者, 查看是否收到数据
  kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic sensor110
  

5.5.2 RedisSink

• 添加Redis Connector依赖

  <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-redis -->
  <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-redis_2.11</artifactId>
      <version>1.1.5</version>
  </dependency>

• 启动Redis服务器

• Sink到Redis的示例代码

  package com.heather.flink.java.chapter_5.sink;
  
  import com.alibaba.fastjson.JSON;
  import com.atguigu.flink.java.chapter_5.WaterSensor;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.streaming.connectors.redis.RedisSink;
  import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
  import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
  import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
  import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
  
  import java.util.ArrayList;
  
  public class Flink02_Sink_Redis {
      public static void main(String[] args) throws Exception {
          ArrayList<WaterSensor> waterSensors = new ArrayList<>();
          waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
          waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
          waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
          waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
          waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));
  
          // 连接到Redis的配置
          FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
            .setHost("hadoop102")
            .setPort(6379)
            .setMaxTotal(100)
            .setTimeout(1000 * 10)
            .build();
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
          env
            .fromCollection(waterSensors)
            .addSink(new RedisSink<>(redisConfig, new RedisMapper<WaterSensor>() {
                /*
                  key                 value(hash)
                  "sensor"            field           value
                                      sensor_1        {"id":"sensor_1","ts":1607527992000,"vc":20}
                                      ...             ...
                 */
  
                @Override
                public RedisCommandDescription getCommandDescription() {
                    // 返回存在Redis中的数据类型  存储的是Hash, 第二个参数是外面的key
                    return new RedisCommandDescription(RedisCommand.HSET, "sensor");
                }
  
                @Override
                public String getKeyFromData(WaterSensor data) {
                    // 从数据中获取Key: Hash的Key
                    return data.getId();
                }
  
                @Override
                public String getValueFromData(WaterSensor data) {
                    // 从数据中获取Value: Hash的value
                    return JSON.toJSONString(data);
                }
            }));
  
          env.execute();
      }
  }

• Redis查看是否收到数据
  redis-cli --raw
  

  **注意: **
  发送了5条数据, redis中只有2条数据. 原因是hash的field的重复了, 后面的会把前面的覆盖掉·

5.5.3 ElasticsearchSink

• 添加Elasticsearch Connector依赖

<!-- [https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch6](https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch6) -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
    <version>1.12.0</version>
</dependency>

• 启动Elasticsearch集群
• Sink到Elasticsearch的示例代码

package com.heather.flink.java.chapter_5.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.java.chapter_5.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class Flink03_Sink_ES {
    public static void main(String[] args) throws Exception {
        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        List<HttpHost> esHosts = Arrays.asList(
          new HttpHost("hadoop102", 9200),
          new HttpHost("hadoop103", 9200),
          new HttpHost("hadoop104", 9200));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        env
          .fromCollection(waterSensors)
          .addSink(new ElasticsearchSink.Builder<WaterSensor>(esHosts, new ElasticsearchSinkFunction<WaterSensor>() {

              @Override
              public void process(WaterSensor element, RuntimeContext ctx, RequestIndexer indexer) {
                  // 1. 创建es写入请求
                  IndexRequest request = Requests
                    .indexRequest("sensor")
                    .type("_doc")
                    .id(element.getId())
                    .source(JSON.toJSONString(element), XContentType.JSON);
                  // 2. 写入到es
                  indexer.add(request);
              }
          }).build());

        env.execute();
    }
}

• Elasticsearch查看是否收到数据
  

  注意：

  如果出现如下错误:

  

• 添加log4j2的依赖:

  <dependency>
      <groupId>org.apache.logging.log4j</groupId>
      <artifactId>log4j-to-slf4j</artifactId>
      <version>2.14.0</version>
  </dependency>
  

• 如果是无界流, 需要配置bulk的缓存

esSinkBuilder.setBulkFlushMaxActions(1);

5.5.4 自定义Sink

如果Flink没有提供给我们可以直接使用的连接器，那我们如果想将数据存储到我们自己的存储设备中，怎么办？
我们自定义一个到Mysql的Sink

• 在mysql中创建数据库和表

create database test;
use test;
CREATE TABLE `sensor` (
  `id` varchar(20) NOT NULL,
  `ts` bigint(20) NOT NULL,
  `vc` int(11) NOT NULL,
  PRIMARY KEY (`id`,`ts`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

• 导入Mysql驱动

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.49</version>
</dependency>

• 写到Mysql的自定义Sink示例代码

package com.heather.flink.java.chapter_5.sink;

import com.atguigu.flink.java.chapter_5.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;

public class Flink04_Sink_Custom {
    public static void main(String[] args) throws Exception {
        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        env.fromCollection(waterSensors)
          .addSink(new RichSinkFunction<WaterSensor>() {

             private PreparedStatement ps;
              private Connection conn;

              @Override
              public void open(Configuration parameters) throws Exception {
                  conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useSSL=false", "root", "000000");
                  ps = conn.prepareStatement("insert into sensor values(?, ?, ?)");
              }

              @Override
              public void close() throws Exception {
                ps.close();
                  conn.close();
              }

              @Override
              public void invoke(WaterSensor value, Context context) throws Exception {
                ps.setString(1, value.getId());
                  ps.setLong(2, value.getTs());
                  ps.setInt(3, value.getVc());
                  ps.execute();
              }
          });


        env.execute();
    }
}

5.6 执行模式(Execution Mode)

? Flink在1.12.0上对流式API新增一项特性:可以根据你的使用情况和Job的特点, 可以选择不同的运行时执行模式(runtime execution modes)。
? 流式API的传统执行模式我们称之为STREAMING执行模式, 这种模式一般用于无界流, 需要持续的在线处理。
? 1.12.0新增了一个BATCH执行模式, 这种执行模式在执行方式上类似于MapReduce框架. 这种执行模式一般用于有界数据。
? 默认是使用的STREAMING执行模式。

5.6.1 选择执行模式

BATCH执行模式仅仅用于有界数据, 而STREAMING 执行模式可以用在有界数据和无界数据.

一个公用的规则就是: 当你处理的数据是有界的就应该使用BATCH执行模式, 因为它更加高效. 当你的数据是无界的, 则必须使用STREAMING 执行模式, 因为只有这种模式才能处理持续的数据流.

5.6.2 配置BATH执行模式

执行模式有3个选择可配:

1. STREAMING(默认)
2. BATCH
3. AUTOMATIC

• 通过命令行配置

bin/flink run -Dexecution.runtime-mode=BATCH ...

• 通过代码配置

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

建议:

不要在运行时配置(代码中配置), 而是使用命令行配置, 引用这样会灵活: 同一个应用即可以用于无界数据也可以用于有界数据

5.6.3 有界数据用STREAMING和BATCH的区别

STREAMING模式下, 数据是来一条输出一次结果。

BATCH模式下, 数据处理完之后, 一次性输出结果。

下面展示WordCount的程序读取文件内容在不同执行模式下的执行结果对比 :

• 流式模式

// 默认流式模式, 可以不用配置
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

• 批处理模式

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

• 自动模式

env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

5. 7 综合案例练习

5.7.1 基埋点日志数据的网络流量统计

5.7.1.1 网站总浏览量（PV）的统计

衡量网站流量一个最简单的指标，就是网站的页面浏览量（Page View，PV）。用户每次打开一个页面便记录1次PV，多次打开同一页面则浏览量累计。一般来说，PV与来访者的数量成正比，但是PV并不直接决定页面的真实来访者数量，如同一个来访者通过不断的刷新页面，也可以制造出非常高的PV。接下来我们就用咱们之前学习的Flink算子来实现在PV的统计

1. 数据准备

在咱们当前的案例中，给大家准备了某电商网站的用户行为日志数据UserBehavior.csv,本日志数据文件中包含了某电商网站一天近五十万随机用户的所有行为（包括点击、购买、收藏、喜欢）。数据集的每一行表示一条用户行为，由用户ID、商品ID、商品类目ID、行为类型和时间戳组成，并以逗号分隔。将数据文件放置在指定目录中，便于读取到Flink中使用

2. 读取日志数据转换为JavaBean对象方便操作