第1章 Flink简介

1.1 初识Flink

1) Flink项目的理念是：“Apache Flink是为分布式、高性能、随时可用以及准确的流处理应用程序打造的开源流处理框架”。

2) Apache Flink是一个框架和分布式处理引擎，用于对无界(nc lk 9999)和有界数据(一个文档)流进行有状态计算。Flink被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。

1.2 Flink的重要特点

1.2.1 事件驱动型(Event-driven)

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

1.2.2 流(flink)与批(spark)的世界观

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

4) 无界数据流：

无界数据流有一个开始但是没有结束，它们不会在生成时终止并提供数据，必须连续处理无界流，也就是说必须在获取后立即处理event。对于无界数据流我们无法等待所有数据都到达，因为输入是无界的，并且在任何时间点都不会完成。处理无界数据通常要求以特定顺序（例如事件发生的顺序）获取event，以便能够推断结果完整性

5) 有界数据流：

有界数据流有明确定义的开始和结束，可以在执行任何计算之前通过获取所有数据来处理有界流，处理有界流不需要有序获取，因为可以始终对有界数据集进行排序，有界流的处理也称为批处理。

1.2.3 分层API

Croe APIs--> spark core Table API --> sql --> spark sql

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定义的表上执行。

6) 目前Flink作为批处理还不是主流，不如Spark成熟，所以DataSet使用的并不是很多。Flink Table API和Flink SQL也并不完善，大多都由各大厂商自己定制。==所以我们主要学习DataStream API的使用==。实际上Flink作为最接近Google DataFlow模型的实现，是流批统一的观点，所以基本上使用DataStream就可以了。

7) 2020年12月8日发布的1.12.0版本, 已经完成实现了真正的流批一体, 写好的一套代码, 即可以处理流式数据, 也可以处理离线数据. 这个与前面版本的处理有界流的方式是不一样的, Flink专门对批处理数据做了优化处理.

1.3 Spark or Flink

Spark 基于微批处理的方式需要同步会有额外开销，因此无法在延迟上做到极致。在大数据处理的低延迟场景，Flink 已经有非常大的优势。

1.4 Flink的应用

1.4.1 应用Flink的场景

1) 事件驱动型应用

事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。

事件驱动型应用是在计算存储分离的传统应用基础上进化而来。在传统架构中，应用需要读写远程事务型数据库。

相反，事件驱动型应用是基于状态化流处理来完成。在该设计中，数据和计算不会分离，应用只需访问本地（内存或磁盘）即可获取数据。系统容错性的实现依赖于定期向远程持久化存储写入 checkpoint。下图描述了传统应用和事件驱动型应用架构的区别。

2) 数据分析应用

Apache Flink 同时支持流式及批量分析应用。

Flink 为持续流式分析和批量分析都提供了良好的支持。具体而言，它内置了一个符合 ANSI 标准的 SQL 接口，将批、流查询的语义统一起来。无论是在记录事件的静态数据集上还是实时事件流上，相同 SQL 查询都会得到一致的结果。同时 Flink 还支持丰富的用户自定义函数，允许在 SQL 中执行定制化代码。如果还需进一步定制逻辑，可以利用 Flink DataStream API 和 DataSet API 进行更低层次的控制。此外，Flink 的 Gelly 库为基于批量数据集的大规模高性能图分析提供了算法和构建模块支持。

3) 数据管道应用

提取-转换-加载（ETL）是一种在存储系统之间进行数据转换和迁移的常用方法。ETL 作业通常会周期性地触发，将数据从事务型数据库拷贝到分析型数据库或数据仓库。

数据管道和 ETL 作业的用途相似，都可以转换、丰富数据，并将其从某个存储系统移动到另一个。但数据管道是以持续流模式运行，而非周期性触发。因此它支持从一个不断生成数据的源头读取记录，并将它们以低延迟移动到终点。例如：数据管道可以用来监控文件系统目录中的新文件，并将其数据写入事件日志；另一个应用可能会将事件流物化到数据库或增量构建和优化查询索引。

1.4.2 应用Flink的行业

电商和市场营销

数据报表、广告投放

物联网（IOT）

传感器实时数据采集和显示、实时报警，交通运输业

物流配送和服务业

订单状态实时更新、通知信息推送、电信业基站流量调配

银行和金融业

实时结算和通知推送，实时检测异常行为

第2章 Flink快速上手

2.1创建maven项目

1) idea.POM文件中添加需要的依赖: 课件上面有

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

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

3) idea的配置 创建目录和基本设置

<!--    1. 编码 file encoding
        2. 换行符 code style
        3. maven
            更新快照 选中
​
            不要托管给maven 不选
​
        4. 创建项目的时候, 万万不要把项目存入到中文路径
​
        5. maven仓库也不要存到中文路径-->

2.2批处理WordCount 也是有界流

public class WcBounded {
​
    public static void main(String[] args) {
​
        // 1. 创建流StreamExecutionEnvironment的执行环境  自动根据当前运行的环境获取合适的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置通过此环境执行的操作的并行度。设置平行度
        env.setParallelism(1);
​
        // 2. 通过环境从数据源(source)获取一个流
        DataStreamSource<String> source = env.readTextFile("input/words.txt");
​
        // 3. 对流做各种转换 ,单词都是String的类型
        SingleOutputStreamOperator<String> wordStream= source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> out) throws Exception {
                for (String word : line.split(" ")) {
                    out.collect(word); // 把单词放入到后序的流中
                }
​
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Long>> map = wordStream.map(new MapFunction<String, Tuple2<String, Long>>()
          {
              @Override
              public Tuple2<String, Long> map(String word) throws Exception {
                  return Tuple2.of(word, 1L);
              }
          }
        );
        KeyedStream<Tuple2<String, Long>, String> tuple2StringKeyedStream = map.keyBy(new KeySelector<Tuple2<String, Long>, String>() {
            // 返回每个元素的key
            @Override
            public String getKey(Tuple2<String, Long> value) throws Exception {
                return value.f0;
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Long>> sumStream = tuple2StringKeyedStream.sum(1);
​
        // 4. 把流输出(sink)
        sumStream.print();
​
        // 5. 执行 执行环境
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

2.3流处理WordCount 也是无界流

就是读取的方式不一样,这个是获取无限的数据过来

public class WcUnBounded {
​
    public static void main(String[] args) {
​
        // 1. 创建流StreamExecutionEnvironment的执行环境  自动根据当前运行的环境获取合适的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置通过此环境执行的操作的并行度。设置平行度
        env.setParallelism(1);
​
        // 2. 通过环境从数据源(source)获取一个流
        DataStreamSource<String> source = env.socketTextStream("hadoop162", 9999);
        // 3. 对流做各种转换 ,单词都是String的类型
        SingleOutputStreamOperator<String> wordStream= source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> out) throws Exception {
                for (String word : line.split(" ")) {
                    out.collect(word); // 把单词放入到后序的流中
                }
​
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Long>> map = wordStream.map(new MapFunction<String, Tuple2<String, Long>>()
          {
              @Override
              public Tuple2<String, Long> map(String word) throws Exception {
                  return Tuple2.of(word, 1L);
              }
          }
        );
        KeyedStream<Tuple2<String, Long>, String> tuple2StringKeyedStream = map.keyBy(new KeySelector<Tuple2<String, Long>, String>() {
            // 返回每个元素的key
            @Override
            public String getKey(Tuple2<String, Long> value) throws Exception {
                return value.f0;
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Long>> sumStream = tuple2StringKeyedStream.sum(1);
​
        // 4. 把流输出(sink)
        sumStream.print();
​
        // 5. 执行 执行环境
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

==2.4 拉姆达表达式和泛型擦除==

1) 例子 1 :new的接口A 实现类是匿名内部类

public class Lambda {
​
    public static void main(String[] args) {
​
        //实现类是匿名内部类
        A a = n -> System.out.println("匿名内部类的方法" + n);
​
        A a2 = new A() {
            @Override
            public void fun(int n) {
                System.out.println("匿名内部类的方法" + n);
            }
        };
        a2.fun(20);
​
        a.fun(10);
    }
}

2) 例子 2 传的参数是接口

public class Lambda2 {
    /*
java中的函数: lambda表达式(拉姆达表达式)
​
接口:
    interface
        常量
        抽象方法
        默认方法(default)
​
函数式接口:
    如果一个接口中只有一个抽象方法, 这样的接口就是函数式接口,
    而且要继承的父类也只有一个相同的抽象方法
​
    如果一个类型是函数式接口类型, 则创建对象的时候可以使用lambda表示
​
 */
​
    public static void main(String[] args) {
​
        test2(new Bb() {
            @Override
            public void fun() {
                System.out.println("匿名实现内部类");
            }
        });
​
        test2(() -> System.out.println("匿名实现内部类2"));
​
        test3(new c() {
            @Override
            public String fun(User2 user) {//形参传了一个类进来
              return user.getName();
            }
        });
        test3(user -> user.getName());
        // 方法引用: 类名::方法名 , 是对特殊lambda表达式的替换,就只有一种方法的时候
        test3(User2::getName);
    }
​
    public static void test3(c F){
        String name = F.fun(new User2("lisi"));
        System.out.println(name);
    }
​
​
​
    public static void test2(Bb f){
        f.fun();
    }
}
​
​
​
interface c{
    String fun(User2 user);
}
​
class User2{
    private String name;
​
    public User2(String name) {
        this.name = name;
    }
​
    public String getName() {
        return name;
    }
​
    public void setName(String name) {
        this.name = name;
    }
}
​
​
interface Bb {
    void fun();
}

3) keyby可以写拉姆达表达式,其他类型写不方便

public class WcUnBoundLambda {
​
    public static void main(String[] args) {
​
        // 1. 创建流StreamExecutionEnvironment的执行环境  自动根据当前运行的环境获取合适的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置通过此环境执行的操作的并行度。设置平行度
        env.setParallelism(1);
​
        // 2. 通过环境从数据源(source)获取一个流
        DataStreamSource<String> source = env.socketTextStream("hadoop162", 9999);
        // 3. 对流做各种转换 ,单词都是String的类型
        SingleOutputStreamOperator<String> wordStream= source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> out) throws Exception {
                for (String word : line.split(" ")) {
                    out.collect(word); // 把单词放入到后序的流中
                }
​
            }
        });
​
        // 泛型擦除
        // 泛型只存在于源码中, 编译成字节码之后泛型是不存在, 这就叫泛型擦除. 在java中正常!!!!
        // 在flink中如果由于泛型擦除的存在导致flink无法知道流中的数据的类型, 则不允许
        SingleOutputStreamOperator<Tuple2<String, Long>> map = wordStream
                .map((MapFunction<String, Tuple2<String, Long>>) word -> Tuple2.of(word, 1L))
                // 其他需要用到就要-->明确指定类型
                .returns(Types.TUPLE(Types.STRING,Types.LONG))
        ;
        // 返回每个元素的key
        KeyedStream<Tuple2<String, Long>, String> tuple2StringKeyedStream =
                // 只在这个方法内使用lambda表达式,因为泛型擦除的在flink中,但是在keyby不会数据转换成object
                map.keyBy((KeySelector<Tuple2<String, Long>, String>) value -> value.f0);
​
        SingleOutputStreamOperator<Tuple2<String, Long>> sumStream = tuple2StringKeyedStream.sum(1);
​
        // 4. 把流输出(sink)
        sumStream.print();
​
        // 5. 执行 执行环境
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

第3章 Flink部署

master 在flink中叫 JobManager

worker 在flink中叫 TaskManager

3.1 开发模式

咱们前面在idea中运行Flink程序的方式就是开发模式.

3.2 local-cluster模式

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

3.2.1 local-cluster模式配置

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

1) 上传Flink的安装包flink-1.13.1-bin-scala_2.12.tgz到hadoop162

2) 解压

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

cp -r flink-1.13.1 flink-local

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

1) 打包idea中的应用

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

3) 启动本地集群

bin/start-cluster.sh

4) 在hadoop162中启动netcat,相当于一个服务端,无界流的数据过来

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

5) 命令行提交Flink应用

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

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

http://hadoop162:8081

7) 也可以在WEB UI提交应用

3.3 Standalone模式

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

# 3.3.1  Standalone模式配置
1.复制flink-standalone
cp -r flink-1.13.1 flink-standalone
​
2.修改配置文件:flink-conf.yaml
jobmanager.rpc.address: hadoop162
​
3.修改配置文件:workers
hadoop163
hadoop164
​
4.分发flink-standalone到其他节点
​
# 3.3.2  Standalone模式运行无界流WorkCount
1.启动standalone集群 先起集群, 多个job共享这个集群
bin/start-cluster.sh
​
2.命令行提交Flink应用   bin/flink run -d -m hadoop162:8081 -c 主类  ...jar
​
bin/flink run -m hadoop162:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream 
./flink-prepare-1.0-SNAPSHOT.jar
​
3.查看执行情况与本地集群一致.
​
4.也支持Web UI界面提交Flink应用
​
# 3.3.3Standalone高可用(HA)
任何时候都有一个 主 JobManager 和多个备用 JobManagers，以便在主节点失败时有备用 JobManagers 来接管集群。这保证了没有单点故障，一旦备 JobManager 接管集群，作业就可以正常运行。主备 JobManager 实例之间没有明显的区别。每个 JobManager 都可以充当主备节点。
​
1:修改配置文件: flink-conf.yaml
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop162:8020/flink/standalone/ha
high-availability.zookeeper.quorum: hadoop162:2181,hadoop163:2181,hadoop164:2181
high-availability.zookeeper.path.root: /flink-standalone
high-availability.cluster-id: /cluster_atguigu
​
2:修改配置文件: masters
hadoop162:8081
hadoop163:8081
​
3:分发修改的后配置文件到其他节点
​
4:在/etc/profile.d/my.sh中配置环境变量
Flink没有依赖这个环境,所以需要配置一个路径才可以找到hadoop命令,但是spark有
export HADOOP_CLASSPATH=`hadoop classpath`
分发到其他节点
​
5:首先启动dfs集群和zookeeper集群
​
6:启动standalone HA集群
bin/start-cluster.sh
​
7:可以分别访问
http://hadoop162:8081
http://hadoop163:8081
​
8:可以在zookeeper上面看 
rest_server_lock这上面就可以看到
​
9:杀死hadoop162上的Jobmanager, 再看leader

3.4 Yarn模式

1) 独立部署（Standalone）模式由Flink自身提供计算资源，无需其他框架提供资源，这种方式降低了和其他第三方资源框架的耦合性，独立性非常强。但是你也要记住，Flink主要是计算框架，而不是资源调度框架，所以本身提供的资源调度并不是它的强项，所以还是和其他专业的资源调度框架集成更靠谱，所以接下来我们来学习在强大的Yarn环境中Flink是如何使用的。（其实是因为在国内工作中，Yarn使用的非常多）

2) 把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.13.6 flink-yarn

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

在/etc/profile.d/my.sh中配置

export HADOOP_CLASSPATH=`hadoop classpath`
Flink没有依赖这个环境,所以需要配置一个路径才可以找到hadoop命令,但是spark有

3.4.2 Flink on Yarn的3种部署模式

1: Session-Cluster模式执行无界流WordCount

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

1) 启动一个Flink-Session 这是这种模式的玩法

bin/yarn-session.sh -d 

2) 在Session上运行Job,会自动去找你启动的程序 -d 看不到日志,进程在后台

bin/flink run -d -c 主类 jar包

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

会自动找到你的yarn-session启动的Flink集群.也可以手动指定你的yarn-session集群:

我不想手动指定

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

1:  一个Job会对应一个Flink集群，每提交一个作业会根据自身的情况，都会单独向yarn申请资源，直到作业执行完成，一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager，按需接受资源申请；适合规模大长时间运行的作业。
​
2:  每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成之后创建的集群也会消失。在客户端执行
​
3:  每启动一个job启动一个集群, main函数在客户端执行,会话随着job的创建和消失
​
4:启动下面的命令会直接创建一个yarn会话 bin/flink run -d -t yarn-per-job -c 主类 jar包
bin/flink run -d -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream 
./flink-prepare-1.0-SNAPSHOT.jar

3: Application Mode模式执行无界流WordCount

1:  Application Mode会在Yarn上启动集群, 应用jar包的main函数(用户类的main函数)将会在JobManager上执行. 只要应用程序执行结束, Flink集群会马上被关闭. 也可以手动停止集群.
​
2:  与Per-Job-Cluster的区别: 就是Application Mode下, 用户的main函数是在集群中(job manager)执行的
3:  官方建议:
    出于生产的需求, 我们建议使用Per-job or Application Mode,因为他们给应用提供了更好的隔离!
4:  每启动一个job启动一个集群    main函数在JobManager执行
​
5:  bin/flink run-application -d -t yarn-application -c 主类 jar包
6:  开始操作
bin/flink run-application -t yarn-application -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream 
./flink-prepare-1.0-SNAPSHOT.jar

3.4.3 Yarn模式高可用

1) Yarn模式的高可用和Standalone模式的高可用原理不一样.

2) Standalone模式中, 同时启动多个Jobmanager, 一个为leader其他为standby, 当leader挂了, 其他的才会有一个成为leader.

3) yarn的高可用是同时只启动一个Jobmanager, 当这个Jobmanager挂了之后, yarn会再次启动一个, 其实是利用的yarn的重试次数来实现的高可用.

4) 在yarn-site.xml中配置 注意: 配置完不要忘记分发, 和重启yarn

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

5) 在flink-conf.yaml中配置

yarn.application-attempts: 3
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop162:8020/flink/yarn/ha
high-availability.zookeeper.quorum: hadoop162:2181,hadoop163:2181,hadoop164:2181
high-availability.zookeeper.path.root: /flink-yarn

6) 启动yarn-session

7) 杀死Jobmanager, 查看他的复活情况

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

注意: 在10秒之内重启三次要是没有完成,还是杀不死yarn-session,这是flink默认设置这个,怕一个程序执行时间需要3天,中断了,但是因为你设置的这个程序不能重启了,所以设置了10秒内完成你设置的这个

3.5 K8S & Mesos模式

容器化部署时目前业界很流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。这里我们也不做过多的讲解.

Mesos是Apache下的开源分布式资源管理框架，就和yarn一样的,它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署，但是在国内，依然使用着传统的Hadoop大数据框架，所以国内使用mesos框架的并不多，这里我们就不做过多讲解了。

第4章Flink运行架构

4.1运行架构

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

2) TaskManager 是一个进程 --> 有多个 slots (线程,封装cpu和内存) --> 1个slots有多个task

4.1.1客户端

严格上说, 客户端不是运行和程序执行的一部分, 而是用于准备和发送dataflow到JobManager. 然后客户端可以断开与JobManager的连接(detached mode), 也可以继续保持与JobManager的连接(attached mode)

客户端作为触发执行的java或者scala代码的一部分运行, 也可以在命令行运行:bin/flink run ...

4.1.2 JobManager

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

4.1.2.1 ResourceManager

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

2) 主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger插槽是Flink中定义的处理资源单元。

3) 当JobManager申请插槽资源时，ResourceManager会将有空闲插槽的TaskManager分配给JobManager。如果ResourceManager没有足够的插槽来满足JobManager的请求，它还可以向资源提供平台发起会话，以提供启动TaskManager进程的容器。另外，ResourceManager还负责终止空闲的TaskManager，释放计算资源。

4.1.2.2 Dispatcher

负责接收用户提供的作业，并且负责为这个新提交的作业启动一个新的JobMaster 组件. Dispatcher也会启动一个Web UI，用来方便地展示和监控作业执行的信息。Dispatcher在架构中可能并不是必需的，这取决于应用提交运行的方式。

4.1.2.3 JobMaster

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

4.1.3 TaskManager

1) Flink中的工作进程。通常在Flink中会有多个TaskManager运行，每一个TaskManager都包含了一定数量的插槽（slots）。插槽的数量限制了TaskManager能够执行的任务数量。

2) 启动之后，TaskManager会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager就会将一个或者多个插槽提供给JobManager调用。JobManager就可以向插槽分配任务（tasks）来执行了。

3) 在执行过程中，一个TaskManager可以跟其它运行同一应用程序的TaskManager交换数据。

4.2 核心概念

4.2.1 TaskManager与Slots

1) Flink中每一个worker(TaskManager)都是一个JVM进程，它可能会在独立的线程上执行一个Task。为了控制一个worker能接收多少个task，worker通过Task Slot来进行控制（一个worker至少有一个Task Slot）。

2) 这里的Slot如何来理解呢？很多的文章中经常会和Spark框架进行类比，将Slot类比为Core，其实简单这么类比是可以的，可实际上，可以考虑下，当Spark申请资源后，这个Core执行任务时有可能是空闲的，但是这个时候Spark并不能将这个空闲下来的Core共享给其他Job使用，所以这里的Core是Job内部共享使用的。接下来我们再回想一下，之前在Yarn Session-Cluster模式时，其实是可以并行执行多个Job的，那如果申请两个Slot，而执行Job时，只用到了一个，剩下的一个怎么办？那我们自认而然就会想到可以将这个Slot给并行的其他Job，对吗？所以Flink中的Slot和Spark中的Core还是有很大区别的。

3) 每个task slot表示TaskManager拥有资源的一个固定大小的子集。假如一个TaskManager有三个slot，那么它会将其管理的内存分成三份给各个slot。资源slot化意味着一个task将不需要跟来自其他job的task竞争被管理的内存，取而代之的是它将拥有一定数量的内存储备。需要注意的是，这里不会涉及到CPU的隔离，slot目前仅仅用来隔离task的受管理的内存。

4.2.2 Parallelism（并行度）

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

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

1) One-to-one： 1对1的关系 ,且并行度相同 stream(比如在source和map operator之间)维护着分区以及元素的顺序。那意味着flatmap 算子的子任务看到的元素的个数以及顺序跟source 算子的子任务生产的元素的个数、顺序相同，map、fliter、flatMap等算子都是one-to-one的对应关系。类似于spark中的窄依赖 2) Redistributing：分区数据发生了改变, stream(map()跟keyBy/window之间或者keyBy/window跟sink之间)的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。例如，keyBy()基于hashCode重分区、broadcast和rebalance会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似于Spark中的shuffle过程。类似于spark中的宽依赖

4.2.3 Task与SubTask

一个算子就是一个Task. 一个算子的并行度是几, 这个Task就有几个SubTask

4.2.4 Operator Chains（优化）

1: 操作链的优化是自动实现:
    算子与算子之间: one-to-one
    算子的并行度一样
​
-- 1: .startNewChain()
    开启一个新链: 当前算子不会和前面的优化在一起 
    如果后面的算子满足优化条件, 也会与当前算子优化到一起
    
-- 2: .disableChaining()
    当前算子不参与任何链的优化
    
--3:  env.disableOperatorChaining()
    当前应用所有算子都不进行优化 
 实际生产环境下, 尽量不要禁止优化: 优化只有好处没有坏处
​
2: 在flink中, 有4种办法设置并行度
​
-- 1. 在flink-conf.yaml 配置文件中设置算子的默认并行度
​
        parallelism.default: 1
        
-- 2. 提交job的是时候通过参数设置
    bin/flink run -d -p 2 ...
​
-- 3. 在代码中通过环境设置
    env.setParallelism(1);  
    注意: 在flink中socket这个source他的并行度只能是1.
    
-- 4. 单独给每个算子设置并行度
    .setParallelism(3)
​

演示代码:

  public class Flink02_WC_UnBounded {
    public static void main(String[] args) {
        System.out.println("Flink02_WC_UnBounded.main");
        // 1. 创建流的执行环境  自动根据当前运行的环境获取合适的执行环境
        
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        //env.disableOperatorChaining();   当前应用所有算子都不进行优化
        // 2. 通过环境从数据源(source)获取一个流
        DataStreamSource<String> source = env.socketTextStream("hadoop162", 9999);
        // 3. 对流做各种转换
        SingleOutputStreamOperator<String> wordStream = source
            .flatMap(new FlatMapFunction<String, String>() {
                @Override
                public void flatMap(String line, Collector<String> out) throws Exception {
                    for (String word : line.split(" ")) {
                        out.collect(word);  // 把单词放入到后序的流中
                    }
                }
            });
        
        SingleOutputStreamOperator<Tuple2<String, Long>> wordOneStream = wordStream
            .map(new MapFunction<String, Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> map(String word) throws Exception {
                    return Tuple2.of(word, 1L);
                }
            })
            //.startNewChain()
            // .disableChaining()
            .filter(t -> true); // 开一起一个新链
        
        
        KeyedStream<Tuple2<String, Long>, String> keyedStream = wordOneStream.keyBy(new KeySelector<Tuple2<String, Long>, String>() {
            // 返回每个元素的key
            @Override
            public String getKey(Tuple2<String, Long> t) throws Exception {
                return t.f0;
            }
        });
        // 对元组中某个位置进行聚合
        SingleOutputStreamOperator<Tuple2<String, Long>> resultStream = keyedStream.sum(1);
        
        
        // 4. 把流输出(sink)
        resultStream.print();
        
        // 5. 执行 执行环境
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}  

4.2.5 ExecutionGraph（执行图）

由Flink程序直接映射成的数据流图是StreamGraph，也被称为"逻辑流图"，因为它们表示的是计算逻辑的高级视图。为了执行一个流处理程序，Flink需要将逻辑流图转换为物理数据流图（也叫执行图），详细说明程序的执行方式。
Flink 中的执行图可以分成四层：StreamGraph -> JobGraph -> ExecutionGraph -> Physical Graph。
-- 1: StreamGraph：
是根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。
-- 2: JobGraph：
StreamGraph经过优化后生成了 JobGraph，是提交给 JobManager 的数据结构。主要的优化为: 将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。
-- 3: ExecutionGraph：
JobManager 根据 JobGraph 生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。
--4: Physical Graph：
JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个TaskManager 上部署 Task 后形成的“图”，并不是一个具体的数据结构。
2个并发度（Source为1个并发度）的 SocketTextStreamWordCount 四层执行图的演变过程

4.3 提交流程

4.3.1 通用提交流程

4.3.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向其分配任务。

第5章 Flink流处理核心编程

5.1 Environment

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

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

5.2 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 从文件读取数据

// 1:从HDFS idea读取文件
DataStreamSource<String> source=env.readTextFile("hdfs://hadoop162:8020/words.txt");
DataStreamSource<String> source=env.readTextFile("input/out.txt"); 

5.2.3 从Java的集合中读取数据

List<String> list = Arrays.asList("a", "aa", "b", "c", "d");
DataStreamSource<String> source = env.fromCollection(list);
    
DataStreamSource<String> source = env.fromElements("a", "b", "c", "d");

5.2.4 从Socket读取数据

  DataStreamSource<String> source = env.socketTextStream("hadoop162", 9999);

5.2.5 从Kafka读取数据

1) 依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>1.13.1</version>
</dependency>
2: hadoop的支持依赖,flink没有自带
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.1.3</version>
    <scope>provided</scope>
</dependency>

2) 代码实现

第一种从kafka消费方式

第二种从kafka消费方式

5.2.6 自定义Source

public class Source_Custom {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
​
        env.addSource(new MySocket()).print();
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
​
    }
    
    public static class MySocket implements SourceFunction<WaterSensor> {
        // 自定义一个可以从socket读取数据的source
​
        boolean isStop = false;
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            Socket socket = new Socket("hadoop162", 9999);
            //字节流转换为字符流InputStreamReader,在用buffer封装BufferedReade
            InputStream is = socket.getInputStream();
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8));
            String line = bufferedReader.readLine();
​
            while (!isStop && line != null) {
                //对读到一行数据做处理
                String[] data = line.split(",");
​
                //把数据放入流中
                ctx.collect(new WaterSensor(data[0],
                                Long.valueOf(data[1]),
                                Integer.valueOf(data[2])
                        )
                );
                line = bufferedReader.readLine();
​
            }
        }
        // 取消: 关闭source
        // 这个方法不会自动调用, 如果有需要的时候, 外部可以通过调用这个方法实现关闭source
        @Override
        public void cancel() {
​
        }
    }
}

5.3 Transform

5.3.1 map和 Rich Map

每个算子都有new Rich的 function, 减少和数据库的连接

public class Map_01 {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(3);
​
        //1: 转换算子 map: 一对一的关系
        DataStreamSource<Integer> source = env.fromElements(1, 6, 9, 20, 10, 2);
        source.map(new MapFunction<Integer, Integer>() {
            @Override
            public Integer map(Integer value) throws Exception {
                return value  ;
            }
        });//.print();
​
        //1.2 Map_Rich
            /*
​
            如果一个元素建立一个连接, 连接数过多, 效率太低
​
            flink, 应该是一个并行度建立一个到数据库的连接
            连接建立的时机: 应该所有的数据道来之前建立
​
            连接关闭: 等到所有的数据都处理完了, 或者程序停止之前关闭
            */
          // 需要查询数据库才能确定如何对value做什么样的操作
        env.fromElements(1, 6, 9, 20, 10, 2)
                .map(new RichMapFunction<Integer, Integer>() {
​
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //程序启动的时候调用. 这个方法, 是每个并行度执行一次, 一般在这里执行创建连接, 开辟资源
                        System.out.println("Flink02_Map_Rich.open");
                    }
​
                    @Override
                    public void close() throws Exception {
                        // 程序终止的时候的调用. 每个并行度执行一次, 关闭连接, 是否资源
                        System.out.println("Flink02_Map_Rich.close");
                    }
​
                    @Override
                    public Integer map(Integer value) throws Exception {
                      //  System.out.println("map" + value);
                        // 建立到数据库的连接
​
                        // 查询
​
                        // 得到结果
​
                        // 关闭数据库的连接
                        return value * value;
​
                    }
                })
        .print();
​
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

5.3.2 filter

5.3.3 flatMap

5.3.4 keyBy

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

keyby --> shuffle --> rebalance --> rescale

5.3.6 split和select

被移除了

5.3.7 connect 和 union

5.3.8 sum简单聚合算子

public class Sum {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
​
        //10:sum: 统计每个传感器的水位和,一定要先keyBy
        env.readTextFile("input/sensor.txt")
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] data = value.split(",");
​
                        WaterSensor waterSensor = new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2])
                        );
                        return waterSensor;
                    }
                    })
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                })
                .sum("vc");
                //.print();
​
​
        /*
        11: max:分组聚合后, 其他字段默认情况是取的是这个组内第一个元素的字段值
            min:
            maxBy:其他字段是否取第一个? false代表不取第一个,取最大最小值同一行的一个
            minBy:
            以上4个聚合+sum聚合: 1)只能对其中的一个字段聚合  2)只能聚合数字类型
         */
        env.readTextFile("input/sensor.txt")
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] data = value.split(",");
​
                        WaterSensor waterSensor = new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2])
                        );
                        return waterSensor;
                    }
                })
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                })
                //.max("vc")
                //.min("vc")
                .minBy("vc",false) //false代表不取第一个,取最大最小同一行的一个
                .print();
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

5.3.9 reduce

==reduce和状态的reduce是不一样的,一定要区分==

public class Reduce {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
​
        env
            .readTextFile("input/sensor.txt")
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
​
                    WaterSensor waterSensor = new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2])
                    );
                    return waterSensor;
                }
            })
            .keyBy(WaterSensor::getId)
            .reduce(new ReduceFunction<WaterSensor>() {
                /*12: reduce:怎么聚合? reduce还是不如上面几个简单
                      1) 聚合后的类型和聚合前的类型一致
                      2) 可以同时聚合对个字段, 要聚合的字段类型无所谓
​
                 参数1: 上次聚合的结果  参数2: 这次需要聚合的元素
                 当某个key第一个数据进来是, 这个方法根本就不执行, 直接输出结果
                 */
                @Override
                public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
​
                    // ts是第一个
                    //System.out.println(value1.getId() + "   ");
                    //value1.setVc(value1.getVc() + value2.getVc());
​
                    //ts是计算到自己的那个
                    //value2.setVc(value1.getVc() + value2.getVc());
                    //return value2;
​
                    //求最大值最小值也可以
                    value1.setVc(Math.max(value1.getVc(),value2.getVc()));
                    return value1;
                }
            })
             .print();
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

5.3.10 process

public class Process {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
​
        //13.1: Process(没有key的计算) 统计每个传感器的水位和 ,没有keyBy的功能
        env
            .readTextFile("input/sensor.txt")
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] data = value.split(",");
                        WaterSensor waterSensor = new WaterSensor(data[0],
                                Long.valueOf(data[1]),
                                Integer.valueOf(data[2]));
                        return waterSensor;
                    }
                })
                .process(new ProcessFunction<WaterSensor, Integer>() {
                    int sum = 0;
                    // 处理流中的数据
                    // 参数1: 就是要处理的元素
                    // 参数2: 上下文对象
                    // 参数3: 输出数据
                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<Integer> out) throws Exception {
​
                       sum += value.getVc();
                       out.collect(sum);
                    }
                });
                //.print();
​
        //13.2: Process(有key的计算) 统计每个传感器的水位和 ,没有keyBy的功能
​
        env
                .readTextFile("input/sensor.txt")
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] data = value.split(",");
                        WaterSensor waterSensor = new WaterSensor(data[0],
                                Long.valueOf(data[1]),
                                Integer.valueOf(data[2]));
                        return waterSensor;
                    }
                })
                .keyBy(WaterSensor::getId).print();
​
//                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
//
//                    int sum = 0; // 不分key, 和并行度有关
//                    @Override
//                    public void processElement(WaterSensor value,
//                                               Context ctx,
//                                               Collector<String> out) throws Exception {
//
//                        sum += value.getVc();
//                        out.collect(ctx.getCurrentKey() + "的水位是" +sum);
//
//                    }
//                });
                //.print();
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

5.4 Sink

5.4.1 KafkaSink

1) 添加Kafka Connector依赖

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

2) sink to kafka 官方和定义两种

public class SingToKafka {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
​
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092");
        env
            .readTextFile("input/words.txt")
            .flatMap(new FlatMapFunction<String, Tuple2<String,Long>>() {
                @Override
                public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                    for (String word : value.split(" ")) {
                        out.collect(Tuple2.of(word,1L));
                    }
                }
            })
                //分组好像什么现象不改变
            .keyBy(new KeySelector<Tuple2<String, Long>, String>() {
                @Override
                public String getKey(Tuple2<String, Long> value) throws Exception {
                    return value.f0;
                }
            })
                .sum(1)
//                .map(new MapFunction<Tuple2<String, Long>, String>() {
//                    @Override
//                    public String map(Tuple2<String, Long> value) throws Exception {
//                        return value.f0 + "_" + value.f1;
//                    }
//                })
                 /*  1: 利用配置把数据写到kafka
                  参数1: 写出的topi
                  参数2: 序列化器
                  参数2: kafka的配置信息
                */
        //.addSink(new FlinkKafkaProducer<String>("s1",new SimpleStringSchema(),props));
​
         // 2:自定义sink to kafka,自己自定义一个生产者
                .addSink(new FlinkKafkaProducer<Tuple2<String, Long>>(
                        "defalut", // 这个topic没用
                        new KafkaSerializationSchema<Tuple2<String, Long>>() {
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(Tuple2<String, Long> element,
                                                                            @Nullable Long timestamp) {
                                String msg = element.f0 + "_" + element.f1;
                                return new ProducerRecord<>("s1", msg.getBytes(StandardCharsets.UTF_8));
                            }
                        },
                        props,
                        AT_LEAST_ONCE  // 写出的一致性语义: 目前只能选至少一次
        ));
​
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

5.4.2 RedisSink

1) 添加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>

2) 代码实现,看官方文档

/*
redis的数据结构: 你可以多种演示
1) string
​
2) list
 key            value
 sensor_1       [json格式字符串, ...]
​
3) set
​
4) hash
 key        field        value
 "s"        "sensor_1"    json格式字符串
 
5) zset
 */
public class SinkToRedis {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
​
​
        //FlinkJedisPoolConfig redisConf = new FlinkJedisPoolConfig.Builder()
        FlinkJedisPoolConfig redisConf = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop162")
                .setPort(6379)
//            .setDatabase(1) // 选择数据库
                .setMaxTotal(100)// 连接提供的最大连接数
                .setMaxIdle(10)  // 最大空闲连接数
                .setMinIdle(2)  // 最小空闲连接数
                .build();//设置参数最后返回这些值了
​
        env
            .readTextFile("input/sensor.txt")
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2]));
                    }
                })
                .keyBy(WaterSensor::getId)
                .sum("vc")
                .addSink(new RedisSink<>(redisConf, new RedisMapper<WaterSensor>() {
​
                    // 返回一个命令描述符   字符串: set list: rpush lpush ...
                    @Override
                    public RedisCommandDescription getCommandDescription() {
                        // 参数1: 只能命令(确定要写的数据类型)
                        // 参数2: 额外的key, 当写出的的类型是 hash和zset的时候有效(他们会包几个)
                        return new RedisCommandDescription(RedisCommand.SET,null);
​
                    }
​
                    // 从数据中返回key. 返回的这条数据写入的时候的key
                    @Override
                    public String getKeyFromData(WaterSensor data) {
                        // 用sensor id 作为每条数据的key
                        return data.getId();
                    }
​
                    // 返回写入的具体数据
                    @Override
                    public String getValueFromData(WaterSensor data) {
                       return JSON.toJSONString(data);
                    }
                }));
​
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

5.4.3 ElasticsearchSink

1) 添加Elasticsearch Connector依赖

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch6 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch6_2.12</artifactId>
    <version>1.13.1</version>
</dependency>

2) 代码实现 idea有

5.4.4 自定义Sink to mysql

1) 在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;

2) 导入Mysql驱动

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

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

public class Flink05_Sink_Custom {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
          
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            
            .keyBy(WaterSensor::getId)  // 方法引用
            .sum("vc")
            .addSink(new MySqlSink());
            
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public static class MySqlSink extends RichSinkFunction<WaterSensor>{
    
        private Connection connection;
    
        @Override
        public void open(Configuration parameters) throws Exception {
            // 获取mysql 连接
            // 1. 加载驱动
            Class.forName("com.mysql.jdbc.Driver");
            // 2. 获取连接 提升成员变量快捷键: alt+ctrl+f
            connection = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test?useSSL=false", "root", "aaaaaa");
        }
    
        @Override
        public void close() throws Exception {
            if (connection != null) {
                connection.close();
            }
        }
    
        @Override
        public void invoke(WaterSensor ws, Context context) throws Exception {
            // 插入数据
            // sql语句
//            String sql = "insert into sensor(id, ts, vc)values(?,?,?)";
            
            // 如果主键不重复就新增, 重复就更新
//            String sql = "insert into sensor(id, ts, vc)values(?,?,?) on duplicate key update vc=?";
            String sql = "replace into sensor(id, ts, vc)values(?,?,?)";  // 等价上面这个
            PreparedStatement ps = connection.prepareStatement(sql);
            // 给sql中的占位符赋值
            ps.setString(1, ws.getId());
            ps.setLong(2, ws.getTs());
            ps.setInt(3, ws.getVc());
//            ps.setInt(4, ws.getVc());
            
            ps.execute();
            connection.commit();  // 提交执行. 当自动提交是true的时候,不能调用这个方法. mysql的自动提交就是true
            ps.close();
            
        }
    }
}

5.5 流和批的执行模式

-- 真正流批一体 流模式: 有界流 无界流
1) 有界流自动选择批(我刚刚测试的就自动选择了批)
2) 无界流不能设置批模式
3) 批模式不能设置时间

-- 1:执行模式有3个选择可配:
1) STREAMING(默认)
2) BATCH
3) AUTOMATIC
自动;如果数据离线数据, 则自动选择批处理, 如果是流数据, 则自动选流处理

-- 2: 配置
1) 通过命令行配置
bin/flink run -Dexecution.runtime-mode=BATCH ...

2) 通过代码配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
建议: 不要在运行时配置(代码中配置), 而是使用命令行配置, 因为这样会更灵活: 同一个应用即可以用于无界数据也可以用于有界数据

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

2) 批处理模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

3) 自动模式
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

-- 4: 注意
1.12.0版批模式下有个bug, 如果单词只有1个, 则不会输出.
1.13开始, 这个bug已经解决了

第6章 Flink的Transform算子的运用

6.1 基于埋点日志网络流量统计

6.1.1 pv的计算

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

1) 封装一个javaBean 这里面会有getset方法 还有构造器 方法

@Data
@NoArgsConstructor
@AllArgsConstructor
public class UserBehavior {
    private Long userId;
    private Long itemId;
    private Integer categoryId;
    private String behavior;
    private Long timestamp;
}

2) pv实现思路1: WordCount

public class F01_Project_PV {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env
            .readTextFile("input/UserBehavior.csv")
            .map(new MapFunction<String, UserBehavior>() {
                @Override
                public UserBehavior map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new UserBehavior(Long.valueOf(data[0]),
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2]),
                            data[3],
                            Long.valueOf(data[4]));
                }
            })
            .filter(value -> "pv".equals(value.getBehavior()))
            //先过完所有的在封装成元组计算wordcount
            .map(new MapFunction<UserBehavior, Tuple2<String,Long>>() {
                @Override
                public Tuple2<String, Long> map(UserBehavior value) throws Exception {
                    return Tuple2.of("pv",1L);
                }
            })

            //使用sum之前一定要聚合
            .keyBy(new KeySelector<Tuple2<String, Long>, String>() {
                @Override
                public String getKey(Tuple2<String, Long> value) throws Exception {
                    return value.f0;
                }
            })
            .sum(1)
            .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3) process实现wordcount

public class F01_Project_PV_2 {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
​
        env
            .readTextFile("input/UserBehavior.csv")
            .map(new MapFunction<String, UserBehavior>() {
                @Override
                public UserBehavior map(String line) throws Exception {
                    String[] data = line.split(",");
                    return new UserBehavior(
                            Long.valueOf(data[0]),
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2]),
                            data[3],
                            Long.valueOf(data[4])
                    );
                }
            })
            .keyBy(UserBehavior::getBehavior)
            .process(new KeyedProcessFunction<String, UserBehavior, String>() {
                int sum = 0 ;
                @Override
                public void processElement(UserBehavior value,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    if ("pv".equals(value.getBehavior())){
                        sum ++;
                        out.collect("pv的值: " + sum);
                    }
​
                }
            })
            .print();
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

6.1.2 uv的计算

1) 去重客户,对pv的数据做一个去重

public class F02_Project_UV {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
​
        env
            .readTextFile("input/UserBehavior.csv")
            .map(new MapFunction<String, UserBehavior>() {
                @Override
                public UserBehavior map(String line) throws Exception {
                    String[] data = line.split(",");
                    return new UserBehavior(
                            Long.valueOf(data[0]),
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2]),
                            data[3],
                            Long.valueOf(data[4])
                    );
                }
            })
            .filter( value -> "pv".equals(value.getBehavior()))
            .keyBy(UserBehavior::getBehavior)
​
            .process(new KeyedProcessFunction<String, UserBehavior, String>() {
           //计算uv的性质,就是计算有不同的客户,客户点击很多次,需要去重, 建议一个set集合就可以解决
​
                HashSet<Long> uidSet = new HashSet<>();
​
                @Override
                public void processElement(UserBehavior value,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
​
                    // 如果set中添加成功就是新用户, 否则就是旧用户
                    if (uidSet.add(value.getUserId())){
                        out.collect("uv的值:" + uidSet.size());
                    }
​
                }
            })
            .print();
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
​
    }
}

6.2 市场营销商业指标统计分析

1) 粒度是 行为 和数据来源一起 的分析 ,可以操作不同的粒度

public class F03_Project_App {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        // 统计不同的行为不同渠道的 数量 两个粒度就一起拼接
        env
                .addSource(new MarketSource2())
                .map(new MapFunction<MarketingUserBehavior, Tuple2<String,Long>>() {
                    @Override
                    public Tuple2<String, Long> map(MarketingUserBehavior value) throws Exception {
                        return Tuple2.of(value.getBehavior() + "_" +value.getChannel(),1L );
                    }
                })
                .keyBy(t -> t.f0)
                .sum(1)
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public static class MarketSource2 implements SourceFunction <MarketingUserBehavior> {

        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            Random random = new Random();
            // app行为: 下载 安装 卸载 更新
            String[] behaviors = {"download", "install", "uninstall", "update"};
            String[] channels = {"huawei", "xiaomi", "oppo", "vivo", "appstore", "zhongxing"};
            while (true) {
                Long userId = Math.abs(random.nextLong()) % 10000 + 1;  // [1, 10000]
                String behavior = behaviors[random.nextInt(behaviors.length)];
                String channel = channels[random.nextInt(channels.length)];
                Long timestamp = System.currentTimeMillis();

                MarketingUserBehavior bean = new MarketingUserBehavior(
                        userId,
                        behavior,
                        channel,
                        timestamp
                );
                ctx.collect(bean);

                Thread.sleep(300);
            }
        }

        @Override
        public void cancel() {

        }
    }
}

6.3 各省份页面广告点击量实时统计

1) 封装数据的JavaBean

@Data
@AllArgsConstructor
@NoArgsConstructor
public class AdsClickLog {
    private Long userId;
    private Long adId;
    private String province;
    private String city;
    private Long timestamp;
}

2) 代码实现 和上年报wordcount是一样的

6.4 订单支付和收款的对账

1) JavaBean类的准备

@Data
@AllArgsConstructor
@NoArgsConstructor
public class OrderEvent {
    private Long orderId;
    private String eventType;
    private String txId;
    private Long eventTime;
}

@Data
@AllArgsConstructor
@NoArgsConstructor
public class TxEvent {
    private String txId;
    private String payChannel;
    private Long eventTime;

2) 需求: 来自两条流的订单交易匹配

1条流是订单的信息,里面有支付的信息,订单

另1条流是商户收款的具体到账没,商家那边支付了,我们这边收到款有这个记录才算成功了

考虑: 两条流的数据有先来后来的问题,我么可以用hashMap的get方法判断是否一一对应

public class F04_Project_Order {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        //需求: 来自两条流的订单交易匹配
        //订单数据 (里面有pay的订单) 两者要对账的   交易数据(支付订单id,微信支付宝,时间)
        SingleOutputStreamOperator<OrderEvent> orderStream = env
            .readTextFile("input/OrderLog.csv")
            .map(new MapFunction<String, OrderEvent>() {
                @Override
                public OrderEvent map(String value) throws Exception {
                    String[] data = value.split(",");
                    return   new OrderEvent(Long.valueOf(data[0]),
                                            data[1],
                                            data[2],
                                            Long.valueOf(data[3]));
                }
            })
            .filter(e -> "pay".equals(e.getEventType()));

        SingleOutputStreamOperator<TxEvent> payStream = env
            .readTextFile("input/ReceiptLog.csv")
            .map(new MapFunction<String, TxEvent>() {
                @Override
                public TxEvent map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new TxEvent(
                        data[0],
                        data[1],
                        Long.valueOf(data[2])
                    );
                }
            });
        orderStream.connect(payStream)
            //一定要keyby数据,要不然可能不在一个并行度就不行
            .keyBy(OrderEvent::getTxId,TxEvent::getTxId)
            .process(new CoProcessFunction<OrderEvent, TxEvent, String>() {

                // 用map集合存储先来的数据
                HashMap<String, OrderEvent> orderEventMap = new HashMap<>();
                HashMap<String, TxEvent> txEventMap = new HashMap<>();


                @Override
                public void processElement1(OrderEvent value,
                                            Context ctx,
                                           Collector<String> out) throws Exception {

                    // 来了支付数据, 去交易对应的map集合中,查找交易数据是否存在

                    TxEvent txEvent = txEventMap.get(value.getTxId());

                    if (txEvent != null) {
                        // 交易数据已经存在
                        out.collect("订单: " + value.getOrderId() + " 对账成功... ");
                    }else{
                        // 把支付数据存入到oderEventMap中
                        orderEventMap.put(value.getTxId(), value);
                    }

                }

                @Override
                public void processElement2(TxEvent value,
                                            Context ctx,
                                           Collector<String> out) throws Exception {
                    OrderEvent orderEvent = orderEventMap.get(value.getTxId());

                    if (orderEvent != null){

                    out.collect("订单: " + orderEvent.getOrderId() + " 对账成功... ");
                    }else{
                        txEventMap.put(value.getTxId(),value);
                    }

                }
            })
            .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

第7章 Flink流处理高阶编程

7.1 Flink的window (窗口)

7.1.1 窗口概述

1) 在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行计算。

2) 流式计算是一种被设计用于处理无限数据集的数据处理引擎，而无限数据集是指一种不断增长的本质上无限的数据集，而Window窗口是一种切割无限数据为有限块进行处理的手段。

3) 在Flink中, 窗口(window)是处理无界流的核心. ==窗口把流切割成有限大小的多个"存储桶"(bucket),== 我们在这些桶上进行计算.

7.1.2 窗口的分类

1: 基于时间的窗口(时间驱动)

时间驱动: 事件时间 (重点) 处理时间

2: 基于元素个数的(数据驱动)

7.1.3 窗口代码实现

7.1.3.1 基于处理时间的窗口

-- 基于处理时间的窗口
    属性: 开始时间和关闭时间  [  )
           1) 滚动 TumblingProcessingTimeWindows
                窗口之间没有间隙也没重叠
                    
           2) 滑动  SlidingProcessingTimeWindows
                窗口之间有间隙也可能重叠
     所以, 如果滑动步长小于窗口长度, 滑动窗口会重叠. 这种情况下, 一个元素可能会被分配到多个窗口中
     例如, 滑动窗口长度10分钟, 滑动步长5分钟, 则, 每5分钟会得到一个包含最近10分钟的数据.
                    
           3)会话  ProcessingTimeSessionWindows
                gap
                超过gap时间没有数据, 窗口关闭
               如果gap在3秒的时间一直有数据,这样的话一直不会被关闭,
             会话窗口分配器会根据活动的元素进行分组. 会话窗口不会有重叠, 与滚动窗口和滑动窗口相比, 会话窗口也没有固定的开启和关闭时间.
         4) 在关闭窗口的时候,不同的key关闭时间是否一样?
            滚动和滑动,是同时关闭 key1 和key2
            session不会同时关闭,不同的key的窗口和关闭时间一般不一样

例子:

public class Flink01_Window_Tumble {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
    
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .keyBy(WaterSensor::getId)
            // 定义一个长度为5s的窗口
//            .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            // 定义一个长度为5, 滑动步长为2窗口
            //.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(2)))
           .window(ProcessingTimeSessionWindows.withGap(Time.seconds(3)))
            // 窗口处理函数
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
            
                // 处理窗口中元素的方法: 当窗口关闭的时候执行一次
                @Override
                public void process(String key,  // 属于哪个key的窗口
                                    Context ctx,  // 上下文对象, 可以获取到窗口信息
                                    Iterable<WaterSensor> elements, // 窗口触发计算的时候, 存储着窗口内所有的元素
                                    Collector<String> out) throws Exception {
                    ArrayList<WaterSensor> list = new ArrayList<>();
                    // Iterable很多操作不能做, 比如排序, 一般是把他转成一个List集合再操作
                    for (WaterSensor ws : elements) {
                        list.add(ws);
                    }
                
                    // 获取窗口的开始时间
                    String stt = AtguiguUtil.toDatTime(ctx.window().getStart());
                    // 获取窗口的结束时间
                    String edt = AtguiguUtil.toDatTime(ctx.window().getEnd());
                
                    out.collect(key + " " + stt + " " + edt + "  " + list);


                }
            })
            .print();
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

7.1.3.2 基于元素个数的窗口

7.1.3.3 没有keyBy加窗口

7.1.4 窗口处理函数

窗口处理函数: 
    1) 增量
        每来一条元素, 就开始计算, 等到窗口关闭的时候, 直接就可以返回最终的结果
        简单
            sum max min maxBy minBy 只能聚合数据
        rduce (是对某个字段的求和 3+10+10)
            输出输出类型一样

        Aggregate (是对任意字段的求和 word 1 word 1)
            由于可以定义累加器(就是可以有中间值), 所以输出和输入可以不一样
            

"额外的功能: 补充窗口的信息:增加processwindow处理函数"

    2) 全量
        process (比较复杂的逻辑都是在这边处理)
            来的元素会先缓存到内存中,等待窗口关闭的时候才开始计算

尽量使用增量处理函数, 如果增量处理不了(排序, topN), 再选择全量

1) 增量代码实现 sum 和reduce

2) 增量代码实现 Aggregate: 求水位平均数

public class Flink05_Window_ProcessFunction_Aggregate {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        env
            .socketTextStream("hadoop162", 8888)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .keyBy(WaterSensor::getId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .aggregate(new AggregateFunction<WaterSensor, Avg, Double>() {
                
               // 初始一个累加器
               // 这个窗口内的第一条元素来的触发一次
               @Override
               public Avg createAccumulator() {
                   System.out.println("Flink05_Window_ProcessFunction_Aggregate.createAccumulator");
                   return new Avg();
               }

               // 真正的累加器方法
               // 每来一条元素触发一次
               @Override
               public Avg add(WaterSensor value, Avg acc) {
                   System.out.println("Flink05_Window_ProcessFunction_Aggregate.add");
                   acc.sum += value.getVc();
                   acc.count++;
                   return acc;
               }

               // 返回最终的结果
               // 窗口关闭的时候触发一次
               @Override
               public Double getResult(Avg acc) {
                   System.out.println("Flink05_Window_ProcessFunction_Aggregate.getResult");
                   return acc.sum * 1.0 / acc.count;
               }

               // 合并累加器
               // 只有窗口是 session窗口的时候才有效,其他窗口不会触发
               @Override
               public Avg merge(Avg a, Avg b) {
                   System.out.println("Flink05_Window_ProcessFunction_Aggregate.merge");
                   a.sum += b.sum;
                   a.count += b.count;
                   return a;
               }
           }
            )
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public static class Avg {
        public Integer sum = 0;
        public Long count = 0L;
    }
}

3) reduce 和 Aggregate的补充窗口的信息:增加processwindow处理函数

public class Flink06_Window_ProcessFunction_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .keyBy(WaterSensor::getId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .reduce(
                new ReduceFunction<WaterSensor>() {
                    @Override
                    public WaterSensor reduce(WaterSensor value1,
                                              WaterSensor value2) throws Exception {
                        value1.setVc(value1.getVc() + value2.getVc());
                        return value1;
                    }
                },
                // 前面聚合函数最终的输出, 就是这里的输入
                new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                               // 这个集合存储的值, 是聚合函数最终的结果
                               // 所以说, 这个集合中有且仅有一个元素, 就是前面聚合的最终结果
                               // 对这个结果最再加工: 补充点窗口信息
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
    
                        System.out.println("窗口关闭的时候执行....");
                        WaterSensor ws = elements.iterator().next();
                        
                        out.collect(key + " " + ctx.window() + " " + ws);
                    }
                }
            )
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

==7.2 事件时间和WaterMark(水印)==

7.2.1 处理时间

1) 处理时间是指的执行操作的各个设备的时间 2) 对于运行在处理时间上的流程序, 所有的基于时间的操作(比如时间窗口)都是使用的设备时钟.比如, 一个长度为1个小时的窗口将会包含设备时钟表示的1个小时内所有的数据. 假设应用程序在 9:15am分启动, 第1个小时窗口将会包含9:15am到10:00am所有的数据, 然后下个窗口是10:00am-11:00am, 等等 处理时间是最简单时间语义, 数据流和设备之间不需要做任何的协调. 他提供了最好的性能和最低的延迟. 但是, 在分布式和异步的环境下, 处理时间没有办法保证确定性, 容易受到数据传递速度的影响: 事件的延迟和乱序 3) 在使用窗口的时候, 如果使用处理时间, 就指定时间分配器为处理时间分配器

7.2.2 事件时间(event time)

1) 事件时间是指的这个事件发生的时间.

2) 在event进入Flink之前, 通常被嵌入到了event中, 一般作为这个event的时间戳存在.

3) 在事件时间体系中, 时间的进度依赖于数据本身, 和任何设备的时间无关. 事件时间程序必须制定如何产生Event Time Watermarks(水印) . 在事件时间体系中, 水印是表示时间进度的标志(作用就相当于现实时间的时钟).

4) 在理想情况下，不管事件时间何时到达或者他们的到达的顺序如何, 事件时间处理将产生完全一致且确定的结果. 事件时间处理会在等待无序事件(迟到事件)时产生一定的延迟。由于只能等待有限的时间，因此这限制了确定性事件时间应用程序的可使用性。

5) 假设所有数据都已到达，事件时间操作将按预期方式运行，即使在处理无序或迟到的事件或重新处理历史数据时，也会产生正确且一致的结果。例如，每小时事件时间窗口将包含带有事件时间戳的所有记录，该记录落入该小时，无论它们到达的顺序或处理时间。

6) 在使用窗口的时候, 如果使用事件时间, 就指定时间分配器为事件时间分配器

注意:

在1.12之前默认的时间语义是处理时间, 从1.12开始, Flink内部已经把默认的语义改成了事件时间

水印是一个时钟，决定事件时间窗口什么时候关

7.2.3 事件时间和水印

// 1: 乱序流中的水印
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    // 返回值必须是毫秒. 是一个绝对时间 getTs是javaBean的时间,数据的时间水印
                    .withTimestampAssigner((element, ts) -> element.getTs())
            )
                
// 2: 有序流中的水印
// 3: 事件时间设置
            滚动  TumblingEventTimeWindows.of
                窗口之间没有间隙也没重叠
            滑动
                窗口之间有间隙也可能重叠
            会话
                gap
                超过gap时间没有数据, 窗口关闭
                
// 4: 水印的生产机制
    最大时间戳-乱序程度-1ms
    水印 随着数据的流动而流动
    时钟
        表示事件时间的进度
​
        窗口的关闭
        定时器的触发
        触发的触发...
                
// 5:时间事件+水印 
1) 解决数据迟到问题的一从保证:
2) 假如你是启动一个5秒的窗口,Duration.ofSeconds(3)乱序程度是3秒,
水印决定了关闭窗口,当你输入s1,8000,10  8000Ts-3000-1 和 [0,5000)的窗口的数据最对比,
这样就算是s1,5000,10的数据都可以进去窗口,水印>5000时候,窗口关闭,这样数据才进不来了.
s1,8000,10 他只会金 [5000,10000)的窗口,看水印决定窗口是否关闭
                                           
// 6: 水印的类型 水印可以不用快速的更新,这样迟到的数据也可以进去
周期性的水印(  每隔3秒中  )（默认每隔200毫秒发射一次水印）
打点是水印(  来一条数据执行一次 )

1: 水印和事件时间的演示

public class WindowWaterMark {
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 今天测试水印的时候: 并行度必须设置 1
        env.setParallelism(1);
​
        // 这里测试的是抽象时间
        // 返回值就是这个与元素的事件时间
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
                /*分配时间时间和水印,一起运用
                <WaterSensor>forBoundedOutOfOrderness是一个乱序的数据,流中的类型指定
                withTimestampAssigner 告诉别人这个时间时间的分配器
                 */
            .assignTimestampsAndWatermarks(
                    WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner
                            ((element, ts) -> element.getTs())
            )
            .keyBy(WaterSensor::getId)
                //这里一定要事件时间的代码
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
​
                // 处理窗口中元素的方法: 当窗口关闭的时候执行一次
                @Override
                public void process(String s,  // 属于哪个key的窗口
                                    Context context,  // 上下文对象, 可以获取到窗口信息
                                    Iterable<WaterSensor> elements,  // 窗口触发计算的时候, 存储着窗口内所有的元素
                                    Collector<String> out) throws Exception {
​
                    ArrayList<WaterSensor> list = new ArrayList<>();
                    // Iterable很多操作不能做, 比如排序, 一般是把他转成一个List集合再操作
                    for (WaterSensor ws : elements) {
                        list.add(ws);
                    }
​
                    // 获取窗口的开始时间
                    String start = AtguiguUtil.toDatTime(context.window().getStart());
                    //获取窗口结束时间
                    String stop = AtguiguUtil.toDatTime(context.window().getEnd());
                    out.collect(s + ":" + start +  " " + stop + " " + list);
​
​
                }
            })
            .print();
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

2: 自定义的水印机制,可以改变的

public class WindowWaterMarkCustom {
​
    public static void main(String[] args) {
​
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 水印更新周期
        env.getConfig().setAutoWatermarkInterval(3000);
        // 今天测试水印的时候: 并行度必须设置 1
        env.setParallelism(1);
​
        // 这里测试的是抽象时间
        // 返回值就是这个与元素的事件时间
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .assignTimestampsAndWatermarks(
                    new WatermarkStrategy<WaterSensor>() {
                        @Override
                        public WatermarkGenerator<WaterSensor> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                            return new MyWM();
                        }
                    }
                    .withTimestampAssigner( (element, recordTimestamp) -> element.getTs())
​
            )
            .keyBy(WaterSensor::getId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
​
                // 处理窗口中元素的方法: 当窗口关闭的时候执行一次
                @Override
                public void process(String key,  // 属于哪个key的窗口
                                    Context ctx,  // 上下文对象, 可以获取到窗口信息
                                    Iterable<WaterSensor> elements, // 窗口触发计算的时候, 存储着窗口内所有的元素
                                    Collector<String> out) throws Exception {
                    ArrayList<WaterSensor> list = new ArrayList<>();
                    // Iterable很多操作不能做, 比如排序, 一般是把他转成一个List集合再操作
                    for (WaterSensor ws : elements) {
                        list.add(ws);
                    }
​
                    // 获取窗口的开始时间
                    String stt = AtguiguUtil.toDatTime(ctx.window().getStart());
                    // 获取窗口的结束时间
                    String edt = AtguiguUtil.toDatTime(ctx.window().getEnd());
​
                    out.collect(key + " " + stt + " " + edt + "  " + list);
                }
            })
            .print();
​
​
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
​
    public static class MyWM implements WatermarkGenerator<WaterSensor> {
        Long MaxTs = Long.MIN_VALUE + 3000 + 1 ;
​
        // 流中每来一个元素触发一次
        // 参数1:就是那个元素
        //参数2: 这个元素的事件时间
        // 参数3: 水印发射器
        @Override
        public void onEvent(WaterSensor event, long ts, WatermarkOutput output) {
​
            System.out.println("MyWMG.onEvent");
​
            // 计算最大时间戳,默认的水印调用这个方法
             MaxTs = Math.max(MaxTs, ts);
​
​
            // 间隙性的水印/打点式 (两种水印的方式)
            output.emitWatermark(new Watermark(MaxTs - 3000 - 1));
​
        }
​
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
​
            // 周期性的执行:默认是200ms
            System.out.println("MyWMG.onPeriodicEmit");
​
            // 周期性的水印
            output.emitWatermark(new Watermark(MaxTs - 3000 - 1));
            //走这里就是没有水印,一直是0
​
        }
    }
}

7.2.4 多并行度WaterMark的传递

总结: 多并行度的条件下, 向下游传递WaterMark的时候, 总是以最小的那个WaterMark为准! 木桶原理!

// 1: 多个并行度传递,就有多个水印?
广播传递:找一个最小的时间传递后面
​
// 2:多个source读取kafka的数据会发生数据倾斜,怎么解决?
1) map之前平局分布数据rebalence(时效降低)
2) 找别的部门解决数据倾斜
3) 1个source(不建议)
4) 把不更新的数据并行度标记空闲,就以其他并行度为标准,水印传递下去
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    // 返回值必须是毫秒. 是一个绝对时间
                    .withTimestampAssigner((element, ts) -> element.getTs())
                    // 如果一个并行度10s每个更新数据(没有更新水印), 水印传递就以其他的为准
                    .withIdleness(Duration.ofSeconds(10))

7.3 数据迟到的三从保证

// 1:解决数据迟到问题的一重保证:
事件时间 + 水印
​
// 2:二重保证:允许迟到 类似他们的最大容忍程度
当到了窗口关闭时间, 窗口先不关闭, 只是对窗口内的元素进行计算.
然后允许迟到期间, 如果有属于这个窗口的数据来了, 则来一条就计算一次.
超过了允许迟到时间, 窗口才真正的关闭!!!!
    // 允许迟到2s
     .allowedLateness(Time.seconds(2))
    
// 3:三重保证 安排一辆车侧输出流
侧输出流:如果窗口真正关闭之后, 这是时候来的属于关闭窗口的数据, 会直接进入侧输出流

1: 二从保证代码实现

2: 侧输出流三重保证

public class Flink05_WMK_Sideout {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 今天测试水印的时候: 并行度必须设置 1
        env.setParallelism(1);
        
        SingleOutputStreamOperator<String> normal = env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    // 返回值必须是毫秒. 是一个绝对时间
                    .withTimestampAssigner((element, ts) -> element.getTs())
            )
            .keyBy(WaterSensor::getId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            // 把真正迟到的迟到(窗口之后进来的数据)的数据写入到侧输出流中
            // 做成匿名内部类的方式,来保存泛型到运行时
            .sideOutputLateData(new OutputTag<WaterSensor>("late"){})
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
            
                // 处理窗口中元素的方法: 当窗口关闭的时候执行一次
                @Override
                public void process(String key,  // 属于哪个key的窗口
                                    Context ctx,  // 上下文对象, 可以获取到窗口信息
                                    Iterable<WaterSensor> elements, // 窗口触发计算的时候, 存储着窗口内所有的元素
                                    Collector<String> out) throws Exception {
                    ArrayList<WaterSensor> list = new ArrayList<>();
                    // Iterable很多操作不能做, 比如排序, 一般是把他转成一个List集合再操作
                    for (WaterSensor ws : elements) {
                        list.add(ws);
                    }
                
                    // 获取窗口的开始时间
                    String stt = AtguiguUtil.toDatTime(ctx.window().getStart());
                    // 获取窗口的结束时间
                    String edt = AtguiguUtil.toDatTime(ctx.window().getEnd());
                
                    out.collect(key + " " + stt + " " + edt + "  " + list);
                }
            });
        normal.print("正常数据");
        normal.getSideOutput(new OutputTag<WaterSensor>("late"){}).print("迟到数据");
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}