42讲轻松通关Flink学习笔记（上）

àì夳堔傛蜴生んèń 2022-12-02 13:58 111阅读 0赞

### 文章目录 ###

*  第一讲 Flink的应用场景和架构模型
 *  第二讲 Flink入门程序WordCount和SQL实现
 *  第三讲 Flink的编程模型与其他框架比较
 *  第四讲 Flink常用的DataSet和DataStream API
 *  第五讲 Flink SQL & Table编程和案例
 *  第六讲 Flink集群安装部署和HA配置
 *  第七讲 Flink常用核心概念分析
 *  第八讲 Flink窗口、时间和水印
 *  第九讲 Flink状态与容错
 *  第十讲 Flink Side OutPut分流
 *  第十一讲 Flink CEP复杂事件处理
 *  第十二讲 Flink常用的Source和Connector
 *  第十三讲 如何实现生产环境中的Flink高可用配置
 *  第十四讲 Flink Exactly-once实现原理解析

# 第一讲 Flink的应用场景和架构模型 #

flink开源于2019年初，迅速成为大数据实时计算领域炙手可热的技术框架。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center]  
Flink支持运行在包括：

*  YARN
 *  Mesos
 *  Kubernetes

在内的多种资源管理框架上，Flink已经可以扩展到数千核心，其状态可以达到TB级别，且仍能保持高吞吐、低延迟的特性。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 1]  
Flink程序的基础构建模块：

*  流（Streams）
 *  转换（Transformations）

每一个数据流起始于一个或多个Source，并终止于一个或多个Sink。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 2]  
**窗口和时间**是Flink中的核心概念之一，在实际生产环境中，对数据流上的聚合需要由窗口来划定范围，比如“计算过去的5分钟”或者“最后100个元素的和”。

Flink支持多种窗口模型

*  滚动窗口（Tunmbling Window）
 *  滑动窗口（Sliding Window）
 *  会话窗口（Session Window）

Flink支持三种时间类型

*  Event Time
 *  Ingestion Time
 *  Processing Time

Flink自身还支持

*  有状态的算子操作
 *  容错机制
 *  Checkpoint
 *  Exactly-once语义
 *  等更多高级特性

# 第二讲 Flink入门程序WordCount和SQL实现 #

任何一门大数据框架在实际生产环境中都是以集群的形式运行

Flink是一个以Java和Scala作为开发语言的开源大数据项目，通常我们推荐使用Java来作为开发语言，Maven作为编译和包管理工具进行项目构建和编译

对于大多数开发者而言，JDK、Maven和Git这三个开发工具是必不可少的

**DataSet WordCount**

WordCount 程序是大数据处理框架的入门程序，俗称“单词计数”。用来统计一段文字每个单词的出现次数，该程序主要分为两个部分：一部分是将文字拆分成单词；另一部分是单词进行分组计数并打印输出结果。

整体代码实现如下：

public class DataSetJob { 
        public static void main(String[] args) throws Exception { 
    
          // 创建Flink运行的上下文环境
          final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
          // 创建DataSet，这里我们的输入是一行一行的文本
          DataSet<String> text = env.fromElements(
                "Flink Spark Storm",
                "Flink Flink Flink",
                "Spark Spark Spark",
                "Storm Storm Storm"
          );
          // 通过Flink内置的转换函数进行计算
          DataSet<Tuple2<String, Integer>> counts =
                text.flatMap(new LineSplitter())
                      .groupBy(0)
                      .sum(1);
          //结果打印
          counts.printToErr();
    
       }
    
       public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> { 
    
          @Override
          public void flatMap(String value, Collector<Tuple2<String, Integer>> out) { 
             // 将文本分割
             String[] tokens = value.toLowerCase().split("\\W+");
    
             for (String token : tokens) { 
                if (token.length() > 0) { 
                   out.collect(new Tuple2<String, Integer>(token, 1));
                }
             }
          }
        }
    }

**DataStream WordCount**

为了模仿一个流式计算环境，我们选择监听一个本地的 Socket 端口，并且使用 Flink 中的滚动窗口，每 5 秒打印一次计算结果。代码如下：

public class StreamingJob { 
    
        public static void main(String[] args) throws Exception { 
    
            // 创建Flink的流式计算环境
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 监听本地9000端口
            DataStream<String> text = env.socketTextStream("127.0.0.1", 9000, "\n");
    
            // 将接收的数据进行拆分，分组，窗口计算并且进行聚合输出
            DataStream<WordWithCount> windowCounts = text
                    .flatMap(new FlatMapFunction<String, WordWithCount>() { 
                        @Override
                        public void flatMap(String value, Collector<WordWithCount> out) { 
                            for (String word : value.split("\\s")) { 
                                out.collect(new WordWithCount(word, 1L));
                            }
                        }
                    })
                    .keyBy("word")
                    .timeWindow(Time.seconds(5), Time.seconds(1))
                    .reduce(new ReduceFunction<WordWithCount>() { 
                        @Override
                        public WordWithCount reduce(WordWithCount a, WordWithCount b) { 
                            return new WordWithCount(a.word, a.count + b.count);
                        }
                    });
    
            // 打印结果
            windowCounts.print().setParallelism(1);
    
            env.execute("Socket Window WordCount");
        }
    
        // Data type for words with count
        public static class WordWithCount { 
    
            public String word;
            public long count;
    
            public WordWithCount() { }
    
            public WordWithCount(String word, long count) { 
                this.word = word;
                this.count = count;
            }
    
            @Override
            public String toString() { 
                return word + " : " + count;
            }
        }
    }

我们在本地使用 netcat 命令启动一个端口：

nc -lk 9000

然后直接运行我们的 main 方法即可。

在 nc 中输入：

$ nc -lk 9000
    Flink Flink Flink 
    Flink Spark Storm

可以在控制台看到：

Flink : 4
    Spark : 1
    Storm : 1

**Flink Table & SQL WordCount**

Flink SQL 是 Flink 实时计算为简化计算模型，降低用户使用实时计算门槛而设计的一套符合标准 SQL 语义的开发语言。

一个完整的 Flink SQL 编写的程序包括如下三部分。

*  Source Operator：是对外部数据源的抽象, 目前 Apache Flink 内置了很多常用的数据源实现，比如 MySQL、Kafka 等。
 *  Transformation Operators：算子操作主要完成比如查询、聚合操作等，目前 Flink SQL 支持了 Union、Join、Projection、Difference、Intersection 及 window 等大多数传统数据库支持的操作。
 *  Sink Operator：是对外结果表的抽象，目前 Apache Flink 也内置了很多常用的结果表的抽象，比如 Kafka Sink 等。

我们也是通过用一个最经典的 WordCount 程序作为入门，上面已经通过 DataSet/DataStream API 开发，那么实现同样的 WordCount 功能， Flink Table & SQL 核心只需要一行代码：

//省略掉初始化环境等公共代码
    SELECT word, COUNT(word) FROM table GROUP BY word;

WordCountSQL：

public class WordCountSQL { 
    
        public static void main(String[] args) throws Exception{ 
    
            //获取运行环境
            ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
            //创建一个tableEnvironment
            BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);
    
            String words = "hello flink hello lagou";
    
            String[] split = words.split("\\W+");
            ArrayList<WC> list = new ArrayList<>();
    
            for(String word : split){ 
                WC wc = new WC(word,1);
                list.add(wc);
            }
            DataSet<WC> input = fbEnv.fromCollection(list);
    
            //DataSet 转sql, 指定字段名
            Table table = fbTableEnv.fromDataSet(input, "word,frequency");
            table.printSchema();
    
            //注册为一个表
            fbTableEnv.createTemporaryView("WordCount", table);
    
            Table table02 = fbTableEnv.sqlQuery("select word as word, sum(frequency) as frequency from WordCount GROUP BY word");
    
            //将表转换DataSet
            DataSet<WC> ds3  = fbTableEnv.toDataSet(table02, WC.class);
            ds3.printToErr();
        }
    
        public static class WC { 
            public String word;
            public long frequency;
    
            public WC() { }
    
            public WC(String word, long frequency) { 
                this.word = word;
                this.frequency = frequency;
            }
    
            @Override
            public String toString() { 
                return  word + ", " + frequency;
            }
        }
    }

我们直接运行该程序，在控制台可以看到输出结果：

lagou, 1
    flink, 1
    hello, 2

# 第三讲 Flink的编程模型与其他框架比较 #

Streams（流）

*  流分为有界流和无界流
 *  有界流指的是有固定大小，不随时间增加而增长的数据，比如保存在Hive中的一个表
 *  无界流指的是数据随着时间增加而增长，计算状态持续进行  
    比如消费Kafka中的消息，消息持续不断，那么计算也会持续进行不会结束

State（状态）

*  所谓的状态指的是在**进行流式计算过程中的信息**
 *  一般用作容错恢复和持久化，流式计算在本质上是**增量计算**，也就是说需要不断地查询过去的状态
 *  状态在Flink中有十分重要的作用，例如为了保证Exactly-once语义需要将数据写到状态中
 *  此外**状态的持久化存储**也是集群出现Fail-over的情况下自动重启的前提条件

Time（时间）

*  Flink支持了Event Time、Ingestion time、Processing time等多种时间语义
 *  时间是我们在进行Flink程序开发时判断业务状态是否滞后和延迟的重要依据

API

*  Flink自身提供了不同级别的抽象来支持我们开发流式或者批量处理程序
 *  由上而下可分为SQL/Table API、DataStream API、ProcessFunction三层
 *  开发者可以根据需要选择不同层级的API进行开发

>  *  Flink程序的基础构建模块是流（Streams）和转换（Transformations）
>  *  每一个数据流起始于一个或多个Source，并终止与一个或多个Sink
>  *  数据流类似于有向无环图（DAG）

> 算子链：在分布式运行环境中，Flink提出了算子链的概念，Flink将多个算子放在一个任务中，由同一个线程执行，减少了线程之间的切换、消息的序列化反序列化、数据在缓冲区的交换，减少延迟的同时提高整体的吞吐量。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 3]  
在实际生产中，Flink都是以集群在运行，在运行的过程中包含了**两类进程**

*  JobManager：它扮演的是集群管理者的角色，负责调度任务，协调checkpoints、协调故障恢复、收集Job的状态信息，管理Flink集群中的从节点TaskManager
 *  TaskManager：实际负责执行计算的Worker，在其上执行Flink Job的一组Task，TaskManager还是所在节点的管理员，它负责把该节点上的服务器信息比如内存、磁盘、任务运行情况等向JobManager汇报
 *  Client：用户在提交编写好的Flink工程时，会先创建一个客户端再进行提交，这个客户端就是Client，Client会根据用户传入的参数选择使用yarn per job模式、stand-alone模式还是yarn-session模式将Flink程序提交到集群
 *  在Flink集群中，一个TaskManager就是一个JVM进程，可以用独立的线程来执行task
 *  为了控制一个TaskManager能接受多少个task，Flink提出了Task Slot的概念，把Task Slot理解为TaskManager的计算资源子集  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 4]
 *  需要注意的是，slot仅仅用来做内存的隔离，对CPU不起作用，运行在同一个JVM的task可以共享TCP连接，减少网络传输  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 5]  
    **Flink的优势及与其他框架的区别**

**架构**

*  Storm的架构是经典的主从结构，并且强依赖ZooKeeper
 *  Spark Streaming的架构是基于Spark的  
    它的本质是微批处理，每个batch都依赖Driver  
    Spark Streaming理解为时间维度上的Spark DAG
 *  Flink采用了经典的主从模式，DataFlow Graph与Storm形成的拓扑Topology结构类似  
    Flink程序启动后，会根据用户的代码处理成Stream Graph  
    进一步优化成为JobGraph，JobManager会根据JobGragh生成ExecutionGraph  
    **ExecutionGraph才是Flink真正能执行的数据结构**  
    当很多个ExecutionGraph分布在集群中，就会形成一张网状的拓扑结构

**容错**

*  Storm在容错方面只支持了Record级别的ACK-FAIL，发送出去的每一条消息都可以确定是被成功处理或失败处理，因此Storm支持至少处理一次语义
 *  针对以前的Spark Streaming任务，我们可以配置对应的checkpoint，也就是保存点
 *  当任务出现failover的时候，会从checkpoint重新加载，使得数据不丢失，但是这个过程会导致原来的数据重复处理，不能做到“只处理一次”语义

**反压**

*  反压是分布式处理系统中经常遇到的问题
 *  当消费者速度低于生产者速度的时候，则需要消费者将信息反馈给生产者，使得生产者的速度能和消费者的速度进行匹配
 *  当下游消费者速度跟不上生产者的速度时，会直接通知生产者停止生产数据。这种方式的缺点是不能实现逐级反压，且调优困难，设置的消费速率过小导致集群吞吐量低下速率过大会导致消费者OOM
 *  Spark Streaming为了实现反压这个功能，在原来的架构基础上构造了一个“速率控制器”  
    这个“速率控制器”会根据几个属性  
    如任务的结束时间、处理时长处理消息的条数等计算一个速率  
    在实现控制速率的接收速率中用到了一个经典的算法，即“PID算法”
 *  Flink没有使用任何复杂的机制来解决反压问题  
    Flink在数据传输过程中，使用了分布式阻塞队列  
    在一个阻塞队列中，当队列满了以后发送者会被天然阻塞住，这种阻塞功能相当于给这个阻塞队列，提供了反压的能力

# 第四讲 Flink常用的DataSet和DataStream API #

**自定义实时数据源**

在本课时中，我们利用 Flink 提供的自定义 Source 功能来实现一个自定义的实时数据源，具体实现如下：

public class MyStreamingSource implements SourceFunction<MyStreamingSource.Item> { 
    
        private boolean isRunning = true;
    
        /** * 重写run方法产生一个源源不断的数据发送源 * @param ctx * @throws Exception */
        @Override
        public void run(SourceContext<Item> ctx) throws Exception { 
            while(isRunning){ 
                Item item = generateItem();
                ctx.collect(item);
    
                //每秒产生一条数据
                Thread.sleep(1000);
            }
        }
        @Override
        public void cancel() { 
            isRunning = false;
        }
    
        //随机产生一条商品数据
        private Item generateItem(){ 
            int i = new Random().nextInt(100);
    
            Item item = new Item();
            item.setName("name" + i);
            item.setId(i);
            return item;
        }
    
        class Item{ 
            private String name;
            private Integer id;
    
            Item() { 
            }
    
            public String getName() { 
                return name;
            }
    
            void setName(String name) { 
                this.name = name;
            }
    
            private Integer getId() { 
                return id;
            }
    
            void setId(Integer id) { 
                this.id = id;
            }
    
            @Override
            public String toString() { 
                return "Item{" +
                        "name='" + name + '\'' +
                        ", id=" + id +
                        '}';
            }
        }
    }
    
    
    class StreamingDemo { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //获取数据源
            DataStreamSource<MyStreamingSource.Item> text = 
            //注意：并行度设置为1,我们会在后面的课程中详细讲解并行度
            env.addSource(new MyStreamingSource()).setParallelism(1); 
            DataStream<MyStreamingSource.Item> item = text.map(
                    (MapFunction<MyStreamingSource.Item, MyStreamingSource.Item>) value -> value);
    
            //打印结果
            item.print().setParallelism(1);
            String jobName = "user defined streaming source";
            env.execute(jobName);
        }
    
    }

在自定义的数据源中，实现了 Flink 中的 SourceFunction 接口，同时实现了其中的 run 方法，在 run 方法中每隔一秒钟随机发送一个自定义的 Item。

可以直接运行 main 方法来进行测试。

下面我们就用自定义的实时数据源来演示 DataStream API 的使用。

**Map**

Map 接受一个元素作为输入，并且根据开发者自定义的逻辑处理后输出。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 6]

class StreamingDemo { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //获取数据源
            DataStreamSource<MyStreamingSource.Item> items = env.addSource(new MyStreamingSource()).setParallelism(1); 
            //Map
            SingleOutputStreamOperator<Object> mapItems = items.map(new MapFunction<MyStreamingSource.Item, Object>() { 
                @Override
                public Object map(MyStreamingSource.Item item) throws Exception { 
                    return item.getName();
                }
            });
            //打印结果
            mapItems.print().setParallelism(1);
            String jobName = "user defined streaming source";
            env.execute(jobName);
        }
    }

我们只取出每个 Item 的 name 字段进行打印。

注意，**Map 算子是最常用的算子之一**，官网中的表述是对一个 DataStream 进行映射，每次进行转换都会调用 MapFunction 函数。从源 DataStream 到目标 DataStream 的转换过程中，返回的是 SingleOutputStreamOperator。当然了，我们也可以在重写的 map 函数中使用 lambda 表达式。

SingleOutputStreamOperator<Object> mapItems = items.map(
          item -> item.getName()
    );

甚至，还可以自定义自己的 Map 函数。通过重写 MapFunction 或 RichMapFunction 来自定义自己的 map 函数。

class StreamingDemo { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //获取数据源
            DataStreamSource<MyStreamingSource.Item> items = env.addSource(new MyStreamingSource()).setParallelism(1);
            SingleOutputStreamOperator<String> mapItems = items.map(new MyMapFunction());
            //打印结果
            mapItems.print().setParallelism(1);
            String jobName = "user defined streaming source";
            env.execute(jobName);
        }
    
        static class MyMapFunction extends RichMapFunction<MyStreamingSource.Item,String> { 
    
            @Override
            public String map(MyStreamingSource.Item item) throws Exception { 
                return item.getName();
            }
        }
    }

此外，在 RichMapFunction 中还提供了 open、close 等函数方法，重写这些方法还能实现更为复杂的功能，比如获取累加器、计数器等。

**FlatMap**

FlatMap 接受一个元素，返回零到多个元素。FlatMap 和 Map 有些类似，但是当返回值是列表的时候，FlatMap 会将列表“平铺”，也就是以单个元素的形式进行输出。

SingleOutputStreamOperator<Object> flatMapItems = items.flatMap(new FlatMapFunction<MyStreamingSource.Item, Object>() { 
        @Override
        public void flatMap(MyStreamingSource.Item item, Collector<Object> collector) throws Exception { 
            String name = item.getName();
            collector.collect(name);
        }
    });

上面的程序会把名字逐个输出。我们也可以在 FlatMap 中实现更为复杂的逻辑，比如过滤掉一些我们不需要的数据等。

**Filter**

顾名思义，Fliter 的意思就是过滤掉不需要的数据，每个元素都会被 filter 函数处理，如果 filter 函数返回 true 则保留，否则丢弃。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 7]  
例如，我们只保留 id 为偶数的那些 item。

SingleOutputStreamOperator<MyStreamingSource.Item> filterItems = items.filter(new FilterFunction<MyStreamingSource.Item>() { 
        @Override
        public boolean filter(MyStreamingSource.Item item) throws Exception { 
    
            return item.getId() % 2 == 0;
        }
    });

当然，我们也可以在 filter 中使用 lambda 表达式：

SingleOutputStreamOperator<MyStreamingSource.Item> filterItems = items.filter( 
        item -> item.getId() % 2 == 0
    );

**KeyBy**

在介绍 KeyBy 函数之前，需要你理解一个概念：KeyedStream。 在实际业务中，我们经常会需要根据数据的某种属性或者单纯某个字段进行分组，然后对不同的组进行不同的处理。举个例子，当我们需要描述一个用户画像时，则需要根据用户的不同行为事件进行加权；再比如，我们在监控双十一的交易大盘时，则需要按照商品的品类进行分组，分别计算销售额。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 8]  
我们在使用 KeyBy 函数时会把 DataStream 转换成为 KeyedStream，事实上 KeyedStream 继承了 DataStream，KeyedStream 中的元素会根据用户传入的参数进行分组。

我们在第 02 课时中讲解的 WordCount 程序，曾经使用过 KeyBy：

// 将接收的数据进行拆分，分组，窗口计算并且进行聚合输出
        DataStream<WordWithCount> windowCounts = text
                .flatMap(new FlatMapFunction<String, WordWithCount>() { 
                    @Override
                    public void flatMap(String value, Collector<WordWithCount> out) { 
                        for (String word : value.split("\\s")) { 
                            out.collect(new WordWithCount(word, 1L));
                        }
                    }
                })
                .keyBy("word")
                .timeWindow(Time.seconds(5), Time.seconds
                ....

在生产环境中使用 KeyBy 函数时要十分注意！该函数会把数据按照用户指定的 key 进行分组，那么相同分组的数据会被分发到一个 subtask 上进行处理，在大数据量和 key 分布不均匀的时非常容易出现数据倾斜和反压，导致任务失败。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 9]  
常见的解决方式是把所有**数据加上随机前后缀**。

**Aggregations**

Aggregations 为聚合函数的总称，常见的聚合函数包括但不限于 sum、max、min 等。Aggregations 也需要指定一个 key 进行聚合，官网给出了几个常见的例子：

keyedStream.sum(0);
    keyedStream.sum("key");
    keyedStream.min(0);
    keyedStream.min("key");
    keyedStream.max(0);
    keyedStream.max("key");
    keyedStream.minBy(0);
    keyedStream.minBy("key");
    keyedStream.maxBy(0);
    keyedStream.maxBy("key");

在上面的这几个函数中，max、min、sum 会分别返回最大值、最小值和汇总值；而 minBy 和 maxBy 则会把最小或者最大的元素全部返回。

我们拿 max 和 maxBy 举例说明：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //获取数据源
    List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();
    data.add(new Tuple3<>(0,1,0));
    data.add(new Tuple3<>(0,1,1));
    data.add(new Tuple3<>(0,2,2));
    data.add(new Tuple3<>(0,1,3));
    data.add(new Tuple3<>(1,2,5));
    data.add(new Tuple3<>(1,2,9));
    data.add(new Tuple3<>(1,2,11));
    data.add(new Tuple3<>(1,2,13));
    
    DataStreamSource<MyStreamingSource.Item> items = env.fromCollection(data);
    items.keyBy(0).max(2).printToErr();
    
    //打印结果
    String jobName = "user defined streaming source";
    env.execute(jobName);

我们直接运行程序，会发现奇怪的一幕：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 10]  
从上图中可以看到，我们希望按照 Tuple3 的第一个元素进行聚合，并且按照第三个元素取最大值。结果如我们所料，的确是按照第三个元素大小依次进行的打印，但是结果却出现了一个这样的元素 (0,1,2)，这在我们的源数据中并不存在。

我们在 Flink 官网中的文档可以发现：

> The difference between min and minBy is that min returns the minimum value, whereas minBy returns the element that has the minimum value in this field (same for max and maxBy).

文档中说：min 和 minBy 的区别在于，min 会返回我们制定字段的最小值，minBy 会返回对应的元素（max 和 maxBy 同理）。

网上很多资料也这么写：min 和 minBy 的区别在于 min 返回最小的值，而 minBy 返回最小值的key，严格来说这是不正确的。

min 和 minBy 都会返回整个元素，只是 min 会根据用户指定的字段取最小值，并且把这个值保存在对应的位置，而对于其他的字段，并不能保证其数值正确。max 和 maxBy 同理。

事实上，对于 Aggregations 函数，Flink 帮助我们封装了状态数据，这些状态数据不会被清理，所以在实际生产环境中应该尽量避免在一个无限流上使用 Aggregations。而且，对于同一个 keyedStream ，只能调用一次 Aggregation 函数。

**Reduce**

Reduce 函数的原理是，会在每一个分组的 keyedStream 上生效，它会按照用户自定义的聚合逻辑进行分组聚合。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 11]  
例如：

List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();
    data.add(new Tuple3<>(0,1,0));
    data.add(new Tuple3<>(0,1,1));
    data.add(new Tuple3<>(0,2,2));
    data.add(new Tuple3<>(0,1,3));
    data.add(new Tuple3<>(1,2,5));
    data.add(new Tuple3<>(1,2,9));
    data.add(new Tuple3<>(1,2,11));
    data.add(new Tuple3<>(1,2,13));
    
    DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);
    //items.keyBy(0).max(2).printToErr();
    
    SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> reduce = items.keyBy(0).reduce(new ReduceFunction<Tuple3<Integer, Integer, Integer>>() { 
        @Override
        public Tuple3<Integer,Integer,Integer> reduce(Tuple3<Integer, Integer, Integer> t1, Tuple3<Integer, Integer, Integer> t2) throws Exception { 
            Tuple3<Integer,Integer,Integer> newTuple = new Tuple3<>();
    
            newTuple.setFields(0,0,(Integer)t1.getField(2) + (Integer) t2.getField(2));
            return newTuple;
        }
    });
    
    reduce.printToErr().setParallelism(1);

我们对下面的元素按照第一个元素进行分组，第三个元素分别求和，并且把第一个和第二个元素都置为 0：

data.add(new Tuple3<>(0,1,0));
    data.add(new Tuple3<>(0,1,1));
    data.add(new Tuple3<>(0,2,2));
    data.add(new Tuple3<>(0,1,3));
    data.add(new Tuple3<>(1,2,5));
    data.add(new Tuple3<>(1,2,9));
    data.add(new Tuple3<>(1,2,11));
    data.add(new Tuple3<>(1,2,13));

那么最终会得到：(0,0,6) 和 (0,0,38)。

# 第五讲 Flink SQL & Table编程和案例 #

**背景**

我们在前面的课时中讲过 Flink 的分层模型，Flink 自身提供了不同级别的抽象来支持我们开发流式或者批量处理程序，下图描述了 Flink 支持的 4 种不同级别的抽象。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 12]  
Table API 和 SQL 处于最顶端，是 Flink 提供的高级 API 操作。Flink SQL 是 Flink 实时计算为简化计算模型，降低用户使用实时计算门槛而设计的一套符合标准 SQL 语义的开发语言。

我们在第 04 课时中提到过，Flink 在编程模型上提供了 DataStream 和 DataSet 两套 API，并没有做到事实上的批流统一，因为用户和开发者还是开发了两套代码。正是因为 Flink Table & SQL 的加入，可以说 Flink 在某种程度上做到了事实上的批流一体。

**原理**

你之前可能都了解过 Hive，在离线计算场景下 Hive 几乎扛起了离线数据处理的半壁江山。它的底层对 SQL 的解析用到了 Apache Calcite，Flink 同样把 SQL 的解析、优化和执行教给了 Calcite。

下图是一张经典的 Flink Table & SQL 实现原理图，可以看到 Calcite 在整个架构中处于绝对核心地位。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 13]  
从图中可以看到无论是批查询 SQL 还是流式查询 SQL，都会经过对应的转换器 Parser 转换成为节点树 SQLNode tree，然后生成逻辑执行计划 Logical Plan，逻辑执行计划在经过优化后生成真正可以执行的物理执行计划，交给 DataSet 或者 DataStream 的 API 去执行。

在这里我们不对 Calcite 的原理过度展开，有兴趣的可以直接在官网上学习。

一个完整的 Flink Table & SQL Job 也是由 Source、Transformation、Sink 构成：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 14]

*  Source 部分来源于外部数据源，我们经常用的有 Kafka、MySQL 等；
 *  Transformation 部分则是 Flink Table & SQL 支持的常用 SQL 算子，比如简单的 Select、Groupby 等，当然在这里也有更为复杂的多流 Join、流与维表的 Join 等；
 *  Sink 部分是指的结果存储比如 MySQL、HBase 或 Kakfa 等。

**动态表**

与传统的表 SQL 查询相比，Flink Table & SQL 在处理流数据时会时时刻刻处于动态的数据变化中，所以便有了一个动态表的概念。

动态表的查询与静态表一样，但是，在查询动态表的时候，SQL 会做连续查询，不会终止。

我们举个简单的例子，Flink 程序接受一个 Kafka 流作为输入，Kafka 中为用户的购买记录：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 15]  
首先，Kafka 的消息会被源源不断的解析成一张不断增长的动态表，我们在动态表上执行的 SQL 会不断生成新的动态表作为结果表。

**Flink Table & SQL 算子和内置函数**

我们在讲解 Flink Table & SQL 所支持的常用算子前，需要说明一点，Flink 自从 0.9 版本开始支持 Table & SQL 功能一直处于完善开发中，且在不断进行迭代。

我们在官网中也可以看到这样的提示：

> Please note that the Table API and SQL are not yet feature complete and are being actively developed. Not all operations are supported by every combination of \[Table API, SQL\] and \[stream, batch\] input.

Flink Table & SQL 的开发一直在进行中，并没有支持所有场景下的计算逻辑。从我个人实践角度来讲，在使用原生的 Flink Table & SQL 时，务必查询官网当前版本对 Table & SQL 的支持程度，尽量选择场景明确，逻辑不是极其复杂的场景。

**常用算子**

目前 Flink SQL 支持的语法主要如下：

query:
      values
      | {
          select
          | selectWithoutFrom
          | query UNION [ ALL ] query
          | query EXCEPT query
          | query INTERSECT query
        }
        [ ORDER BY orderItem [, orderItem ]* ]
        [ LIMIT { count | ALL } ]
        [ OFFSET start { ROW | ROWS } ]
        [ FETCH { FIRST | NEXT } [ count ] { ROW | ROWS } ONLY]
    
    orderItem:
      expression [ ASC | DESC ]
    
    select:
      SELECT [ ALL | DISTINCT ]
      { * | projectItem [, projectItem ]* }
      FROM tableExpression
      [ WHERE booleanExpression ]
      [ GROUP BY { groupItem [, groupItem ]* } ]
      [ HAVING booleanExpression ]
      [ WINDOW windowName AS windowSpec [, windowName AS windowSpec ]* ]
    
    selectWithoutFrom:
      SELECT [ ALL | DISTINCT ]
      { * | projectItem [, projectItem ]* }
    
    projectItem:
      expression [ [ AS ] columnAlias ]
      | tableAlias . *
    
    tableExpression:
      tableReference [, tableReference ]*
      | tableExpression [ NATURAL ] [ LEFT | RIGHT | FULL ] JOIN tableExpression [ joinCondition ]
    
    joinCondition:
      ON booleanExpression
      | USING '(' column [, column ]* ')'
    
    tableReference:
      tablePrimary
      [ matchRecognize ]
      [ [ AS ] alias [ '(' columnAlias [, columnAlias ]* ')' ] ]
    
    tablePrimary:
      [ TABLE ] [ [ catalogName . ] schemaName . ] tableName
      | LATERAL TABLE '(' functionName '(' expression [, expression ]* ')' ')'
      | UNNEST '(' expression ')'
    
    values:
      VALUES expression [, expression ]*
    
    groupItem:
      expression
      | '(' ')'
      | '(' expression [, expression ]* ')'
      | CUBE '(' expression [, expression ]* ')'
      | ROLLUP '(' expression [, expression ]* ')'
      | GROUPING SETS '(' groupItem [, groupItem ]* ')'
    
    windowRef:
        windowName
      | windowSpec
    
    windowSpec:
        [ windowName ]
        '('
        [ ORDER BY orderItem [, orderItem ]* ]
        [ PARTITION BY expression [, expression ]* ]
        [
            RANGE numericOrIntervalExpression {
       PRECEDING}
          | ROWS numericExpression {
       PRECEDING}
        ]
        ')'
    ...

可以看到 Flink SQL 和传统的 SQL 一样，支持了包含查询、连接、聚合等场景，另外还支持了包括窗口、排序等场景。下面我就以最常用的算子来做详细的讲解。

**SELECT/AS/WHERE**

SELECT、WHERE 和传统 SQL 用法一样，用于筛选和过滤数据，同时适用于 DataStream 和 DataSet。

SELECT * FROM Table;
    SELECT name，age FROM Table;

当然我们也可以在 WHERE 条件中使用 =、<、>、<>、>=、<=，以及 AND、OR 等表达式的组合：

SELECT name，age FROM Table where name LIKE '%小明%';
    SELECT * FROM Table WHERE age = 20;
    SELECT name, age
    FROM Table
    WHERE name IN (SELECT name FROM Table2)

**GROUP BY / DISTINCT/HAVING**

GROUP BY 用于进行分组操作，DISTINCT 用于结果去重。  
HAVING 和传统 SQL 一样，可以用来在聚合函数之后进行筛选。

SELECT DISTINCT name FROM Table;
    SELECT name, SUM(score) as TotalScore FROM Table GROUP BY name;
    SELECT name, SUM(score) as TotalScore FROM Table GROUP BY name HAVING
    SUM(score) > 300;

**JOIN**

JOIN 可以用于把来自两个表的数据联合起来形成结果表，目前 Flink 的 Join 只支持等值连接。Flink 支持的 JOIN 类型包括：

JOIN - INNER JOIN
    LEFT JOIN - LEFT OUTER JOIN
    RIGHT JOIN - RIGHT OUTER JOIN
    FULL JOIN - FULL OUTER JOIN

例如，用用户表和商品表进行关联：

SELECT *
    FROM User LEFT JOIN Product ON User.name = Product.buyer
    
    SELECT *
    FROM User RIGHT JOIN Product ON User.name = Product.buyer
    
    SELECT *
    FROM User FULL OUTER JOIN Product ON User.name = Product.buyer

LEFT JOIN、RIGHT JOIN 、FULL JOIN 相与我们传统 SQL 中含义一样。

**WINDOW**

根据窗口数据划分的不同，目前 Apache Flink 有如下 3 种：

*  滚动窗口，窗口数据有固定的大小，窗口中的数据不会叠加；
 *  滑动窗口，窗口数据有固定大小，并且有生成间隔；
 *  会话窗口，窗口数据没有固定的大小，根据用户传入的参数进行划分，窗口数据无叠加；

滚动窗口的语法：

SELECT 
        [gk],
        [TUMBLE_START(timeCol, size)], 
        [TUMBLE_END(timeCol, size)], 
        agg1(col1), 
        ... 
        aggn(colN)
    FROM Tab1
    GROUP BY [gk], TUMBLE(timeCol, size)

举例说明，我们需要计算每个用户每天的订单数量：

SELECT user, TUMBLE_START(timeLine, INTERVAL '1' DAY) as winStart, SUM(amount) FROM Orders GROUP BY TUMBLE(timeLine, INTERVAL '1' DAY), user;

其中，TUMBLE\_START 和 TUMBLE\_END 代表窗口的开始时间和窗口的结束时间，TUMBLE (timeLine, INTERVAL ‘1’ DAY) 中的 timeLine 代表时间字段所在的列，INTERVAL ‘1’ DAY 表示时间间隔为一天。

滑动窗口的语法与滚动窗口相比，只多了一个 slide 参数：

SELECT 
        [gk], 
        [HOP_START(timeCol, slide, size)] ,
        [HOP_END(timeCol, slide, size)],
        agg1(col1), 
        ... 
        aggN(colN) 
    FROM Tab1
    GROUP BY [gk], HOP(timeCol, slide, size)

例如，我们要每间隔一小时计算一次过去 24 小时内每个商品的销量：

SELECT product, SUM(amount) FROM Orders GROUP BY HOP(rowtime, INTERVAL '1' HOUR, INTERVAL '1' DAY), product

上述案例中的 INTERVAL ‘1’ HOUR 代表滑动窗口生成的时间间隔。

会话窗口定义了一个非活动时间，假如在指定的时间间隔内没有出现事件或消息，则会话窗口关闭。

会话窗口的语法如下：

SELECT 
        [gk], 
        SESSION_START(timeCol, gap) AS winStart,
        SESSION_END(timeCol, gap) AS winEnd,
        agg1(col1),
         ... 
        aggn(colN)
    FROM Tab1
    GROUP BY [gk], SESSION(timeCol, gap)

举例，我们需要计算每个用户过去 1 小时内的订单量：

SELECT user, SESSION_START(rowtime, INTERVAL '1' HOUR) AS sStart, SESSION_ROWTIME(rowtime, INTERVAL '1' HOUR) AS sEnd, SUM(amount) FROM Orders GROUP BY SESSION(rowtime, INTERVAL '1' HOUR), user

**内置函数**

Flink 中还有大量的内置函数，我们可以直接使用，将内置函数分类如下，具体可查阅官方文档：

*  比较函数
 *  逻辑函数
 *  算术函数
 *  字符串处理函数
 *  时间函数

**比较函数**

**Flink Table & SQL 案例**

上面分别介绍了 Flink Table & SQL 的原理和支持的算子，我们模拟一个实时的数据流，然后讲解 SQL JOIN 的用法。

在上一课时中，我们利用 Flink 提供的自定义 Source 功能来实现一个自定义的实时数据源，具体实现如下：

public class MyStreamingSource implements SourceFunction<Item> {
    
        private boolean isRunning = true;
    
        /** * 重写run方法产生一个源源不断的数据发送源 * @param ctx * @throws Exception */
        public void run(SourceContext<Item> ctx) throws Exception {
            while(isRunning){
                Item item = generateItem();
                ctx.collect(item);
    
                //每秒产生一条数据
                Thread.sleep(1000);
            }
        }
        @Override
        public void cancel() {
            isRunning = false;
        }
    
        //随机产生一条商品数据
        private Item generateItem(){
            int i = new Random().nextInt(100);
            ArrayList<String> list = new ArrayList();
            list.add("HAT");
            list.add("TIE");
            list.add("SHOE");
            Item item = new Item();
            item.setName(list.get(new Random().nextInt(3)));
            item.setId(i);
            return item;
        }
    }

我们把实时的商品数据流进行分流，分成 even 和 odd 两个流进行 JOIN，条件是名称相同，最后，把两个流的 JOIN 结果输出。

class StreamingDemo {
        public static void main(String[] args) throws Exception {
    
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
    
            SingleOutputStreamOperator<Item> source = bsEnv.addSource(new MyStreamingSource()).map(new MapFunction<Item, Item>() {
                @Override
                public Item map(Item item) throws Exception {
                    return item;
                }
            });
    
            DataStream<Item> evenSelect = source.split(new OutputSelector<Item>() {
                @Override
                public Iterable<String> select(Item value) {
                    List<String> output = new ArrayList<>();
                    if (value.getId() % 2 == 0) {
                        output.add("even");
                    } else {
                        output.add("odd");
                    }
                    return output;
                }
            }).select("even");
    
            DataStream<Item> oddSelect = source.split(new OutputSelector<Item>() {
                @Override
                public Iterable<String> select(Item value) {
                    List<String> output = new ArrayList<>();
                    if (value.getId() % 2 == 0) {
                        output.add("even");
                    } else {
                        output.add("odd");
                    }
                    return output;
                }
            }).select("odd");
    
    
            bsTableEnv.createTemporaryView("evenTable", evenSelect, "name,id");
            bsTableEnv.createTemporaryView("oddTable", oddSelect, "name,id");
    
            Table queryTable = bsTableEnv.sqlQuery("select a.id,a.name,b.id,b.name from evenTable as a join oddTable as b on a.name = b.name");
    
            queryTable.printSchema();
    
            bsTableEnv.toRetractStream(queryTable, TypeInformation.of(new TypeHint<Tuple4<Integer,String,Integer,String>>(){})).print();
    
            bsEnv.execute("streaming sql job");
        }
    
    }

直接右键运行，在控制台可以看到输出：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 16]

# 第六讲 Flink集群安装部署和HA配置 #

Flink 常见的部署模式：本地模式、Standalone 模式和 Flink On Yarn 模式。

**环境准备**

在绝大多数情况下，我们的 Flink 都是运行在 Unix 环境中的，推荐在 Mac OS 或者 Linux 环境下运行 Flink。如果是集群模式，那么可以在自己电脑上安装虚拟机，保证有一个 master 节点和两个 slave 节点。

同时，要注意在所有的机器上都应该安装 JDK 和 SSH。JDK 是我们运行 JVM 语言程序必须的，而 SSH 是为了在服务器之间进行跳转和执行命令所必须的。关于服务器之间通过 SSH 配置公钥登录，你可以直接搜索安装和配置方法，我们不做过度展开。

Flink 的安装包可以在[这里][Link 1]下载。需要注意的是，如果你要和 Hadoop 进行集成，那么我们需要使用到对应的 Hadoop 依赖，下面将会详细讲解。

**Local 模式**

Local 模式是 Flink 提供的最简单部署模式，一般用来本地测试和演示使用。

我们在[这里][Link 2]下载 [Apache Flink 1.10.0 for Scala 2.11][] 版本进行演示，该版本对应 Scala 2.11 版本。

将压缩包下载到本地，并且直接进行解压，使用 Flink 默认的端口配置，直接运行脚本启动：

➜  [SoftWare]# tar -zxvf flink-1.10.0-bin-scala_2.11.tgz

然后，我们可以直接运行脚本启动 Flink ：

➜  [flink-1.10.0]# ./bin/start-cluster.sh

![在这里插入图片描述][20200903101004668.png_pic_center]  
上图显示我们的 Flink 启动成功。

我们直接访问本地的 8081 端口，可以看到 Flink 的后台管理界面，验证 Flink 是否成功启动。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 17]  
可以看到 Flink 已经成功启动。当然，我们也可以查看运行日志来确认 Flink 是不是成功启动了，在 log 目录下有程序的启动日志。

我们尝试提交一个测试任务：

./bin/flink run examples/batch/WordCount.jar

我们在控制台直接看到输出。同样，在 Flink 的后台管理界面 Completed Jobs 一栏可以看到刚才提交执行的程序。

**Standalone 模式**

Standalone 模式是集群模式的一种，但是这种模式一般并不运行在生产环境中，原因和 on yarn 模式相比：

*  Standalone 模式的部署相对简单，可以支持小规模，少量的任务运行；
 *  Stabdalone 模式缺少系统层面对集群中 Job 的管理，容易遭成资源分配不均匀；
 *  资源隔离相对简单，任务之间资源竞争严重。

我们在 3 台虚拟机之间搭建 standalone 集群：  
![在这里插入图片描述][20200903101605383.png_pic_center]  
在 master 节点，将 [Apache Flink 1.10.0 for Scala 2.11][] 包进行解压：

➜  [SoftWare]# tar -zxvf flink-1.10.0-bin-scala_2.11.tgz

重点来啦，我们需要修改 Flink 的配置文件，并且将修改好的解压目录完整的拷贝到两个从节点中去。在这里，我强烈建议主节点和从节点的目录要保持一致。

我们修改 conf 目录下的 flink-conf.yaml:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 18]  
flink-conf.yaml 文件中有大量的配置参数，我们挑选其中必填的最基本参数进行修改：

jobmanager.rpc.address: master
    jobmanager.heap.size: 1024m
    jobmanager.rpc.port: 6123
    taskmanager.memory.process.size: 1568m
    taskmanager.numberOfTaskSlots: 1
    parallelism.default: 1
    jobmanager.execution.failover-strategy: region
    io.tmp.dirs: /tmp

它们分别代表：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 19]  
如果你对其他的参数有兴趣的话，可以直接参考[官网][Link 3]。

接下来我们修改 conf 目录下的 master 和 slave 文件。

vim master，将内容修改为：

master

vim slave，将内容修改为：

slave01
    slave02

然后，将整个修改好的 Flink 解压目录使用 scp 远程拷贝命令发送到从节点：

scp -r /SoftWare/flink-1.10.0 slave01:/SoftWare/
    scp -r /SoftWare/flink-1.10.0 slave02:/SoftWare/

在 master、slave01、slave02 上分别配置环境变量，vim /etc/profile，将内容修改为：

export FLINK_HOME=/SoftWare/flink-1.10.0
    export PATH=$PATH:$FLINK_HOME/bin

到此为止，我们整个的基础配置已经完成，下面需要启动集群，登录 master 节点执行：

/SoftWare/flink-1.10.0/bin/start-cluster.sh

可以在浏览器访问：http://master-ip:8081/ 检查集群是否启动成功。

集群搭建过程中，可能出现的问题：

*  端口被占用，我们需要手动杀掉占用端口的程序；
 *  目录找不到或者文件找不到，我们在 flink-conf.yaml 中配置过 io.tmp.dirs ，这个目录需要手动创建。

**On Yarn 模式和 HA 配置**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 20]  
上图是 Flink on Yarn 模式下，Flink 和 Yarn 的交互流程。Yarn 是 Hadoop 三驾马车之一，主要用来做资源管理。我们在 Flink on Yarn 模式中也是借助 Yarn 的资源管理优势，需要在三个节点中配置 YARN\_CONF\_DIR、HADOOP\_CONF\_DIR、HADOOP\_CONF\_PATH 中的任意一个环境变量即可。

本课时中集群的高可用 HA 配置是基于独立的 ZooKeeper 集群。当然，Flink 本身提供了内置 ZooKeeper 插件，可以直接修改 conf/zoo.cfg，并且使用 /bin/start-zookeeper-quorum.sh 直接启动。

环境准备：

*  ZooKeeper-3.x
 *  Flink-1.10.0
 *  Hadoop-2.6.5

我们使用 5 台虚拟机搭建 on yarn 的高可用集群：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 21]  
如果你在使用 Flink 的最新版本 1.10.0 时，那么需要在本地安装 Hadoop 环境并进行下面的操作。

首先，添加环境变量：

vi /etc/profile
    # 添加环境变量
    export HADOOP_CONF_DIR=/Software/hadoop-2.6.5/etc/hadoop
    # 环境变量生效
    source /etc/profile

其次，下载对应的的依赖包，并将对应的 Hadoop 依赖复制到 flink 的 lib 目录下，对应的 hadoop 依赖可以在[这里][Link 4]下载。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 22]  
与 standalone 集群不同的是，我们需要修改 flink-conf.yaml 文件中的一些配置：

high-availability: zookeeper
    high-availability.storageDir: hdfs://cluster/flinkha/
    high-availability.zookeeper.quorum: slave01:2181,slave02:2181,slave03:2181

它们分别代表：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 23]  
然后分别修改 master、slave、zoo.cfg 三个配置文件。  
vim master，将内容修改为：

master01:8081
    master02:8081

vim slave，将内容修改为：

slave01
    slave02
    slave03

vim zoo.cfg，将内容修改为：

server.1=slave01:2888:3888
    server.2=slave02:2888:3888
    server.3=slave03:2888:3888

然后，我们将整个修改好的 Flink 解压目录使用 scp 远程拷贝命令发送到从节点：

scp -r /SoftWare/flink-1.10.0 slave01:/SoftWare/
    scp -r /SoftWare/flink-1.10.0 slave02:/SoftWare/
    scp -r /SoftWare/flink-1.10.0 slave03:/SoftWare/

分别启动 Hadoop 和 ZooKeeper，然后在主节点，使用命令启动集群：

/SoftWare/flink-1.10.0/bin/start-cluster.sh

我们同样直接访问 http://192.168.2.100:8081/ 端口，可以看到 Flink 的后台管理界面，验证 Flink 是否成功启动。

在 Flink on yarn 模式下，启动集群的方式有两种：

*  直接在 yarn 上运行任务
 *  yarn session 模式

直接在 yarn 上运行任务相当于将 job 直接提交到 yarn 上，每个任务会根据用户的指定进行资源申请，任务之间互不影响。

./bin/flink run -yjm 1024m -ytm 4096m -ys 2  ./examples/batch/WordCount.jar

更多关于参数的含义，可以参考[官网][Link 5]。

使用 yarn session 模式，我们需要先启动一个 yarn-session 会话，相当于启动了一个 yarn 任务，这个任务所占用的资源不会变化，并且一直运行。我们在使用 flink run 向这个 session 任务提交作业时，如果 session 的资源不足，那么任务会等待，直到其他资源释放。当这个 yarn-session 被杀死时，所有任务都会停止。

例如我们启动一个 yarn session 任务，该任务拥有 8G 内存、32 个槽位。

./bin/yarn-session.sh -tm 8192 -s 32

我们在 yarn 的界面上可以看到这个任务的 ID，然后向这个 session ID 提交 Flink 任务：

./bin/flink run -m yarn-cluster -yid application_xxxx ./examples/batch/WordCount.jar

其中，application\_xxxx 即为上述的 yarn session 任务 ID。

# 第七讲 Flink常用核心概念分析 #

在 Flink 这个框架中，有很多独有的概念，比如分布式缓存、重启策略、并行度等，这些概念是我们在进行任务开发和调优时必须了解的。

**分布式缓存**

熟悉 Hadoop 的你应该知道，分布式缓存最初的思想诞生于 Hadoop 框架，Hadoop 会将一些数据或者文件缓存在 HDFS 上，在分布式环境中让所有的计算节点调用同一个配置文件。在 Flink 中，Flink 框架开发者们同样将这个特性进行了实现。

Flink 提供的分布式缓存类型 Hadoop，目的是为了在分布式环境中让每一个 TaskManager 节点保存一份相同的数据或者文件，当前计算节点的 task 就像读取本地文件一样拉取这些配置。

分布式缓存在我们实际生产环境中最广泛的一个应用，就是在进行表与表 Join 操作时，如果一个表很大，另一个表很小，那么我们就可以把较小的表进行缓存，在每个 TaskManager 都保存一份，然后进行 Join 操作。

那么我们应该怎样使用 Flink 的分布式缓存呢？举例如下：

public static void main(String[] args) throws Exception { 
    
    	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
       	env.registerCachedFile("/Users/wangzhiwu/WorkSpace/quickstart/distributedcache.txt", "distributedCache");
        //1：注册一个文件,可以使用hdfs上的文件 也可以是本地文件进行测试
        DataSource<String> data = env.fromElements("Linea", "Lineb", "Linec", "Lined");
    
        DataSet<String> result = data.map(new RichMapFunction<String, String>() { 
            private ArrayList<String> dataList = new ArrayList<String>();
    
            @Override
            public void open(Configuration parameters) throws Exception { 
                super.open(parameters);
                //2：使用该缓存文件
                File myFile = getRuntimeContext().getDistributedCache().getFile("distributedCache");
                List<String> lines = FileUtils.readLines(myFile);
                for (String line : lines) { 
                    this.dataList.add(line);
                    System.err.println("分布式缓存为:" + line);
                }
            }
    
            @Override
            public String map(String value) throws Exception { 
                //在这里就可以使用dataList
                System.err.println("使用datalist：" + dataList + "-------" +value);
                //业务逻辑
                return dataList +"：" +  value;
            }
        });
    
        result.printToErr();
    }

从上面的例子中可以看出，使用分布式缓存有两个步骤。

*  第一步：首先需要在 env 环境中注册一个文件，该文件可以来源于本地，也可以来源于 HDFS ，并且为该文件取一个名字。
 *  第二步：在使用分布式缓存时，可根据注册的名字直接获取。

可以看到，在上述案例中，我们把一个本地的 distributedcache.txt 文件注册为 distributedCache，在下面的 map 算子中直接通过这个名字将缓存文件进行读取并且进行了处理。

我们直接运行该程序，在控制台可以看到如下输出：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 24]  
在使用分布式缓存时也需要注意一些问题，需要我们缓存的文件在任务运行期间最好是只读状态，否则会造成数据的一致性问题。另外，缓存的文件和数据不宜过大，否则会影响 Task 的执行速度，在极端情况下会造成 OOM。

**故障恢复和重启策略**

自动故障恢复是 Flink 提供的一个强大的功能，在实际运行环境中，我们会遇到各种各样的问题从而导致应用挂掉，比如我们经常遇到的非法数据、网络抖动等。

Flink 提供了强大的可配置故障恢复和重启策略来进行自动恢复。

**故障恢复**

我们在上一课时中介绍过 Flink 的配置文件，其中有一个参数 `jobmanager.execution.failover-strategy: region`。

Flink 支持了不同级别的故障恢复策略，`jobmanager.execution.failover-strategy` 的可配置项有两种：`full` 和 `region`。

当我们配置的故障恢复策略为 full 时，集群中的 Task 发生故障，那么该任务的所有 Task 都会发生重启。而在实际生产环境中，我们的大作业可能有几百个 Task，出现一次异常如果进行整个任务重启，那么经常会导致长时间任务不能正常工作，导致数据延迟。

但是事实上，我们可能只是集群中某一个或几个 Task 发生了故障，只需要重启有问题的一部分即可，这就是 Flink 基于 Region 的局部重启策略。在这个策略下，Flink 会把我们的任务分成不同的 Region，当某一个 Task 发生故障时，Flink 会计算需要故障恢复的最小 Region。

Flink 在判断需要重启的 Region 时，采用了以下的判断逻辑：

*  发生错误的 Task 所在的 Region 需要重启；
 *  如果当前 Region 的依赖数据出现损坏或者部分丢失，那么生产数据的 Region 也需要重启；
 *  为了保证数据一致性，当前 Region 的下游 Region 也需要重启。

**重启策略**

Flink 提供了多种类型和级别的重启策略，常用的重启策略包括：

*  固定延迟重启策略模式
 *  失败率重启策略模式
 *  无重启策略模式

Flink 在判断使用的哪种重启策略时做了默认约定，如果用户配置了 checkpoint，但没有设置重启策略，那么会按照固定延迟重启策略模式进行重启；如果用户没有配置 checkpoint，那么默认不会重启。

下面我们分别对这三种模式进行详细讲解。

**无重启策略模式**

在这种情况下，如果我们的作业发生错误，任务会直接退出。  
我们可以在 flink-conf.yaml 中配置：

restart-strategy: none

也可以在程序中使用代码指定：

final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    env.setRestartStrategy(RestartStrategies.noRestart());

**固定延迟重启策略模式**

固定延迟重启策略会通过在 flink-conf.yaml 中设置如下配置参数，来启用此策略：

restart-strategy: fixed-delay

固定延迟重启策略模式需要指定两个参数，首先 Flink 会根据用户配置的重试次数进行重试，每次重试之间根据配置的时间间隔进行重试，如下表所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 25]  
举个例子，假如我们需要任务重试 3 次，每次重试间隔 5 秒，那么需要进行一下配置：

restart-strategy.fixed-delay.attempts: 3
    restart-strategy.fixed-delay.delay: 5 s

当前我们也可以在代码中进行设置：

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
            3, // 重启次数
            Time.of(5, TimeUnit.SECONDS) // 时间间隔
    ));

**失败率重启策略模式**

首先我们在 flink-conf.yaml 中指定如下配置：

restart-strategy: failure-rate

这种重启模式需要指定三个参数，如下表所示。失败率重启策略在 Job 失败后会重启，但是超过失败率后，Job 会最终被认定失败。在两个连续的重启尝试之间，重启策略会等待一个固定的时间。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 26]  
这种策略的配置理解较为困难，我们举个例子，假如 5 分钟内若失败了 3 次，则认为该任务失败，每次失败的重试间隔为 5 秒。

那么我们的配置应该是：

restart-strategy.failure-rate.max-failures-per-interval: 3
    restart-strategy.failure-rate.failure-rate-interval: 5 min
    restart-strategy.failure-rate.delay: 5 s

当然，也可以在代码中直接指定：

env.setRestartStrategy(RestartStrategies.failureRateRestart(
            3, // 每个时间间隔的最大故障次数
            Time.of(5, TimeUnit.MINUTES), // 测量故障率的时间间隔
            Time.of(5, TimeUnit.SECONDS) // 每次任务失败时间间隔
    ));

最后，需要注意的是，在实际生产环境中由于每个任务的负载和资源消耗不一样，我们推荐在代码中指定每个任务的重试机制和重启策略。

**并行度**

并行度是 Flink 执行任务的核心概念之一，它被定义为在分布式运行环境中我们的一个算子任务被切分成了多少个子任务并行执行。我们提高任务的并行度（Parallelism）在很大程度上可以大大提高任务运行速度。

一般情况下，我们可以通过四种级别来设置任务的并行度。

*  算子级别

在代码中可以调用 setParallelism 方法来设置每一个算子的并行度。例如：

DataSet<Tuple2<String, Integer>> counts =
          text.flatMap(new LineSplitter())
                .groupBy(0)
                .sum(1).setParallelism(1);

事实上，Flink 的每个算子都可以单独设置并行度。这也是我们最推荐的一种方式，可以针对每个算子进行任务的调优。

*  执行环境级别

我们在创建 Flink 的上下文时可以显示的调用 env.setParallelism() 方法，来设置当前执行环境的并行度，这个配置会对当前任务的所有算子、Source、Sink 生效。当然你还可以在算子级别设置并行度来覆盖这个设置。

final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(5);

*  提交任务级别

用户在提交任务时，可以显示的指定 -p 参数来设置任务的并行度，例如：

./bin/flink run -p 10 WordCount.jar

*  系统配置级别

我们在上一课时中提到了 flink-conf.yaml 中的一个配置：parallelism.default，该配置即是在系统层面设置所有执行环境的并行度配置。

整体上讲，这四种级别的配置生效优先级如下：**算子级别 > 执行环境级别 > 提交任务级别 > 系统配置级别**。

在这里，要特别提一下 Flink 中的 Slot 概念。我们知道，Flink 中的 TaskManager 是执行任务的节点，那么在每一个 TaskManager 里，还会有“槽位”，也就是 Slot。Slot 个数代表的是每一个 TaskManager 的并发执行能力。

假如我们指定 taskmanager.numberOfTaskSlots:3，即每个 taskManager 有 3 个 Slot ，那么整个集群就有 3 \* taskManager 的个数多的槽位。这些槽位就是我们整个集群所拥有的所有执行任务的资源。

# 第八讲 Flink窗口、时间和水印 #

**Flink 的窗口和时间**

根据窗口数据划分的不同，目前 Flink 支持如下 3 种：

*  滚动窗口，窗口数据有固定的大小，窗口中的数据不会叠加；
 *  滑动窗口，窗口数据有固定的大小，并且有生成间隔；
 *  会话窗口，窗口数据没有固定的大小，根据用户传入的参数进行划分，窗口数据无叠加。

Flink 中的时间分为三种：

*  事件时间（Event Time），即事件实际发生的时间；
 *  摄入时间（Ingestion Time），事件进入流处理框架的时间；
 *  处理时间（Processing Time），事件被处理的时间。

下面的图详细说明了这三种时间的区别和联系：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 27]  
**事件时间（Event Time）**

事件时间（Event Time）指的是数据产生的时间，这个时间一般由数据生产方自身携带，比如 Kafka 消息，每个生成的消息中自带一个时间戳代表每条数据的产生时间。Event Time 从消息的产生就诞生了，不会改变，也是我们使用最频繁的时间。

利用 Event Time 需要指定如何生成事件时间的“水印”，并且一般和窗口配合使用，具体会在下面的“水印”内容中详细讲解。

我们可以在代码中指定 Flink 系统使用的时间类型为 EventTime：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //设置时间属性为 EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
    DataStream<MyEvent> stream = env.addSource(new FlinkKafkaConsumer09<MyEvent>(topic, schema, props));
    
    stream
        .keyBy( (event) -> event.getUser() )
        .timeWindow(Time.hours(1))
        .reduce( (a, b) -> a.add(b) )
        .addSink(...);

Flink 注册 EventTime 是通过 InternalTimerServiceImpl.registerEventTimeTimer 来实现的：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 28]  
可以看到，该方法有两个入参：namespace 和 time，其中 time 是触发定时器的时间，namespace 则被构造成为一个 TimerHeapInternalTimer 对象，然后将其放入 KeyGroupedInternalPriorityQueue 队列中。

那么 Flink 什么时候会使用这些 timer 触发计算呢？答案在这个方法里：

InternalTimeServiceImpl.advanceWatermark。
    public void advanceWatermark(long time) throws Exception { 
       currentWatermark = time;
    
       InternalTimer<K, N> timer;
    
       while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) { 
          eventTimeTimersQueue.poll();
          keyContext.setCurrentKey(timer.getKey());
          triggerTarget.onEventTime(timer);
       }
    }

这个方法中的 while 循环部分会从 eventTimeTimersQueue 中依次取出触发时间小于参数 time 的所有定时器，调用 triggerTarget.onEventTime() 方法进行触发。

这就是 EventTime 从注册到触发的流程。

**处理时间（Processing Time）**

处理时间（Processing Time）指的是数据被 Flink 框架处理时机器的系统时间，Processing Time 是 Flink 的时间系统中最简单的概念，但是这个时间存在一定的不确定性，比如消息到达处理节点延迟等影响。

我们同样可以在代码中指定 Flink 系统使用的时间为 Processing Time：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

同样，也可以在源码中找到 Flink 是如何注册和使用 Processing Time 的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 29]  
registerProcessingTimeTimer() 方法为我们展示了如何注册一个 ProcessingTime 定时器：  
每当一个新的定时器被加入到 processingTimeTimersQueue 这个优先级队列中时，如果新来的 Timer 时间戳更小，那么更小的这个 Timer 会被重新注册 ScheduledThreadPoolExecutor 定时执行器上。

Processing Time 被触发是在 InternalTimeServiceImpl 的 onProcessingTime() 方法中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 30]  
一直循环获取时间小于入参 time 的所有定时器，并运行 triggerTarget 的 onProcessingTime() 方法。

**摄入时间（Ingestion Time）**

摄入时间（Ingestion Time）是事件进入 Flink 系统的时间，在 Flink 的 Source 中，每个事件会把当前时间作为时间戳，后续做窗口处理都会基于这个时间。理论上 Ingestion Time 处于 Event Time 和 Processing Time之间。

> 与事件时间相比，摄入时间无法处理延时和无序的情况，但是不需要明确执行如何生成 watermark。在系统内部，摄入时间采用更类似于事件时间的处理方式进行处理，但是有自动生成的时间戳和自动的 watermark。

> 可以防止 Flink 内部处理数据是发生乱序的情况，但无法解决数据到达 Flink 之前发生的乱序问题。如果需要处理此类问题，建议使用 EventTime。

Ingestion Time 的时间类型生成相关的代码在 AutomaticWatermarkContext 中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 31]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 32]  
我们可以看出，这里会设置一个 watermark 发送定时器，在 watermarkInterval 时间之后触发。

处理数据的代码在 processAndCollect() 方法中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 33]  
**水印（WaterMark）**

水印（WaterMark）是 Flink 框架中最晦涩难懂的概念之一，有很大一部分原因是因为翻译的原因。

WaterMark 在正常的英文翻译中是水位，但是在 Flink 框架中，翻译为“水位线”更为合理，它在本质上是一个时间戳。

在上面的时间类型中我们知道，Flink 中的时间：

*  EventTime 每条数据都携带时间戳；
 *  ProcessingTime 数据不携带任何时间戳的信息；
 *  IngestionTime 和 EventTime 类似，不同的是 Flink 会使用系统时间作为时间戳绑定到每条数据，可以防止 Flink 内部处理数据是发生乱序的情况，但无法解决数据到达 Flink 之前发生的乱序问题。

所以，我们在处理消息乱序的情况时，会用 EventTime 和 WaterMark 进行配合使用。

首先我们要明确几个基本问题。

**水印的本质是什么**

水印的出现是为了**解决实时计算中的数据乱序问题**，它的本质是 DataStream 中一个带有时间戳的元素。如果 Flink 系统中出现了一个 WaterMark T，那么就意味着 EventTime < T 的数据都已经到达，窗口的结束时间和 T 相同的那个窗口被触发进行计算了。

也就是说：水印是 Flink 判断迟到数据的标准，同时也是窗口触发的标记。

在程序并行度大于 1 的情况下，会有多个流产生水印和窗口，这时候 Flink 会选取时间戳最小的水印。

**水印是如何生成的**

Flink 提供了 assignTimestampsAndWatermarks() 方法来实现水印的提取和指定，该方法接受的入参有 AssignerWithPeriodicWatermarks 和 AssignerWithPunctuatedWatermarks 两种。

整体的类图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 34]  
**水印种类**

**周期性水印**

我们在使用 AssignerWithPeriodicWatermarks 周期生成水印时，周期默认的时间是 200ms，这个时间的指定位置为：

@PublicEvolving
    public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) { 
        this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
        if (characteristic == TimeCharacteristic.ProcessingTime) { 
            getConfig().setAutoWatermarkInterval(0);
        } else { 
            getConfig().setAutoWatermarkInterval(200);
        }
    }

是否还记得上面我们在讲时间类型时会通过 env.setStreamTimeCharacteristic() 方法指定 Flink 系统的时间类型，这个 setStreamTimeCharacteristic() 方法中会做判断，如果用户传入的是 TimeCharacteristic.eventTime 类型，那么 AutoWatermarkInterval 的值则为 200ms ，如上述代码所示。当前我们也可以使用 ExecutionConfig.setAutoWatermarkInterval() 方法来指定自动生成的时间间隔。

在上述的类图中可以看出，我们需要通过 TimestampAssigner 的 extractTimestamp() 方法来提取 EventTime。

Flink 在这里提供了 3 种提取 EventTime() 的方法，分别是：

*  AscendingTimestampExtractor
 *  BoundedOutOfOrdernessTimestampExtractor
 *  IngestionTimeExtractor

这三种方法中 BoundedOutOfOrdernessTimestampExtractor() 用的最多，需特别注意，在这个方法中的 maxOutOfOrderness 参数，该参数指的是允许数据乱序的时间范围。简单说，这种方式允许数据迟到 maxOutOfOrderness 这么长的时间。

public BoundedOutOfOrdernessTimestampExtractor(Time maxOutOfOrderness) { 
            if (maxOutOfOrderness.toMilliseconds() < 0) { 
                throw new RuntimeException("Tried to set the maximum allowed " +
                    "lateness to " + maxOutOfOrderness + ". This parameter cannot be negative.");
            }
            this.maxOutOfOrderness = maxOutOfOrderness.toMilliseconds();
            this.currentMaxTimestamp = Long.MIN_VALUE + this.maxOutOfOrderness;
        }
    
        public abstract long extractTimestamp(T element);
    
        @Override
        public final Watermark getCurrentWatermark() { 
            long potentialWM = currentMaxTimestamp - maxOutOfOrderness;
            if (potentialWM >= lastEmittedWatermark) { 
                lastEmittedWatermark = potentialWM;
            }
            return new Watermark(lastEmittedWatermark);
        }
    
        @Override
        public final long extractTimestamp(T element, long previousElementTimestamp) { 
            long timestamp = extractTimestamp(element);
            if (timestamp > currentMaxTimestamp) { 
                currentMaxTimestamp = timestamp;
            }
            return timestamp;
        }

**PunctuatedWatermark 水印**

这种水印的生成方式 Flink 没有提供内置实现，它适用于根据接收到的消息判断是否需要产生水印的情况，用这种水印生成的方式并不多见。

举个简单的例子，假如我们发现接收到的数据 MyData 中以字符串 watermark 开头则产生一个水印：

data.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<UserActionRecord>() { 
    
          @Override
          public Watermark checkAndGetNextWatermark(MyData data, long l) { 
            return data.getRecord.startsWith("watermark") ? new Watermark(l) : null;
          }
    
          @Override
          public long extractTimestamp(MyData data, long l) { 
            return data.getTimestamp();
          }
        });
    
    class MyData{ 
        private String record;
        private Long timestamp;
        public String getRecord() { 
            return record;
        }
        public void setRecord(String record) { 
            this.record = record;
        }
        public Timestamp getTimestamp() { 
            return timestamp;
        }
        public void setTimestamp(Timestamp timestamp) { 
            this.timestamp = timestamp;
        }
    }

**案例**

我们上面讲解了 Flink 关于水印和时间的生成，以及使用，下面举一个例子来讲解。

模拟一个实时接收 Socket 的 DataStream 程序，代码中使用 AssignerWithPeriodicWatermarks 来设置水印，将接收到的数据进行转换，分组并且在一个 5 秒的窗口内获取该窗口中第二个元素最小的那条数据。

public static void main(String[] args) throws Exception { 
    
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
    
        //设置为eventtime事件类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置水印生成时间间隔100ms
        env.getConfig().setAutoWatermarkInterval(100);
    
        DataStream<String> dataStream = env
                .socketTextStream("127.0.0.1", 9000)
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() { 
                    private Long currentTimeStamp = 0L;
                    //设置允许乱序时间
                    private Long maxOutOfOrderness = 5000L;
    
                    @Override
                    public Watermark getCurrentWatermark() { 
    
                        return new Watermark(currentTimeStamp - maxOutOfOrderness);
                    }
    
                    @Override
                    public long extractTimestamp(String s, long l) { 
                        String[] arr = s.split(",");
                        long timeStamp = Long.parseLong(arr[1]);
                        currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
                        System.err.println(s + ",EventTime:" + timeStamp + ",watermark:" + (currentTimeStamp - maxOutOfOrderness));
                        return timeStamp;
                    }
                });
    
        dataStream.map(new MapFunction<String, Tuple2<String, Long>>() { 
            @Override
            public Tuple2<String, Long> map(String s) throws Exception { 
    
                String[] split = s.split(",");
                return new Tuple2<String, Long>(split[0], Long.parseLong(split[1]));
            }
        })
                .keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .minBy(1)
                .print();
    
        env.execute("WaterMark Test Demo");
    
    }

我们第一次试验的数据如下：

flink,1588659181000
    flink,1588659182000
    flink,1588659183000
    flink,1588659184000
    flink,1588659185000

可以做一个简单的判断，第一条数据的时间戳为 1588659181000，窗口的大小为 5 秒，那么应该会在 flink,1588659185000 这条数据出现时触发窗口的计算。

我们用 nc -lk 9000 命令启动端口，然后输出上述试验数据，看到控制台的输出：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 35]  
很明显，可以看到当第五条数据出现后，窗口触发了计算。

下面再模拟一下数据乱序的情况，假设我们的数据来源如下：

flink,1588659181000
    flink,1588659182000
    flink,1588659183000
    flink,1588659184000
    flink,1588659185000
    flink,1588659180000
    flink,1588659186000
    flink,1588659187000
    flink,1588659188000
    flink,1588659189000
    flink,1588659190000

其中的 flink,1588659180000 为乱序消息，来看看会发生什么？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 36]  
可以看到，时间戳为 1588659180000 的这条消息并没有被处理，因为此时代码中的允许乱序时间 private Long maxOutOfOrderness = 0L 即不处理乱序消息。

下面修改 private Long maxOutOfOrderness = 5000L，即代表允许消息的乱序时间为 5 秒，然后把同样的数据发往 socket 端口。

可以看到，我们把所有数据发送出去仅触发了一次窗口计算，并且输出的结果中 watermark 的时间往后顺延了 5 秒钟。所以，maxOutOfOrderness 的设置会影响窗口的计算时间和水印的时间，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 37]  
假如我们继续向 socket 中发送数据：

flink,1588659191000
    flink,1588659192000
    flink,1588659193000
    flink,1588659194000
    flink,1588659195000

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 38]  
在这里要特别说明，Flink 在用时间 + 窗口 + 水印来解决实际生产中的数据乱序问题，有如下的触发条件：

*  watermark 时间 >= window\_end\_time；
 *  在 \[window\_start\_time,window\_end\_time) 中有数据存在，这个窗口是左闭右开的。

此外，因为 WaterMark 的生成是以对象的形式发送到下游，同样会消耗内存，因此水印的生成时间和频率都要进行严格控制，否则会影响我们的正常作业。

# 第九讲 Flink状态与容错 #

在 Flink 的框架中，进行有状态的计算是 Flink 最重要的特性之一。所谓的状态，其实指的是 Flink 程序的中间计算结果。Flink 支持了不同类型的状态，并且针对状态的持久化还提供了专门的机制和状态管理器。

**状态**

我们在 Flink 的官方博客中找到这样一段话，可以认为这是对状态的定义：

> When working with state, it might also be useful to read about Flink’s state backends. Flink provides different state backends that specify how and where state is stored. State can be located on Java’s heap or off-heap. Depending on your state backend, Flink can also manage the state for the application, meaning Flink deals with the memory management (possibly spilling to disk if necessary) to allow applications to hold very large state. State backends can be configured without changing your application logic.

这段话告诉我们，所谓的状态指的是，在流处理过程中那些需要记住的数据，而这些数据既可以包括业务数据，也可以包括元数据。Flink 本身提供了不同的状态管理器来管理状态，并且这个状态可以非常大。

Flink 的状态数据可以存在 JVM 的堆内存或者堆外内存中，当然也可以借助第三方存储，例如 Flink 已经实现的对 RocksDB 支持。Flink 的官网同样给出了适用于状态计算的几种情况：

*  When an application searches for certain event patterns, the state will store the sequence of events encountered so far
 *  When aggregating events per minute/hour/day, the state holds the pending aggregates
 *  When training a machine learning model over a stream of data points, the state holds the current version of the model parameters
 *  When historic data needs to be managed, the state allows efficient access to events that occurred in the past

以上四种情况分别是：复杂事件处理获取符合某一特定时间规则的事件、聚合计算、机器学习的模型训练、使用历史的数据进行计算。

**Flink 状态分类和使用**

我们在之前的课时中提到过 KeyedStream 的概念，并且介绍过 KeyBy 这个算子的使用。在 Flink 中，根据数据集是否按照某一个 Key 进行分区，将状态分为 Keyed State 和 Operator State（Non-Keyed State）两种类型。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 39]  
如上图所示，Keyed State 是经过分区后的流上状态，每个 Key 都有自己的状态，图中的八边形、圆形和三角形分别管理各自的状态，并且只有指定的 key 才能访问和更新自己对应的状态。

与 Keyed State 不同的是，Operator State 可以用在所有算子上，每个算子子任务或者说每个算子实例共享一个状态，流入这个算子子任务的数据可以访问和更新这个状态。每个算子子任务上的数据共享自己的状态。

但是有一点需要说明的是，无论是 Keyed State 还是 Operator State，Flink 的状态都是基于本地的，即每个算子子任务维护着这个算子子任务对应的状态存储，算子子任务之间的状态不能相互访问。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 40]  
我们可以看一下 State 的类图，对于 Keyed State，Flink 提供了几种现成的数据结构供我们使用，State 主要有四种实现，分别为 ValueState、MapState、AppendingState 和 ReadOnlyBrodcastState ，其中 AppendingState 又可以细分为 ReducingState、AggregatingState 和 ListState。

那么我们怎么访问这些状态呢？Flink 提供了 StateDesciptor 方法专门用来访问不同的 state，类图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 41]  
下面演示一下如何使用 StateDesciptor 和 ValueState，代码如下：

public static void main(String[] args) throws Exception { 
    
       final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
       env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 5L), Tuple2.of(1L, 2L))
             .keyBy(0)
             .flatMap(new CountWindowAverage())
             .printToErr();
    
           env.execute("submit job");
    
    }
    
    
       public static class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> { 
    
           private transient ValueState<Tuple2<Long, Long>> sum;
           public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception { 
    
               Tuple2<Long, Long> currentSum;
               // 访问ValueState
               if(sum.value()==null){ 
                   currentSum = Tuple2.of(0L, 0L);
               }else { 
                   currentSum = sum.value();
               }
    
               // 更新
               currentSum.f0 += 1;
    
               // 第二个元素加1
               currentSum.f1 += input.f1;
    
               // 更新state
               sum.update(currentSum);
    
               // 如果count的值大于等于2，求知道并清空state
               if (currentSum.f0 >= 2) { 
                   out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
                   sum.clear();
               }
       }
    
    
       public void open(Configuration config) { 
           ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                   new ValueStateDescriptor<>(
                           "average", // state的名字
                           TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() { })
                           ); // 设置默认值
    
    
           StateTtlConfig ttlConfig = StateTtlConfig
                   .newBuilder(Time.seconds(10))
                   .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                   .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                   .build();
    
           descriptor.enableTimeToLive(ttlConfig);
    
           sum = getRuntimeContext().getState(descriptor);
           }
    }

在上述例子中，我们通过继承 RichFlatMapFunction 来访问 State，通过 getRuntimeContext().getState(descriptor) 来获取状态的句柄。而真正的访问和更新状态则在 Map 函数中实现。

我们这里的输出条件为，每当第一个元素的和达到二，就把第二个元素的和与第一个元素的和相除，最后输出。我们直接运行，在控制台可以看到结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 42]  
Operator State 的实际应用场景不如 Keyed State 多，一般来说它会被用在 Source 或 Sink 等算子上，用来保存流入数据的偏移量或对输出数据做缓存，以保证 Flink 应用的 Exactly-Once 语义。

同样，我们对于任何状态数据还可以设置它们的过期时间。如果一个状态设置了 TTL，并且已经过期，那么我们之前保存的值就会被清理。

想要使用 TTL，我们需要首先构建一个 StateTtlConfig 配置对象；然后，可以通过传递配置在任何状态描述符中启用 TTL 功能。

StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.seconds(10))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();
    
    descriptor.enableTimeToLive(ttlConfig);

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 43]  
StateTtlConfig 这个类中有一些配置需要我们注意：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 44]  
UpdateType 表明了过期时间什么时候更新，而对于那些过期的状态，是否还能被访问则取决于 StateVisibility 的配置。

**状态后端种类和配置**

我们在上面的内容中讲到了 Flink 的状态数据可以存在 JVM 的堆内存或者堆外内存中，当然也可以借助第三方存储。默认情况下，Flink 的状态会保存在 taskmanager 的内存中，Flink 提供了三种可用的状态后端用于在不同情况下进行状态后端的保存。

*  MemoryStateBackend
 *  FsStateBackend
 *  RocksDBStateBackend

**MemoryStateBackend**

顾名思义，MemoryStateBackend 将 state 数据存储在内存中，一般用来进行本地调试用，我们在使用 MemoryStateBackend 时需要注意的一些点包括：

> 每个独立的状态（state）默认限制大小为 5MB，可以通过构造函数增加容量  
> 状态的大小不能超过 akka 的 Framesize 大小  
> 聚合后的状态必须能够放进 JobManager 的内存中

MemoryStateBackend 可以通过在代码中显示指定：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStateBackend(new MemoryStateBackend(DEFAULT_MAX_STATE_SIZE,false));

其中，new MemoryStateBackend(DEFAULT\_MAX\_STATE\_SIZE,false) 中的 false 代表关闭异步快照机制。关于快照，我们在后面的课时中有单独介绍。

很明显 MemoryStateBackend 适用于我们本地调试使用，来记录一些状态很小的 Job 状态信息。

**FsStateBackend**

FsStateBackend 会把状态数据保存在 TaskManager 的内存中。CheckPoint 时，将状态快照写入到配置的文件系统目录中，少量的元数据信息存储到 JobManager 的内存中。

使用 FsStateBackend 需要我们指定一个文件路径，一般来说是 HDFS 的路径，例如，hdfs://namenode:40010/flink/checkpoints。

我们同样可以在代码中显示指定：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints", false));

FsStateBackend 因为将状态存储在了外部系统如 HDFS 中，所以它适用于大作业、状态较大、全局高可用的那些任务。

**RocksDBStateBackend**

RocksDBStateBackend 和 FsStateBackend 有一些类似，首先它们都需要一个外部文件存储路径，比如 HDFS 的 hdfs://namenode:40010/flink/checkpoints，此外也适用于大作业、状态较大、全局高可用的那些任务。

但是与 FsStateBackend 不同的是，RocksDBStateBackend 将正在运行中的状态数据保存在 RocksDB 数据库中，RocksDB 数据库默认将数据存储在 TaskManager 运行节点的数据目录下。

这意味着，RocksDBStateBackend 可以存储远超过 FsStateBackend 的状态，可以避免向 FsStateBackend 那样一旦出现状态暴增会导致 OOM，但是因为将状态数据保存在 RocksDB 数据库中，吞吐量会有所下降。

此外，需要注意的是，RocksDBStateBackend 是唯一支持增量快照的状态后端。

# 第十讲 Flink Side OutPut分流 #

**分流场景**

我们在生产实践中经常会遇到这样的场景，需把输入源按照需要进行拆分，比如我期望把订单流按照金额大小进行拆分，或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢？

**分流的方法**

通常来说针对不同的场景，有以下三种办法进行流的拆分。

**Filter 分流**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 45]  
Filter 方法我们在第 04 课时中（Flink 常用的 DataSet 和 DataStream API）讲过，这个算子用来根据用户输入的条件进行过滤，每个元素都会被 filter() 函数处理，如果 filter() 函数返回 true 则保留，否则丢弃。那么用在分流的场景，我们可以做多次 filter，把我们需要的不同数据生成不同的流。

来看下面的例子：

在上面的例子中我们使用 filter 算子将原始流进行了拆分，输入数据第一个元素为 0 的数据和第一个元素为 1 的数据分别被写入到了 zeroStream 和 oneStream 中，然后把两个流进行了打印。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 46]  
可以看到 zeroStream 和 oneStream 分别被打印出来。

Filter 的弊端是显而易见的，为了得到我们需要的流数据，需要多次遍历原始流，这样无形中浪费了我们集群的资源。

**Split 分流**

Split 也是 Flink 提供给我们将流进行切分的方法，需要在 split 算子中定义 OutputSelector，然后重写其中的 select 方法，将不同类型的数据进行标记，最后对返回的 SplitStream 使用 select 方法将对应的数据选择出来。

我们来看下面的例子：

public static void main(String[] args) throws Exception { 
    
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取数据源
        List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();
        data.add(new Tuple3<>(0,1,0));
        data.add(new Tuple3<>(0,1,1));
        data.add(new Tuple3<>(0,2,2));
        data.add(new Tuple3<>(0,1,3));
        data.add(new Tuple3<>(1,2,5));
        data.add(new Tuple3<>(1,2,9));
        data.add(new Tuple3<>(1,2,11));
        data.add(new Tuple3<>(1,2,13));
    
    
        DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);
    
    
        SplitStream<Tuple3<Integer, Integer, Integer>> splitStream = items.split(new OutputSelector<Tuple3<Integer, Integer, Integer>>() { 
            @Override
            public Iterable<String> select(Tuple3<Integer, Integer, Integer> value) { 
                List<String> tags = new ArrayList<>();
                if (value.f0 == 0) { 
                    tags.add("zeroStream");
                } else if (value.f0 == 1) { 
                    tags.add("oneStream");
                }
                return tags;
            }
        });
    
        splitStream.select("zeroStream").print();
        splitStream.select("oneStream").printToErr();
    
        //打印结果
        String jobName = "user defined streaming source";
        env.execute(jobName);
    }

同样，我们把来源的数据使用 split 算子进行了切分，并且打印出结果。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 47]  
但是要注意，使用 split 算子切分过的流，是不能进行二次切分的，假如把上述切分出来的 zeroStream 和 oneStream 流再次调用 split 切分，控制台会抛出以下异常。

Exception in thread "main" java.lang.IllegalStateException: Consecutive multiple splits are not supported. Splits are deprecated. Please use side-outputs.

这是什么原因呢？我们在源码中可以看到注释，该方式已经废弃并且建议使用最新的 SideOutPut 进行分流操作。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 48]  
**SideOutPut 分流**

SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法，在使用 SideOutPut 时，需要按照以下步骤进行：

*  定义 OutputTag
 *  调用特定函数进行数据拆分
    
     *  ProcessFunction
     *  KeyedProcessFunction
     *  CoProcessFunction
     *  KeyedCoProcessFunction
     *  ProcessWindowFunction
     *  ProcessAllWindowFunction

在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut：

public static void main(String[] args) throws Exception { 
    
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取数据源
        List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();
        data.add(new Tuple3<>(0,1,0));
        data.add(new Tuple3<>(0,1,1));
        data.add(new Tuple3<>(0,2,2));
        data.add(new Tuple3<>(0,1,3));
        data.add(new Tuple3<>(1,2,5));
        data.add(new Tuple3<>(1,2,9));
        data.add(new Tuple3<>(1,2,11));
        data.add(new Tuple3<>(1,2,13));
    
    
        DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);
    
        OutputTag<Tuple3<Integer,Integer,Integer>> zeroStream = new OutputTag<Tuple3<Integer,Integer,Integer>>("zeroStream") { };
        OutputTag<Tuple3<Integer,Integer,Integer>> oneStream = new OutputTag<Tuple3<Integer,Integer,Integer>>("oneStream") { };
    
    
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> processStream= items.process(new ProcessFunction<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() { 
            @Override
            public void processElement(Tuple3<Integer, Integer, Integer> value, Context ctx, Collector<Tuple3<Integer, Integer, Integer>> out) throws Exception { 
    
                if (value.f0 == 0) { 
                    ctx.output(zeroStream, value);
                } else if (value.f0 == 1) { 
                    ctx.output(oneStream, value);
                }
            }
        });
    
        DataStream<Tuple3<Integer, Integer, Integer>> zeroSideOutput = processStream.getSideOutput(zeroStream);
        DataStream<Tuple3<Integer, Integer, Integer>> oneSideOutput = processStream.getSideOutput(oneStream);
    
        zeroSideOutput.print();
        oneSideOutput.printToErr();
    
    
        //打印结果
        String jobName = "user defined streaming source";
        env.execute(jobName);
    }

可以看到，我们将流进行了拆分，并且成功打印出了结果。这里要注意，Flink 最新提供的 SideOutPut 方式拆分流是可以多次进行拆分的，无需担心会爆出异常。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 49]

# 第十一讲 Flink CEP复杂事件处理 #

Complex Event Processing（CEP）是 Flink 提供的一个非常亮眼的功能，关于 CEP 的解释我们引用维基百科中的一段话：

> CEP, is event processing that combines data from multiple sources to infer events or patterns that suggest more complicated circumstances. The goal of complex event processing is to identify meaningful events (such as opportunities or threats) and respond to them as quickly as possible.

在我们的实际生产中，随着数据的实时性要求越来越高，实时数据的量也在不断膨胀，在某些业务场景中需要根据连续的实时数据，发现其中有价值的那些事件。

说到底，Flink 的 CEP 到底解决了什么样的问题呢？

比如，我们需要在大量的订单交易中发现那些虚假交易，在网站的访问日志中寻找那些使用脚本或者工具“爆破”登录的用户，或者在快递运输中发现那些滞留很久没有签收的包裹等。

如果你对 CEP 的理论基础非常感兴趣，推荐一篇论文“Efﬁcient Pattern Matching over Event Streams”。

Flink 对 CEP 的支持非常友好，并且支持复杂度非常高的模式匹配，其吞吐和延迟都令人满意。

**程序结构**

Flink CEP 的程序结构主要分为两个步骤：

*  定义模式
 *  匹配结果

我们在官网中可以找到一个 Flink 提供的案例：

DataStream<Event> input = ...
    Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(
        new SimpleCondition<Event>() { 
            @Override
            public boolean filter(Event event) { 
                return event.getId() == 42;
            }
        }
    ).next("middle").subtype(SubEvent.class).where(
        new SimpleCondition<SubEvent>() { 
            @Override
            public boolean filter(SubEvent subEvent) { 
                return subEvent.getVolume() >= 10.0;
            }
        }
    ).followedBy("end").where(
         new SimpleCondition<Event>() { 
            @Override
            public boolean filter(Event event) { 
                return event.getName().equals("end");
            }
         }
    );
    PatternStream<Event> patternStream = CEP.pattern(input, pattern);
    DataStream<Alert> result = patternStream.process(
    new PatternProcessFunction<Event, Alert>() { 
        @Override
        public void processMatch(
                Map<String, List<Event>> pattern,
                Context ctx,
                Collector<Alert> out) throws Exception { 
            out.collect(createAlertFrom(pattern));
        }
    });

在这个案例中可以看到程序结构分别是：

*  第一步，定义一个模式 Pattern，在这里定义了一个这样的模式，即在所有接收到的事件中匹配那些以 id 等于 42 的事件，然后匹配 volume 大于 10.0 的事件，继续匹配一个 name 等于 end 的事件；
 *  第二步，匹配模式并且发出报警，根据定义的 pattern 在输入流上进行匹配，一旦命中我们的模式，就发出一个报警。

**模式定义**

Flink 支持了非常丰富的模式定义，这些模式也是我们实现复杂业务逻辑的基础。我们把支持的模式简单做了以下分类，完整的模式定义 API 支持可以参考[官网资料][Link 6]。

**简单模式**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 50]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 51]  
**联合模式**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 52]  
**匹配后的忽略模式**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 53]  
**源码解析**

我们在上面的官网案例中可以发现，Flink CEP 的整个过程是：

*  从一个 Source 作为输入
 *  经过一个 Pattern 算子转换为 PatternStream
 *  经过 select/process 算子转换为 DataStream

我们来看一下 select 和 process 算子都做了什么？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 54]  
可以看到最终的逻辑都是在 PatternStream 这个类中进行的。

public <R> SingleOutputStreamOperator<R> process(
          final PatternProcessFunction<T, R> patternProcessFunction,
          final TypeInformation<R> outTypeInfo) { 
       return builder.build(
          outTypeInfo,
          builder.clean(patternProcessFunction));
    }

最终经过 PatternStreamBuilder 的 build 方法生成了一个 SingleOutputStreamOperator，这个类继承了 DataStream。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 55]  
最终的处理计算逻辑其实都封装在了 CepOperator 这个类中，而在 CepOperator 这个类中的 processElement 方法则是对每一条数据的处理逻辑。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 56]  
同时由于 CepOperator 实现了 Triggerable 接口，所以会执行定时器。所有核心的处理逻辑都在 updateNFA 这个方法中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 57]  
入口在这里：

private void processEvent(NFAState nfaState, IN event, long timestamp) throws Exception { 
       try (SharedBufferAccessor<IN> sharedBufferAccessor = partialMatches.getAccessor()) { 
          Collection<Map<String, List<IN>>> patterns =
             nfa.process(sharedBufferAccessor, nfaState, event, timestamp, afterMatchSkipStrategy, cepTimerService);
          processMatchedSequences(patterns, timestamp);
       }
    }

NFA 的全称为 非确定有限自动机，NFA 中包含了模式匹配中的各个状态和状态间的转换。

在 NFA 这个类中的核心方法是：process 和 advanceTime，这两个方法的实现有些复杂，总体来说可以归纳为，每当一条新来的数据进入状态机都会驱动整个状态机进行状态转换。

**实战案例**

我们模拟电商网站用户搜索的数据来作为数据的输入源，然后查找其中重复搜索某一个商品的人，并且发送一条告警消息。

代码如下：

public static void main(String[] args) throws Exception{ 
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource source = env.fromElements(
                //浏览记录
                Tuple3.of("Marry", "外套", 1L),
                Tuple3.of("Marry", "帽子",1L),
                Tuple3.of("Marry", "帽子",2L),
                Tuple3.of("Marry", "帽子",3L),
                Tuple3.of("Ming", "衣服",1L),
                Tuple3.of("Marry", "鞋子",1L),
                Tuple3.of("Marry", "鞋子",2L),
                Tuple3.of("LiLei", "帽子",1L),
                Tuple3.of("LiLei", "帽子",2L),
                Tuple3.of("LiLei", "帽子",3L)
        );
        //定义Pattern,寻找连续搜索帽子的用户
        Pattern<Tuple3<String, String, Long>, Tuple3<String, String, Long>> pattern = Pattern
                .<Tuple3<String, String, Long>>begin("start")
                .where(new SimpleCondition<Tuple3<String, String, Long>>() { 
                    @Override
                    public boolean filter(Tuple3<String, String, Long> value) throws Exception { 
                        return value.f1.equals("帽子");
                    }
                }) //.timesOrMore(3);
                .next("middle")
                .where(new SimpleCondition<Tuple3<String, String, Long>>() { 
                    @Override
                    public boolean filter(Tuple3<String, String, Long> value) throws Exception { 
                        return value.f1.equals("帽子");
                    }
                });
    
        KeyedStream keyedStream = source.keyBy(0);
        PatternStream patternStream = CEP.pattern(keyedStream, pattern);
        SingleOutputStreamOperator matchStream = patternStream.select(new PatternSelectFunction<Tuple3<String, String, Long>, String>() { 
            @Override
            public String select(Map<String, List<Tuple3<String, String, Long>>> pattern) throws Exception { 
                List<Tuple3<String, String, Long>> middle = pattern.get("middle");
                return middle.get(0).f0 + ":" + middle.get(0).f2 + ":" + "连续搜索两次帽子!";
            }
        });
        matchStream.printToErr();
        env.execute("execute cep");
    }

上述代码的逻辑我们可以分解如下。

首先定义一个数据源，模拟了一些用户的搜索数据，然后定义了自己的 Pattern。这个模式的特点就是连续两次搜索商品“帽子”，然后进行匹配，发现匹配后输出一条提示信息，直接打印在控制台上。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 58]  
可以看到，提示信息已经打印在了控制台上。

# 第十二讲 Flink常用的Source和Connector #

Flink 作为实时计算领域强大的计算能力，以及与其他系统进行对接的能力都非常强大。Flink 自身实现了多种 Source 和 Connector 方法，并且还提供了多种与第三方系统进行对接的 Connector。

我们可以把这些 Source、Connector 分成以下几个大类。

**预定义和自定义 Source**

在前面的第 04 课时“Flink 常用的 DataSet 和 DataStream API”中提到过几种 Flink 已经实现的新建 DataStream 方法。

**基于文件**

我们在本地环境进行测试时可以方便地从本地文件读取数据：

readTextFile(path)
    readFile(fileInputFormat, path)
    ...

可以直接在 ExecutionEnvironment 和 StreamExecutionEnvironment 类中找到 Flink 支持的读取本地文件的方法，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 59]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 60]

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    // read text file from local files system
    DataSet<String> localLines = env.readTextFile("file:///path/to/my/textfile");
    // read text file from an HDFS running at nnHost:nnPort
    DataSet<String> hdfsLines = env.readTextFile("hdfs://nnHost:nnPort/path/to/my/textfile");
    // read a CSV file with three fields
    DataSet<Tuple3<Integer, String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
    	                       .types(Integer.class, String.class, Double.class);
    // read a CSV file with five fields, taking only two of them
    DataSet<Tuple2<String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
                                   .includeFields("10010")  // take the first and the fourth field
    	                       .types(String.class, Double.class);
    // read a CSV file with three fields into a POJO (Person.class) with corresponding fields
    DataSet<Person>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
                             .pojoType(Person.class, "name", "age", "zipcode");

**基于 Collections**

我们也可以基于内存中的集合、对象等创建自己的 Source。一般用来进行本地调试或者验证。

例如：

fromCollection(Collection)
    fromElements(T ...)

我们也可以在源码中看到 Flink 支持的方法，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 61]

DataSet<String> text = env.fromElements(
          "Flink Spark Storm",
          "Flink Flink Flink",
          "Spark Spark Spark",
          "Storm Storm Storm"
    );
    List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();
    data.add(new Tuple3<>(0,1,0));
    data.add(new Tuple3<>(0,1,1));
    data.add(new Tuple3<>(0,2,2));
    data.add(new Tuple3<>(0,1,3));
    data.add(new Tuple3<>(1,2,5));
    data.add(new Tuple3<>(1,2,9));
    data.add(new Tuple3<>(1,2,11));
    data.add(new Tuple3<>(1,2,13));
    DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);

**基于 Socket**

通过监听 Socket 端口，我们可以在本地很方便地模拟一个实时计算环境。

StreamExecutionEnvironment 中提供了 socketTextStream 方法可以通过 host 和 port 从一个 Socket 中以文本的方式读取数据。

DataStream<String> text = env.socketTextStream("127.0.0.1", 9000, "\n");

**自定义 Source**

我们可以通过实现 Flink 的SourceFunction 或者 ParallelSourceFunction 来实现单个或者多个并行度的 Source。

例如，我们在之前的课程中用到的：

public class MyStreamingSource implements SourceFunction<Item> { 
        private boolean isRunning = true;
        /** * 重写run方法产生一个源源不断的数据发送源 * @param ctx * @throws Exception */
        public void run(SourceContext<Item> ctx) throws Exception { 
            while(isRunning){ 
                Item item = generateItem();
                ctx.collect(item);
                //每秒产生一条数据
                Thread.sleep(1000);
            }
        }
        @Override
        public void cancel() { 
            isRunning = false;
        }
        //随机产生一条商品数据
        private Item generateItem(){ 
            int i = new Random().nextInt(100);
            ArrayList<String> list = new ArrayList();
            list.add("HAT");
            list.add("TIE");
            list.add("SHOE");
            Item item = new Item();
            item.setName(list.get(new Random().nextInt(3)));
            item.setId(i);
            return item;
        }
    }

**自带连接器**

Flink 中支持了比较丰富的用来连接第三方的连接器，可以在官网中找到 Flink 支持的各种各样的连接器：

*  [Apache Kafka][] (source/sink)
 *  [Apache Cassandra][] (sink)
 *  [Amazon Kinesis Streams][] (source/sink)
 *  [Elasticsearch][] (sink)
 *  [Hadoop FileSystem][] (sink)
 *  [RabbitMQ][] (source/sink)
 *  [Apache NiFi][] (source/sink)
 *  [Twitter Streaming API][] (source)
 *  [Google PubSub][] (source/sink)

> 需注意，我们在使用这些连接器时通常需要引用相对应的 Jar 包依赖。而且一定要注意，对于某些连接器比如 Kafka 是有版本要求的，一定要去官方网站找到对应的依赖版本。

**基于 Apache Bahir 发布**

Flink 还会基于 Apache Bahir 来发布一些 Connector，比如我们常用的 Redis 等。

> Apache Bahir 的代码最初是从 Apache Spark 项目中提取的，后作为一个独立的项目提供。Apache Bahir 通过提供多样化的流连接器（Streaming Connectors）和 SQL 数据源扩展分析平台的覆盖面，最初只是为 Apache Spark 提供拓展。目前也为 Apache Flink 提供，后续还可能为 Apache Beam 和更多平台提供拓展服务。

我们可以在 Bahir 的首页中找到目前支持的 Flink 连接器：

*  Flink streaming connector for ActiveMQ
 *  Flink streaming connector for Akka
 *  Flink streaming connector for Flume
 *  Flink streaming connector for InfluxDB
 *  Flink streaming connector for Kudu
 *  Flink streaming connector for Redis
 *  Flink streaming connector for Netty

其中就有我们非常熟悉的 Redis，很多同学 Flink 项目中访问 Redis 的方法都是自己进行的实现，推荐使用 Bahir 连接器。

在本地单机情况下：

public static class RedisExampleMapper implements RedisMapper<Tuple2<String, String>>{ 
        @Override
        public RedisCommandDescription getCommandDescription() { 
            return new RedisCommandDescription(RedisCommand.HSET, "HASH_NAME");
        }
        @Override
        public String getKeyFromData(Tuple2<String, String> data) { 
            return data.f0;
        }
        @Override
        public String getValueFromData(Tuple2<String, String> data) { 
            return data.f1;
        }
    }
    FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build();
    DataStream<String> stream = ...;
    stream.addSink(new RedisSink<Tuple2<String, String>>(conf, new RedisExampleMapper());

当然我们也可以使用在集群或者哨兵模式下使用 Redis 连接器。

集群模式：

FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
        .setNodes(new HashSet<InetSocketAddress>(Arrays.asList(new InetSocketAddress(5601)))).build();
    DataStream<String> stream = ...;
    stream.addSink(new RedisSink<Tuple2<String, String>>(conf, new RedisExampleMapper());

哨兵模式：

FlinkJedisSentinelConfig conf = new FlinkJedisSentinelConfig.Builder()
        .setMasterName("master").setSentinels(...).build();
    DataStream<String> stream = ...;
    stream.addSink(new RedisSink<Tuple2<String, String>>(conf, new RedisExampleMapper());

**基于异步 I/O 和可查询状态**

异步 I/O 和可查询状态都是 Flink 提供的非常底层的与外部系统交互的方式。

其中异步 I/O 是为了解决 Flink 在实时计算中访问外部存储产生的延迟问题，如果我们按照传统的方式使用 MapFunction，那么所有对外部系统的访问都是同步进行的。在很多情况下，计算性能受制于外部系统的响应速度，长时间进行等待，会导致整体吞吐低下。

我们可以通过继承 RichAsyncFunction 来使用异步 I/O：

/** * 实现 'AsyncFunction' 用于发送请求和设置回调 */
    class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> { 
        /** 能够利用回调函数并发发送请求的数据库客户端 */
        private transient DatabaseClient client;
        @Override
        public void open(Configuration parameters) throws Exception { 
            client = new DatabaseClient(host, post, credentials);
        }
        @Override
        public void close() throws Exception { 
            client.close();
        }
        @Override
        public void asyncInvoke(String key, final ResultFuture<Tuple2<String, String>> resultFuture) throws Exception { 
            // 发送异步请求，接收 future 结果
            final Future<String> result = client.query(key);
            // 设置客户端完成请求后要执行的回调函数
            // 回调函数只是简单地把结果发给 future
            CompletableFuture.supplyAsync(new Supplier<String>() { 
                @Override
                public String get() { 
                    try { 
                        return result.get();
                    } catch (InterruptedException | ExecutionException e) { 
                        // 显示地处理异常
                        return null;
                    }
                }
            }).thenAccept( (String dbResult) -> { 
                resultFuture.complete(Collections.singleton(new Tuple2<>(key, dbResult)));
            });
        }
    }
    // 创建初始 DataStream
    DataStream<String> stream = ...;
    // 应用异步 I/O 转换操作
    DataStream<Tuple2<String, String>> resultStream =
        AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);

其中，ResultFuture 的 complete 方法是异步的，不需要等待返回。

我们在之前讲解 Flink State 时，提到过 Flink 提供了 StateDesciptor 方法专门用来访问不同的 state，StateDesciptor 同时还可以通过 setQueryable 使状态变成可以查询状态。可查询状态目前是一个 Beta 功能，暂时不推荐使用。

# 第十三讲 如何实现生产环境中的Flink高可用配置 #

**概述**

事实上，集群的高可用（High Availablility，以下简称 HA）配置是大数据领域经典的一个问题。

> 通常 HA 用来描述一个系统经过专门的设计，从而减少停工时间，而保持其服务的高度可用性。

我们在第 03 课时“Flink 的编程模型与其他框架比较”中也提到过 Flink 集群中的角色，其中 JobManager 扮演的是集群管理者的角色，负责调度任务、协调 Checkpoints、协调故障恢复、收集 Job 的状态信息，并管理 Flink 集群中的从节点 TaskManager。

在默认的情况下，我们的每个集群都只有一个 JobManager 实例，假如这个 JobManager 崩溃了，那么将会导致我们的作业运行失败，并且无法提交新的任务。

因此，在生产环境中我们的集群应该如何配置以达到高可用的目的呢？针对不同模式进行部署的集群，我们需要不同的配置。

**源码分析**

Flink 中的 JobManager、WebServer 等组件都需要高可用保障，并且 Flink 还需要进行 Checkpoint 元数据的持久化操作。与 Flink HA 相关的类图如下图所示，我们跟随源码简单看一下 Flink HA 的实现。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 62]  
HighAvailabilityMode 类中定义了三种高可用性模式枚举，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 63]

*  NONE：非 HA 模式
 *  ZOOKEEPER：基于 ZK 实现 HA
 *  FACTORY\_CLASS：自定义 HA 工厂类，该类需要实现 HighAvailabilityServicesFactory 接口

具体的高可用实例对象创建则在 HighAvailabilityServicesUtils 类中有体现，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 64]  
创建 HighAvailabilityServices 的实例方法如下：

public static HighAvailabilityServices createHighAvailabilityServices(
    		Configuration configuration,
    		Executor executor,
    		AddressResolution addressResolution) throws Exception { 
    		HighAvailabilityMode highAvailabilityMode = LeaderRetrievalUtils.getRecoveryMode(configuration);
    		switch (highAvailabilityMode) { 
    			case NONE:
    				// 省略部分代码
    				// 返回非HA服务类实例
    				return new StandaloneHaServices(
    					resourceManagerRpcUrl,
    					dispatcherRpcUrl,
    					jobManagerRpcUrl,
    					String.format("%s%s:%s", protocol, address, port));
    			case ZOOKEEPER:
    				BlobStoreService blobStoreService = BlobUtils.createBlobStoreFromConfig(configuration);
    				
    				// 返回ZK HA 服务类实例
    				return new ZooKeeperHaServices(
    					ZooKeeperUtils.startCuratorFramework(configuration),
    					executor,
    					configuration,
    					blobStoreService);
    			case FACTORY_CLASS:
    				// 返回自定义 HA 服务类实例
    				return createCustomHAServices(configuration, executor);
    			default:
    				throw new Exception("Recovery mode " + highAvailabilityMode + " is not supported.");
    		}
    	}

HighAvailabilityServices 接口定义了 HA 服务类应当实现的方法，实现类主要有 StandaloneHaServices（非 HA）、ZooKeeperHaServices、YarnHighAvailabilityServices。

ZooKeeperHaServices 主要提供了创建 LeaderRetrievalService 和 LeaderElectionService 等方法，并给出了各个服务组件使用的 ZK 节点名称。

ZooKeeperLeaderElectionService 实现了 LeaderElectionService 中 leader 选举和获取 leader 的方法。

public interface LeaderElectionService { 
    	// 启动 leader 选举服务
    	void start(LeaderContender contender) throws Exception;
    	// 停止 leader 选举服务
    	void stop() throws Exception;
    	// 获取新的 leader session ID
    	void confirmLeaderSessionID(UUID leaderSessionID);
    	// 是否拥有 leader
    	boolean hasLeadership(@Nonnull UUID leaderSessionId);
    }

**Standalone 集群高可用配置**

**简介**

如果你的集群是 Standalone 模式，那么此时需要对 JobManager 做主备，一般推荐一个主 JobManager 和多个备用的 JobManagers。当你的主 JobManager 发生故障时，备用的 JobManager 会接管集群，以保证我们的任务正常运行。这里需要注意的是，主和备 JobManager 只是我们人为的区分，实际上它们并没有区别，每一个 JobManager 都可以当作主或者备。

Standalone 模式下的 HA 配置，Flink 依赖 ZooKeeper 实现。ZooKeeper 集群独立于 Flink 集群之外，主要被用来进行 Leader 选举和轻量级状态一致性存储。[更多关于 ZooKeeper 的资料可以直接点击这里查看][ZooKeeper]。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 65]  
**文件配置**  
在这里我们要特别说明的是，Flink 自带了一个简单的 ZooKeeper 集群，并且提供了一键启动的脚本。在实际生产环境中建议自己搭建 ZooKeeper 集群，以方便我们进行配置管理。

假设我们在 3 台虚拟机之间搭建 standalone 集群，并且进行高可用配置：

<table> 
 <thead> 
  <tr> 
   <th>IP</th> 
   <th>hostname</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>192.168.2.100</td> 
   <td>master</td> 
   <td>主节点、ZK 01</td> 
  </tr> 
  <tr> 
   <td>192.168.2.101</td> 
   <td>slave01</td> 
   <td>从节点 01、ZK 02</td> 
  </tr> 
  <tr> 
   <td>192.168.2.102</td> 
   <td>slave02</td> 
   <td>从节点 02、ZK 03</td> 
  </tr> 
 </tbody> 
</table>

我们需要在 3 台机器上同时修改 Flink 配置文件中的 master 文件：

master:8081
    slave01:8081
    slave02:8081

表示指定 ZooKeeper 集群的访问地址。  
然后，需要修改 conf/flink-conf.yaml 文件，与高可用的配置相关的几个参数，如下所示：

#========================================================
    # High Availability
    #=====================================================================
    high-availability: zookeeper
    high-availability.zookeeper.quorum: localhost:2181
    high-availability.zookeeper.path.root: /flink
    high-availability.cluster-id: /cluster_one
    high-availability.storageDir: hdfs:///flink/recovery

它们分别代表：

*  high-availability，高可用性模式设置为 zookeeper，用来打开高可用模式；
 *  high-availability.zookeeper.quorum，指定一组 ZooKeeper 服务器，它提供分布式协调服务，Flink 可以在指定的地址和端口访问 ZooKeeper；
 *  high-availability.zookeeper.path.root，指定 ZooKeeper 的根节点，并且在该节点下放置所有集群节点；
 *  high-availability.cluster-id，为每个集群指定一个 ID，用来存储该集群的相关数据；
 *  high-availability.storageDir，高可用存储目录，JobManager 的元数据保存在文件系统 storageDir 中，一般来讲是 HDFS 的地址。

对于 flink-conf.yaml 文件中的配置，除了 jobmanager.rpc.address 和 jobmanager.web.address 都各自配置自己机器的 IP 之外，其他的关于高可用的配置一模一样。

这里特别要注意下对于高可用性配置的部分。其中，high-availability、high-availability.storageDir 和 high-availability.zookeeper.quorum 这三项是必须配置的；后两项 high-availability.zookeeper.path.root 和 high-availability.cluster-id 配置是可选的，但是通常我们建议都手动进行配置，方便排查问题。

**Yarn 集群高可用配置**

与 Standalone 集群不同的是，Flink on Yarn 的高可用配置只需要一个 JobManager。当 JobManager 发生失败时，Yarn 负责将其重新启动。

我们需要修改 yarn-site.yaml 文件中的配置，如下所示：

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

yarn.resourcemanager.am.max-attempts 表示 Yarn 的 application master 的最大重试次数。

除了上述 HA 配置之外，还需要配置 flink-conf.yaml 中的最大重试次数：

yarn.application-attempts: 10

默认情况下，该配置的值为 2：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 66]  
我们在 Flink 的官网中可以查到，当你的 yarn.application-attempts 配置为 10 的时候：

> 这意味着如果程序启动失败，YARN 会再重试 9 次（9 次重试 + 1 次启动），如果 YARN 启动 10 次作业还失败，则 YARN 才会将该任务的状态置为失败。如果发生进程抢占，节点硬件故障或重启，NodeManager 重新同步等，YARN 会继续尝试启动应用。 这些重启不计入 yarn.application-attempts 个数中。

同时官网给出了重要提示：

*  **YARN 2.3.0 < version < 2.4.0**：All containers are restarted if the application master fails.
 *  **YARN 2.4.0 < version < 2.6.0**：TaskManager containers are kept alive across application master failures. This has the advantage that the startup time is faster and that the user does not have to wait for obtaining the container resources again.
 *  **YARN 2.6.0 <= version**：Sets the attempt failure validity interval to the Flinks’ Akka timeout value. The attempt failure validity interval says that an application is only killed after the system has seen the maximum number of application attempts during one interval. This avoids that a long lasting job will deplete it’s application attempts.

不同 Yarn 版本的容器关闭行为不同，需要我们特别注意。

*  **YARN 2.3.0 < YARN 版本 < 2.4.0**：如果 application master 进程失败，则所有的 container 都会重启。
 *  **YARN 2.4.0 < YARN 版本 < 2.6.0**：TaskManager container 在 application master 故障期间，会继续工作。这样的优点是：启动时间更快，且缩短了所有 task manager 启动时申请资源的时间。
 *  **YARN 2.6.0 <= YARN 版本**：失败重试的间隔会被设置为 Akka 的超时时间。在一次时间间隔内达到最大失败重试次数才会被置为失败。

另外，需要注意的是，假如你的 ZooKeeper 集群使用 Kerberos 安全模式运行，那么可以根据需要添加下面的配置：

zookeeper.sasl.service-name
    zookeeper.sasl.login-context-name

如果你不想搭建自己的 ZooKeeper 集群或者简单地进行本地测试，你可以使用 Flink 自带的 ZooKeeper 集群，但是并不推荐，我们建议读者搭建自己的 ZooKeeper 集群。

# 第十四讲 Flink Exactly-once实现原理解析 #

Flink 的“精确一次”处理语义是，Flink 提供了一个强大的语义保证，也就是说在任何情况下都能保证数据对应用产生的效果只有一次，不会多也不会少。

那么 Flink 是如何实现“端到端的精确一次处理”语义的呢？

**背景**

通常情况下，流式计算系统都会为用户提供指定数据处理的可靠模式功能，用来表明在实际生产运行中会对数据处理做哪些保障。一般来说，流处理引擎通常为用户的应用程序提供三种数据处理语义：最多一次、至少一次和精确一次。

*  **最多一次（At-most-Once）**：这种语义理解起来很简单，用户的数据只会被处理一次，不管成功还是失败，不会重试也不会重发。
 *  **至少一次（At-least-Once）**：这种语义下，系统会保证数据或事件至少被处理一次。如果中间发生错误或者丢失，那么会从源头重新发送一条然后进入处理系统，所以同一个事件或者消息会被处理多次。
 *  **精确一次（Exactly-Once）**：表示每一条数据只会被精确地处理一次，不多也不少。

Exactly-Once 是 Flink、Spark 等流处理系统的核心特性之一，这种语义会保证每一条消息只被流处理系统处理一次。“精确一次” 语义是 Flink 1.4.0 版本引入的一个重要特性，而且，Flink 号称支持“端到端的精确一次”语义。

在这里我们解释一下“端到端（End to End）的精确一次”，它指的是 Flink 应用从 Source 端开始到 Sink 端结束，数据必须经过的起始点和结束点。Flink 自身是无法保证外部系统“精确一次”语义的，所以 Flink 若要实现所谓“端到端（End to End）的精确一次”的要求，那么外部系统必须支持“精确一次”语义；然后借助 Flink 提供的分布式快照和两阶段提交才能实现。

**分布式快照机制**

我们在之前的课程中讲解过 Flink 的容错机制，Flink 提供了失败恢复的容错机制，而这个容错机制的核心就是持续创建分布式数据流的快照来实现。

同 Spark 相比，Spark 仅仅是针对 Driver 的故障恢复 Checkpoint。而 Flink 的快照可以到**算子级别**，并且对全局数据也可以做快照。Flink 的分布式快照受到 **Chandy-Lamport 分布式快照算法**启发，同时进行了量身定做，有兴趣的同学可以搜一下。

**Barrier**

Flink 分布式快照的核心元素之一是 Barrier（数据栅栏），我们也可以把 Barrier 简单地理解成一个标记，该标记是严格有序的，并且随着数据流往下流动。每个 Barrier 都带有自己的 ID，Barrier 极其轻量，并不会干扰正常的数据处理。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 67]  
如上图所示，假如我们有一个从左向右流动的数据流，Flink 会依次生成 snapshot 1、 snapshot 2、snapshot 3……Flink 中有一个专门的“协调者”负责收集每个 snapshot 的位置信息，这个“协调者”也是高可用的。

Barrier 会随着正常数据继续往下流动，每当遇到一个算子，算子会插入一个标识，这个标识的插入时间是上游所有的输入流都接收到 snapshot n。与此同时，当我们的 sink 算子接收到所有上游流发送的 Barrier 时，那么就表明这一批数据处理完毕，Flink 会向“协调者”发送确认消息，表明当前的 snapshot n 完成了。当所有的 sink 算子都确认这批数据成功处理后，那么本次的 snapshot 被标识为完成。

这里就会有一个问题，因为 Flink 运行在分布式环境中，一个 operator 的上游会有很多流，每个流的 barrier n 到达的时间不一致怎么办？这里 Flink 采取的措施是：**快流等慢流**。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 68]

> 拿上图的 barrier n 来说，其中一个流到的早，其他的流到的比较晚。当第一个 barrier n到来后，当前的 operator 会继续等待其他流的 barrier n。直到所有的barrier n 到来后，operator 才会把所有的数据向下发送。

**异步和增量**

按照上面我们介绍的机制，每次在把快照存储到我们的状态后端时，如果是同步进行就会阻塞正常任务，从而引入延迟。因此 Flink 在做快照存储时，可采用异步方式。

此外，由于 checkpoint 是一个全局状态，用户保存的状态可能非常大，多数达 G 或者 T 级别。在这种情况下，checkpoint 的创建会非常慢，而且执行时占用的资源也比较多，因此 Flink 提出了增量快照的概念。也就是说，每次都是进行的全量 checkpoint，是基于上次进行更新的。

**两阶段提交**

上面我们讲解了基于 checkpoint 的快照操作，快照机制能够保证作业出现 fail-over 后可以从最新的快照进行恢复，即分布式快照机制可以保证 Flink 系统内部的“精确一次”处理。但是我们在实际生产系统中，Flink 会对接各种各样的外部系统，比如 Kafka、HDFS 等，一旦 Flink 作业出现失败，作业会重新消费旧数据，这时候就会出现重新消费的情况，也就是重复消费。

针对这种情况，Flink 1.4 版本引入了一个很重要的功能：两阶段提交，也就是 TwoPhaseCommitSinkFunction。两阶段搭配特定的 source 和 sink（特别是 0.11 版本 Kafka）使得“精确一次处理语义”成为可能。

在 Flink 中两阶段提交的实现方法被封装到了 TwoPhaseCommitSinkFunction 这个抽象类中，我们只需要实现其中的beginTransaction、preCommit、commit、abort 四个方法就可以实现“精确一次”的处理语义，实现的方式我们可以在官网中查到：

*  beginTransaction，在开启事务之前，我们在目标文件系统的临时目录中创建一个临时文件，后面在处理数据时将数据写入此文件；
 *  preCommit，在预提交阶段，刷写（flush）文件，然后关闭文件，之后就不能写入到文件了，我们还将为属于下一个检查点的任何后续写入启动新事务；
 *  commit，在提交阶段，我们将预提交的文件原子性移动到真正的目标目录中，请注意，这会增加输出数据可见性的延迟；
 *  abort，在中止阶段，我们删除临时文件。

**Flink-Kafka Exactly-once**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 69]  
如上图所示，我们用 Kafka-Flink-Kafka 这个案例来介绍一下实现“端到端精确一次”语义的过程，整个过程包括：

*  从 Kafka 读取数据
 *  窗口聚合操作
 *  将数据写回 Kafka

整个过程可以总结为下面四个阶段：

*  一旦 Flink 开始做 checkpoint 操作，那么就会进入 pre-commit 阶段，同时 Flink JobManager 会将检查点 Barrier 注入数据流中 ；
 *  当所有的 barrier 在算子中成功进行一遍传递，并完成快照后，则 pre-commit 阶段完成；
 *  等所有的算子完成“预提交”，就会发起一个“提交”动作，但是任何一个“预提交”失败都会导致 Flink 回滚到最近的 checkpoint；
 *  pre-commit 完成，必须要确保 commit 也要成功，上图中的 Sink Operators 和 Kafka Sink 会共同来保证。

**现状**

目前 Flink 支持的精确一次 Source 列表如下表所示，你可以使用对应的 connector 来实现对应的语义要求：

<table> 
 <thead> 
  <tr> 
   <th>数据源</th> 
   <th>语义保证</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Apache Kafka</td> 
   <td>exactly once</td> 
   <td>需要对应的 Kafka 版本</td> 
  </tr> 
  <tr> 
   <td>AWS Kinesis Streams</td> 
   <td>exactly once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>RabbitMQ</td> 
   <td>at most once (v 0.10) / exactly once (v 1.0)</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Twitter Streaming API</td> 
   <td>at most once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Collections</td> 
   <td>exactly once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Files</td> 
   <td>exactly once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Sockets</td> 
   <td>at most once</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

如果你需要实现真正的“端到端精确一次语义”，则需要 sink 的配合。目前 Flink 支持的列表如下表所示：

<table> 
 <thead> 
  <tr> 
   <th>数据源</th> 
   <th>语义保证</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>HDFS rolling sink</td> 
   <td>exactly once</td> 
   <td>依赖 Hadoop 版本</td> 
  </tr> 
  <tr> 
   <td>Elasticsearch</td> 
   <td>at least once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Kafka producer</td> 
   <td>at least once / exactly once</td> 
   <td>需要 Kafka 0.11 及以上</td> 
  </tr> 
  <tr> 
   <td>Cassandra sink</td> 
   <td>at least once / exactly once</td> 
   <td>幂等更新</td> 
  </tr> 
  <tr> 
   <td>AWS Kinesis Streams</td> 
   <td>at least once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>File sinks</td> 
   <td>at least once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Socket sinks</td> 
   <td>at least once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Standard output</td> 
   <td>at least once</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>Redis sink</td> 
   <td>at least once</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

**总结**

由于强大的异步快照机制和两阶段提交，Flink 实现了“端到端的精确一次语义”，在特定的业务场景下十分重要，我们在进行业务开发需要语义保证时，要十分熟悉目前 Flink 支持的语义特性。

这一课时的内容较为晦涩，建议你从源码中去看一下具体的实现。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/0bf00ea84966483894baaedaff1734dd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/77724c6d18d3412c80e70faeca59d181.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/a0586ba0d0f348e3aa2b755adea66a7a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/1b2ca912fb1349df872bc222721c2453.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/eb08713f1ad64e27a7be55096900d9fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/838da948352f4e9c8803a0cf831cc72b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/4d3d0baf22844a248dbc117963ed23c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/9177f9378f2b4523b79b4f79c95ce7c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/d760a4fff7854be4a23d34d44cc9db1e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/7e5cbbaa28a34a13a5abde2804d3a711.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/921c70b4dac947e590355ba31926da65.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221123/b025f9f2ab364130b7276ca567cf374f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221123/f782d8bd225f47ecbeeff41aec368c97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 13]: /images/20221123/0b4f512731974365ade9d30fecc81274.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 14]: /images/20221123/67d52ed4e7464529be577091ab420376.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 15]: /images/20221123/5286f511c1ce42f6acd4cee4ffa2907b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 16]: /images/20221123/a5127edbe99d4272a4d65ec9def55966.png
[Link 1]: https://flink.apache.org/downloads.html
[Link 2]: https://flink.apache.org/downloads.html#apache-flink-1100
[Apache Flink 1.10.0 for Scala 2.11]: https://www.apache.org/dyn/closer.lua/flink/flink-1.10.0/flink-1.10.0-bin-scala_2.11.tgz
[20200903101004668.png_pic_center]: /images/20221123/9a9ba910d9e2420d8424d2a81e51c6de.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 17]: /images/20221123/1d158ded99fa4785959ccd883942c30d.png
[20200903101605383.png_pic_center]: /images/20221123/30a595616f44442d86661560d4a1e52c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 18]: /images/20221123/e4ab18e3f0664eb190567aefc58b1d66.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 19]: /images/20221123/a1c9f3741d5e4fefa201d0050c56ac27.png
[Link 3]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/ops/config.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 20]: /images/20221123/308342cb23a645c7bf08a1ef394ecdfb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 21]: /images/20221123/1dda4a2d3a094efda09c5feaa3428c5d.png
[Link 4]: https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 22]: /images/20221123/d9383914f90b420db61658dd8e746a94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 23]: /images/20221123/81d1ac306a914d479b2b97719abb3bc9.png
[Link 5]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/cli.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 24]: /images/20221123/98716f63dfca43da924a92d4ee0d076d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 25]: /images/20221123/06e4ba1b9259422b959bceccd39ed930.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 26]: /images/20221123/342e7571cd2b48989cf59b793642a936.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 27]: /images/20221123/c6161f98d8444051b944de9c915c028d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 28]: /images/20221123/361c31d69e7b481cadd4962671528946.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 29]: /images/20221123/117522e6f3e54f698ba2aa95b04b984a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 30]: /images/20221123/bbe982a21c9b4c7c8ac154e214f5d867.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 31]: /images/20221123/11a085349c94483f95034069e263fea2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 32]: /images/20221123/b241b696af1f4e6cbf6ef961ae50455e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 33]: /images/20221123/e913eefc74c843e58778546cdb82cb5b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 34]: /images/20221123/25cf6808f78143b8bbcffe67f390fd4f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 35]: /images/20221123/f3cf3d2c621048cab8fc952ca5352da5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 36]: /images/20221123/4b1fb7be6fdb4691b1d42984847116f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 37]: /images/20221123/8a4aeedab0944c769bb38abf19f3e1ac.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 38]: /images/20221123/9fa0c28d99754d878b1f94fcb4c1acb5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 39]: /images/20221123/981593dccc534876bcd33192aeaf4b29.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 40]: /images/20221123/3033afd95d5648588a746009ebb88578.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 41]: /images/20221123/f3168dd7aa4048cc9b821679ae433041.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 42]: /images/20221123/2038d221bd09449a9a004af9a6c1b76e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 43]: /images/20221123/28e9f6b96d214a86b059cecf61570926.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 44]: /images/20221123/88370c1f6fbd49caa10d34031b4e7c7a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 45]: /images/20221123/e8313ccb7d86456fb4e7dca2ba80a3f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 46]: /images/20221123/22de8d7aa895449590f116aacb3e6846.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 47]: /images/20221123/db3901bae669494fbc2ba0fd94938e77.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 48]: /images/20221123/817e08c3efa14ee8803c8f5678eba7b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 49]: /images/20221123/f02713d6cdc6414bafb4514a867da1aa.png
[Link 6]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/libs/cep.html#individual-patterns
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 50]: /images/20221123/29b9b3fbb30e426bbdde2d1b72ed70ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 51]: /images/20221123/6e238d712c734ab5a9449f1e5a9dae0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 52]: /images/20221123/3f2ff54f9ca248129ddf42d2047c225e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 53]: /images/20221123/e296d3d80fb74a1f918cedae01ec9acb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 54]: /images/20221123/a2a8c3aef1874ab28d447983893d825a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 55]: /images/20221123/dadf6aa226fe49ffbe22f2fc87ed470a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 56]: /images/20221123/4c76a472a9f34b4c918f434f76d02f2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 57]: /images/20221123/beee54599eaa432cafa4ca79b0a31345.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 58]: /images/20221123/c53489f2f2ef4d2fa7473f8b645aed41.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 59]: /images/20221123/c6a9815790f944528b4a3170446ff4c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 60]: /images/20221123/7e71ddc02f5e465d87f65e678579e222.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 61]: /images/20221123/93f072472b4f47bc9624b02a389ef25a.png
[Apache Kafka]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/kafka.html
[Apache Cassandra]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/cassandra.html
[Amazon Kinesis Streams]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/kinesis.html
[Elasticsearch]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/elasticsearch.html
[Hadoop FileSystem]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/filesystem_sink.html
[RabbitMQ]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/rabbitmq.html
[Apache NiFi]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/nifi.html
[Twitter Streaming API]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/twitter.html
[Google PubSub]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/connectors/pubsub.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 62]: /images/20221123/82d7f00bcbf54d1bb82f8178ae78161a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 63]: /images/20221123/38084004dd3843aeac7ab73795b3fc59.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 64]: /images/20221123/c76fd9db55404f4888bd2e0919d35845.png
[ZooKeeper]: https://zookeeper.apache.org/doc/current/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 65]: /images/20221123/aa400c2642314e87a6994de769446316.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 66]: /images/20221123/c2b3f12e64a945169671c02f78665e95.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 67]: /images/20221123/28a1c09822514a45bc03e34ab3a581d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 68]: /images/20221123/b5262abf847841b8b451069b0362c4d5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 69]: /images/20221123/ffaf8c03dffb41a492f12698ffbfd0e8.png