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

谁借莪１个温暖的怀抱￠ 2022-12-11 07:26 125阅读 0赞

### 文章目录 ###

*  第二十九讲 项目背景和实时处理系统架构设计
 *  第三十讲 Flume 和 Kafka 整合和部署
 *  第三十一讲 Kafka 模拟数据生成和发送
 *  第三十二讲 Flink 和 Kafka 整合时间窗口设计
 *  第三十三讲 Flink 计算 PV、UV 代码实现
 *  第三十四讲 Flink 和 Redis 整合以及 Redis Sink 实现
 *  第三十五讲 项目背景和 Flink CEP 简介
 *  第三十六讲 自定义消息事件
 *  第三十七讲 自定义 Pattern 和报警规则
 *  第三十八讲 Flink 调用 CEP 实现报警功能
 *  第三十九讲 Flink 面试-基础篇
 *  第四十讲 Flink 面试-进阶篇
 *  第四十一讲 Flink 面试-源码篇
 *  第四十二讲 Flink 面试-方案设计篇

# 第二十九讲 项目背景和实时处理系统架构设计 #

从这一课时开始我们进入“Flink 实时统计 PV、UV”项目的学习。本课时先介绍实时统计项目的背景、架构设计和技术选型。

**背景**

PV（Page View，网站的浏览量）即页面的浏览次数，一般用来衡量网站用户访问的网页数量。我们可以简单地认为，一个用户每次打开一个页面便会记录一次 PV，也就是说，PV 是指页面刷新的次数，刷新一次页面，记录一次 PV。

在互联网项目中，PV 的度量方法是指发起一次请求（Request）从浏览器到网络服务器，网络服务器在接收到请求后会将对应的网页返回给访问者，这个过程就是一次 PV。

UV（Unique Visitor，独立访客次数）是一天内访问某个站点的人数。一天内同一个用户不管有多少次访问网站，那么只记录一次 UV。一般来讲，会通过 IP 或者 Cookie 来进行判断。

如果网站的一个页面有 100 次访问，其中有一些访问者多次访问和点击，那么页面的 UV 一定是小于 100 的。

PV 和 UV 是我们业务中十分常见的场景，对于这种需求一般是通过什么样的技术架构实现呢？

**技术选型和整体架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center]  
在互联网常见的后端应用中，技术选型可能多种多样，比如基于 Spring、Dubbo、Spring Cloud 等，但是基本的处理框架大同小异。最常见的数据就是日志数据，如果是 Web 应用，那么我们的日志数据就是用户的访问日志或者用户的点击日志等。

从上图中可以看到有几个关键的处理步骤：

*  日志数据如何上报
 *  日志数据的清洗
 *  实时计算程序
 *  前端展示

所以基于以上的业务处理流程，我们常用的实时处理技术选型和架构如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 1]  
在上图的架构图中，涉及几个关键的技术选型和处理步骤，我们下面一一进行讲解。

**日志数据埋点**

“埋点”是数据采集领域中的一个术语，特别是采集用户行为数据，针对用户的某些事件和行为进行捕捉、处理并发送的过程。

埋点数据是十分重要的数据，也可用来进行：

*  流量监控、页面访问的热点分析
 *  构建用户的行为数据，获取用户整个访问的路径和习惯
 *  可以根据用户的访问数据进行偏好分析、个性化推荐
 *  AB 测试
 *  ……

数据埋点的方式多种多样，可以在页面中使用 JavaScript 埋点开发，也可以简单地通过 Nginx 代理服务器进行日志的规范化。

当前大型网站的架构一般都是通过 Nginx 等服务器做网管和代理，Nginx 服务器本身支持日志格式的定制化。

服务端也可以先定义好全局的规范化日志，然后再进行采集，可以更加精细化的定制，满足精细化分析的需求。

我们在本案例中的场景是计算 PV 和 UV 数据，所以数据来源最有可能的是 Nginx 定制化的日志格式，例如：

log_format access $remote_addr - $remote_user [$time_local] "$request"' '$status $body_bytes_sent "$http_referer"' '"$http_user_agent" $http_x_forwarded_for;

Nginx 服务器的配置中跟日志相关的命令主要有 log\_format 和 access\_log。其中 log\_format 可以用来定义日志格式，access\_log 则指定了日志的存储路径。

**日志数据采集**

日志数据的采集需要根据日志的生成方式进行。在这里介绍两款普遍使用的日志收集中间件：

*  Flume
 *  Logstash/FileBeat

根据官网的介绍：

> Flume 是一个分布式、可靠和高可用的海量日志采集、聚合和传输的系统。Flume 可以采集文件、Socket 数据包等各种形式源数据，又可以将采集到的数据输出到 HDFS、HBase、Hive、Kafka 等众多外部存储系统中。一般的采集需求，通过对 Flume 的简单配置即可实现 Flume 针对特殊场景也具备良好的自定义扩展能力。因此，Flume 可以适用于大部分的日常数据采集场景。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 2]  
Flume 分布式系统中核心角色是 Agent，整个采集系统由一个个 Agent 进行的串联。我们可以把每个 Agent 想象为一个数据的投递员，它由三个组件构成：

*  Source，用来进行数据采集
 *  Channel，数据传输的通道
 *  Sink，用于把采集后的数据进行处理

我们同样也可以在官网找到 Logstash 和 FileBeat 的介绍：

> Logstash 能够动态地采集、转换和传输数据，不受格式或复杂度的影响。利用 Grok 从非结构化数据中派生出结构，从 IP 地址解码出地理坐标，匿名化或排除敏感字段，并简化整体的处理过程。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 3]  
Logstash 支持各种输入选择，比如 log4j、Redis 等，可以同时从众多常用来源捕捉事件。能够以连续的流式传输方式，轻松地从你的日志、指标、Web 应用、数据存储以及各种 AWS 服务中采集数据。

同样 FileBeat 也经常用来进行本地文件的日志数据采集，同时可以监控日志目录或者指定的日志文件，并且进行 tail 等操作，也可以将结果转发给 Elasticsearch 或 Logstatsh 等。

如果你的日志分析架构是 ES 系列技术栈，完全可以使用它们来代替 Flume。

**实时计算**

我们在上个项目的架构设计中指出了 Flink 的优势，在这里不再过多展开：

*  状态管理
 *  丰富的 API
 *  生态完善
 *  批流一体

但需要注意的是，我们在进行 PV、UV 的统计时，Flink 会被用到两次：

*  日志清洗
 *  数据统计

我们采集到的日志数据根据需要先进行清洗，包括丢弃掉不符合要求的数据，对日志数据进行分类，这个过程也会基于 Flink 进行，然后 Sink 到 Kafka 中给下游进行消费。

**实时统计主要内容**

在本次的案例中，我们不但对 PV 和 UV 进行计算，同时还会讲解 Flume 和 Kafka 的整合，Flink 日志清洗并 Sink 到 Kafka 等。

整个课程的设计包含以下几个部分：

*  Flume 和 Kafka 整合和部署
 *  Kafka 模拟数据生成和发送
 *  Flink 和 Kafka 整合时间窗口设计
 *  Flink 计算 PV、UV 代码实现
 *  Flink 和 Redis 整合以及 Redis Sink 实现

# 第三十讲 Flume 和 Kafka 整合和部署 #

**Flume 概述**

Flume 是 Hadoop 生态圈子中的一个重要组件，在上一课时中提过，它是一个分布式的、高可靠的、高可用的日志采集工具。

Flume 具有基于流式数据的简单灵活的架构，同时兼具高可靠性、高可用机制和故障转移机制。当我们使用 Flume 收集数据的速度超过下游的写入速度时，Flume 会自动做调整，使得数据的采集和推送能够平稳进行。

Flume 支持多路径采集、多管道数据接入和多管道数据输出。数据源可以是 HBase、HDFS 和文本文件，也可以是 Kafka 或者其他的 Flume 客户端。

**Flume 的组件介绍**

Flume 中有很多组件和概念，下面我把 Flume 中的核心组件一一进行介绍：  
![在这里插入图片描述][20200929143252973.png_pic_center]

*  Client：客户端，用来运行 Flume Agent。
 *  Event：Flume 中的数据单位，可以是一行日志、一条消息。
 *  Agent：代表一个独立的 Flume 进程，包含三个组件：Source、Channel 和 Sink。
 *  Source：数据的收集入口，用来获取 Event 并且传递给 Channel。
 *  Channel：Event 的一个临时存储，是数据的临时通道，可以认为是一个队列。
 *  Sink：从 Channel 中读取 Event，将 Event 中的数据传递给下游。
 *  Flow：一个抽象概念，可以认为是一个从 Source 到达 Sink 的数据流向图。

**Flume 本地环境搭建**

我们在 Flume 的 [官网][Link 1]下载安装包，在这里下载一个 1.8.0 的稳定版本，然后进行解压：

tar zxf apache-flume-1.8.0-bin.tar.gz

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 4]  
可以看到有几个关键的目录，其中 conf/ 目录则是我们存放配置文件的目录。

**Flume 测试**

我们在下载 Flume 后，需要进行测试，可以通过监听本地的端口输入，并且在控制台进行打印。

首先，需要修改 conf/ 目录下的 flume-env.sh，在里面配置 JAVA\_HOME 等配置：

cd /usr/local/apache-flume-1.8.0-bin/conf 
    cp　flume-env.sh.template　flume-env.sh

然后在 flume-env.sh 里面设置 JAVA\_HOME 和 FLUME\_CLASSPATH 变量：

export JAVA_HOME=/usr/local/java/jdk1.8 
    FLUME_CLASSPATH="/usr/local/apache-flume-1.8.0-bin/"

创建一个配置文件 nc\_logger.conf :

vim nc_logger.conf

更改配置如下：

# 定义这个 agent 中各组件的名字 
    a1.sources = r1 
    a1.sinks = k1 
    a1.channels = c1 
    # 描述和配置 source 组件：r1 
    a1.sources.r1.type = netcat 
    a1.sources.r1.bind = localhost 
    a1.sources.r1.port = 9000 
    # 描述和配置 sink 组件：k1 
    a1.sinks.k1.type = logger 
    # 描述和配置channel组件，此处使用是内存缓存的方式 
    a1.channels.c1.type = memory 
    a1.channels.c1.capacity = 1000 
    a1.channels.c1.transactionCapacity = 100 
    # 描述和配置 source channel sink 之间的连接关系 
    a1.sources.r1.channels = c1 
    a1.sinks.k1.channel = c1

我们使用如下命令启动一个 Flume Agent:

bin/flume-ng agent  
    -c conf  
    -f conf/nc_logger.conf  
    -n a1 -Dflume.root.logger=INFO,console

其中有几个关键的命令：

*  –conf (-c) 用来指定配置文件夹路径；
 *  –conf-file(-f) 用来指定采集方案文件；
 *  –name(-n) 用来指定 agent 名字；
 *  \-Dflume.root.logger=INFO,console 开启 flume 日志输出到终端。

用 nc 命令打开本地 9000 端口：

nc localhost 9000

![在这里插入图片描述][20200929144545885.png_pic_center]  
向端口输入几个单词，可以在另一端的控制台看到输出结果，如下图所示：  
![在这里插入图片描述][20200929144604436.png_pic_center]  
我们可以看到 9000 端口中输入的数据已经被打印出来了。

**Flume + Kafka 整合**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 5]  
整体整合思路为，我们的两个 Flume Agent 分别部署在两台 Web 服务器上，用来采集两台服务器的业务日志，并且 Sink 到另一台 Flume Agent 上，然后将数据 Sink 到 Kafka 集群。在这里需要配置三个 Flume Agent。

首先在 Flume Agent 1 和 Flume Agent 2 上创建配置文件，具体如下。

修改 source、channel 和 sink 的配置，vim log\_kafka.conf 代码如下：

# 定义这个 agent 中各组件的名字 
    a1.sources = r1 
    a1.sinks = k1 
    a1.channels = c1 
    # source的配置，监听日志文件中的新增数据 
    a1.sources.r1.type = exec 
    a1.sources.r1.command  = tail -F /home/logs/access.log 
    #sink配置，使用avro日志做数据的消费 
    a1.sinks.k1.type = avro 
    a1.sinks.k1.hostname = flumeagent03 
    a1.sinks.k1.port = 9000 
    #channel配置，使用文件做数据的临时缓存 
    a1.channels.c1.type = file 
    a1.channels.c1.checkpointDir = /home/temp/flume/checkpoint 
    a1.channels.c1.dataDirs = /home/temp/flume/data 
    #描述和配置 source channel sink 之间的连接关系 
    a1.sources.r1.channels = c1 
    a1.sinks.k1.channel = c

上述配置会监听 /home/logs/access.log 文件中的数据变化，并且将数据 Sink 到 flumeagent03 的 9000 端口。

然后我们分别启动 Flume Agent 1 和 Flume Agent 2，命令如下：

$ flume-ng agent  
    -c conf  
    -n a1  
    -f conf/log_kafka.conf >/dev/null 2>&1 &

第三个 Flume Agent 用来接收上述两个 Agent 的数据，并且发送到 Kafka。我们需要启动本地 Kafka，并且创建一个名为 log\_kafka 的 Topic。

然后，我们创建 Flume 配置文件，具体如下。

修改 source、channel 和 sink 的配置，vim flume\_kafka.conf 代码如下：

# 定义这个 agent 中各组件的名字 
    a1.sources = r1 
    a1.sinks = k1 
    a1.channels = c1 
    #source配置 
    a1.sources.r1.type = avro 
    a1.sources.r1.bind = 0.0.0.0 
    a1.sources.r1.port = 9000 
    #sink配置 
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink 
    a1.sinks.k1.topic = log_kafka 
    a1.sinks.k1.brokerList = 127.0.0.1:9092 
    a1.sinks.k1.requiredAcks = 1 
    a1.sinks.k1.batchSize = 20 
    #channel配置 
    a1.channels.c1.type = memory 
    a1.channels.c1.capacity = 1000 
    a1.channels.c1.transactionCapacity = 100 
    #描述和配置 source channel sink 之间的连接关系 
    a1.sources.r1.channels = c1 
    a1.sinks.k1.channel = c1

配置完成后，我们启动该 Flume Agent：

$ flume-ng agent  
    -c conf  
    -n a1  
    -f conf/flume_kafka.conf >/dev/null 2>&1 &

我们在第 24 课时“Flink 消费 Kafka 数据开发” 中详细讲解了 Flink 消费 Kafka 数据的开发。当 Flume Agent 1 和 2 中监听到新的日志数据后，数据就会被 Sink 到 Kafka 指定的 Topic，我们就可以消费 Kafka 中的数据了。

# 第三十一讲 Kafka 模拟数据生成和发送 #

第 29 课时讲过，在计算 PV 和 UV 的过程中关键的一个步骤就是进行日志数据的清洗。实际上在其他业务，比如订单数据的统计中，我们也需要过滤掉一些“脏数据”。

所谓“脏数据”是指与我们定义的标准数据结构不一致，或者不需要的数据。因为在数据清洗 ETL 的过程中经常需要进行数据的反序列化解析和 Java 类的映射，在这个映射过程中“脏数据”会导致反序列化失败，从而使得任务失败进行重启。在一些大作业中，重启会导致任务不稳定，而过多的“脏数据”会导致我们的任务频繁报错，最终彻底失败。

**架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 6]  
我们在第 30 课时中提过整个 PV 和 UV 计算过程中的数据处理架构，其中使用 Flume 收集业务数据并且发送到 Kafka 中，那么在计算 PV、UV 前就需要消费 Kafka 中的数据，并且将“脏数据”过滤掉。

在实际业务中，我们消费原始 Kafka 日志数据进行处理后，会同时把明细数据写到类似 Elasticsearch 这样的引擎中查询；也会把汇总数据写入 HBase 或者 Redis 等数据库中提供给前端查询展示用。同时，还会把数据再次写入 Kafka 中提供给其他业务使用。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 7]  
**日志清洗**

我们的原始数据被处理成 JSON 的形式发送到 Kafka 中，将原始数据反序列化成对应的访问日志对象进行计算，过滤掉一些“不合法”数据，比如访问者 ID 为空等。在这个过程中我们的 Source 是 Kafka Consumer，并且使用处理时间作为时间特征。

首先新建环境，设置检查点等参数：

StreamExecutionEnviro nment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); 
    env.enableCheckpointing(5000);

接下来接入 Kafka Source，我们在第 30 课时中用 Flume 将收集到的原始日志写到了名为 log\_kafka 的 Topic 中，Flume 的配置如下：

#sink配置 
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink 
    a1.sinks.k1.topic = log_kafka 
    a1.sinks.k1.brokerList = 127.0.0.1:9092 
    a1.sinks.k1.requiredAcks = 1 
    a1.sinks.k1.batchSize = 20

我们构造的 Kafka Consumer 属性如下：

Properties properties = new Properties(); 
    properties.setProperty("bootstrap.servers", "127.0.0.1:9092"); 
    //设置消费组 
    FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("log_kafka", new SimpleStringSchema(), properties);

使用 Filter 算子将不合法的数据过滤。过滤的逻辑是：如果消息体中 userId 为空或者事件消息不是“点击”的事件。

定义的消息格式如下所示：

public class UserClick {  
        private String userId; 
        private Long timestamp; 
        private String action; 
        public String getUserId() {  
            return userId; 
        } 
        public void setUserId(String userId) {  
            this.userId = userId; 
        } 
        public Long getTimestamp() {  
            return timestamp; 
        } 
        public void setTimestamp(Long timestamp) {  
            this.timestamp = timestamp; 
        } 
        public String getAction() {  
            return action; 
        } 
        public void setAction(String action) {  
            this.action = action; 
        } 
    } 
    enum UserAction{  
        //点击 
        CLICK("CLICK"), 
        //购买 
        PURCHASE("PURCHASE"), 
        //其他 
        OTHER("OTHER"); 
        private String action; 
        UserAction(String action) {  
            this.action = action; 
        } 
    }

我们简化后的消息体如上，其中 userId 为用户的 ID，timestamp 是时间戳，UserAction 是用户动作枚举：点击、购买和其他。我们需要其中的“点击”事件。

首先需要过滤掉字符串中明显不合法的数据：

public class UserActionFilter implements FilterFunction<String> {  
        @Override 
        public boolean filter(String input) throws Exception {  
            return input.contains("CLICK") && input.startsWith("{") && input.endsWith("}"); 
        } 
    }

在实际生产中，我们的日志数据并不像我们想象的那样规范，其中有可能夹杂大量的“非法数据”，这些数据需要根据业务需求进行过滤。

我们在这里只获取那些用户事件为“点击”的消息，并且需要满足 JSON 格式数据的基本要求：以“\{" 开头，以 "\}”结尾。

然后在 FlatMap 中进一步将日志数据进行处理，过滤掉 userId 为空或者 action 类型为空的数据：

public class MyFlatMapFunction implements FlatMapFunction<String,String> {  
        @Override 
        public void flatMap(String input, Collector out) throws Exception {  
            JSONObject jsonObject = JSON.parseObject(input); 
            String user_id = jsonObject.getString("user_id"); 
            String action = jsonObject.getString("action"); 
            Long timestamp = jsonObject.getLong("timestamp"); 
            if(!StringUtils.isEmpty(user_id) && !StringUtils.isEmpty(action)){  
                UserClick userClick = new UserClick(); 
                userClick.setUserId(user_id); 
                userClick.setTimestamp(timestamp); 
                userClick.setAction(action); 
                out.collect(JSON.toJSONString(userClick)); 
            } 
        } 
    }

处理完成的数据最后被转成 JSONString 发往下游：

env.addSource(consumer) 
            .filter(new UserActionFilter()) 
            .flatMap(new MyFlatMapFunction()) 
            .returns(TypeInformation.of(String.class)) 
            .addSink(new FlinkKafkaProducer( 
                    "127.0.0.1:9092", 
                    "log_user_action", 
                    new SimpleStringSchema() 
            ));

当然你也可以在 ProcessFunction 中将所有的过滤转化逻辑放在一起进行处理。

public class UserActionProcessFunction extends ProcessFunction<String, String> {  
        @Override 
        public void processElement(String input, Context ctx, Collector<String> out) throws Exception {  
            if(! input.contains("CLICK") || input.startsWith("{") || input.endsWith("}")){  
                return; 
            } 
            JSONObject jsonObject = JSON.parseObject(input); 
            String user_id = jsonObject.getString("user_id"); 
            String action = jsonObject.getString("action"); 
            Long timestamp = jsonObject.getLong("timestamp"); 
            if(!StringUtils.isEmpty(user_id) || !StringUtils.isEmpty(action)){  
                UserClick userClick = new UserClick(); 
                userClick.setUserId(user_id); 
                userClick.setTimestamp(timestamp); 
                userClick.setAction(action); 
                out.collect(JSON.toJSONString(userClick)); 
            } 
        } 
    }

我们在上面的例子中将数据发送到了新的 Kafka Topic 中，当然在实际业务中你也可以发往 Elasticsearch、MySQL 等数据库中查询明细。这里我们将处理干净的数据发往了新的 Kafka Topic 中。

到目前为止，我们的数据就被清理干净了，在后续的业务中，可以直接消费新的 Kafka 数据进行 PV 和 UV 的统计了。

# 第三十二讲 Flink 和 Kafka 整合时间窗口设计 #

我们在第 31 课时中讲过，在计算 PV 和 UV 等指标前，用 Flink 将原始数据进行了清洗，清洗完毕的数据被发送到另外的 Kafka Topic 中，接下来我们只需要消费指定 Topic 的数据，然后就可以进行指标计算了。

**Flink 消费 Kafka 数据反序列化**

上一课时定义了用户的行为信息的 Java 对象，我们现在需要消费新的 Kafka Topic 信息，并且把序列化的消息转化为用户的行为对象：

public class UserClick {  
        private String userId; 
        private Long timestamp; 
        private String action; 
        public String getUserId() {  
            return userId; 
        } 
        public void setUserId(String userId) {  
            this.userId = userId; 
        } 
        public Long getTimestamp() {  
            return timestamp; 
        } 
        public void setTimestamp(Long timestamp) {  
            this.timestamp = timestamp; 
        } 
        public String getAction() {  
            return action; 
        } 
        public void setAction(String action) {  
            this.action = action; 
        } 
        public UserClick(String userId, Long timestamp, String action) {  
            this.userId = userId; 
            this.timestamp = timestamp; 
            this.action = action; 
        } 
    } 
    enum UserAction{  
        //点击 
        CLICK("CLICK"), 
        //购买 
        PURCHASE("PURCHASE"), 
        //其他 
        OTHER("OTHER"); 
        private String action; 
        UserAction(String action) {  
            this.action = action; 
        } 
    }

首先，我们需要新建自己的 Kafka Consumer，在第 12 和 24 课时中详细讲解了 Flink 消费 Kafak 消息的原理和实现。在计算 PV 和 UV 的业务场景中，我们选择使用消息中自带的事件时间作为时间特征，代码如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
    // 检查点配置,如果要用到状态后端，那么必须配置 
    env.setStateBackend(new MemoryStateBackend(true)); 
    Properties properties = new Properties(); 
    properties.setProperty("bootstrap.servers", "127.0.0.1:9092"); 
    properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "10"); 
    FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("log_user_action", new SimpleStringSchema(), properties); 
    //设置从最早的offset消费 
    consumer.setStartFromEarliest(); 
    DataStream<UserClick> dataStream = env 
            .addSource(consumer) 
            .name("log_user_action") 
            .map(message -> {  
                JSONObject record = JSON.parseObject(message); 
                return new UserClick( 
                        record.getString("user_id"), 
                        record.getLong("timestamp"), 
                        record.getString("action") 
                ); 
            }) 
            .returns(TypeInformation.of(UserClick.class));

在上面的代码中，我们消费第 31 课时中写入的新的 Kafka Topic，并且将读取到的数据反序列化为 UserClick 的 DataStream。

**水印和窗口设计**

我们得到 UserClick 的 DataStream 后，需要进行水印和窗口的设计，可在 DataStream 上调用 assignTimestampsAndWatermarks 方法。我们在第 8 和 25 课时中，详细讲解过 Flink 支持的时间戳提取器和水印发射器：

*  周期性水印：AssignerWithPeriodicWatermarks
 *  特定事件触发水印：PunctuatedWatermark

由于我们的用户访问日志可能存在乱序，所以使用 BoundedOutOfOrdernessTimestampExtractor 来处理乱序消息和延迟时间，我们指定消息的乱序时间 30 秒，具体代码如下：

SingleOutputStreamOperator<UserClick> userClickSingleOutputStreamOperator = dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<UserClick>(Time.seconds(30)) {  
        @Override 
        public long extractTimestamp(UserClick element) {  
            return element.getTimestamp(); 
        } 
    });

**窗口触发器**

**窗口设置**

到目前为止，我们已经通过读取 Kafka 中的数据，序列化为用户点击事件的 DataStream，并且完成了水印和时间戳的设计和开发。

接下来，按照业务需要，我们需要开窗并且进行一天内用户点击事件的 PV、UV 计算。在第 8 课时中讲解了 Flink 支持的窗口类型：

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

很明显，我们需要使用滚动窗口：TumblingProcessingTimeWindow，并且按照一般逻辑，需每天 0 点到 24 点进行一次计算和输出。

这里我们使用 Flink 提供的滚动窗口，在构造滚动窗口之前先来看一下滚动窗口的实现：

public class TumblingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {  
       private static final long serialVersionUID = 1L; 
       private final long size; 
       private final long offset; 
       protected TumblingEventTimeWindows(long size, long offset) {  
          if (Math.abs(offset) >= size) {  
             throw new IllegalArgumentException("TumblingEventTimeWindows parameters must satisfy abs(offset) < size"); 
          } 
          this.size = size; 
          this.offset = offset; 
       } 
       ...

这里需要注意的是，很多人会认为，窗口时间的开始时间会是我们的代码启动时间。事实上，根据上面的源码可见，TumblingEventTimeWindows 在构造时，需要指定两个参数：窗口的长度和窗口的 offset（默认为 0）。

/** * Creates a new {@code TumblingEventTimeWindows} {@link WindowAssigner} that assigns * elements to time windows based on the element timestamp and offset. * * <p>For example, if you want window a stream by hour,but window begins at the 15th minutes * of each hour, you can use {@code of(Time.hours(1),Time.minutes(15))},then you will get * time windows start at 0:15:00,1:15:00,2:15:00,etc. * * <p>Rather than that,if you are living in somewhere which is not using UTC±00:00 time, * such as China which is using UTC+08:00,and you want a time window with size of one day, * and window begins at every 00:00:00 of local time,you may use {@code of(Time.days(1),Time.hours(-8))}. * The parameter of offset is {@code Time.hours(-8))} since UTC+08:00 is 8 hours earlier than UTC time. * * @param size The size of the generated windows. * @param offset The offset which window start would be shifted by. * @return The time policy. */ 
    public static TumblingEventTimeWindows of(Time size, Time offset) {  
       return new TumblingEventTimeWindows(size.toMilliseconds(), offset.toMilliseconds()); 
    }

我们如何取得某一天的 0 点这个时间起始点呢？TumblingEventTimeWindows 的 of 方法中已经给了提示：

window begins at every 00:00:00 of local time,you may use { @code of(Time.days(1),Time.hours(-8))}.

我们只需要通过 TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)) 就可以指定在中国的 0 点开始创建窗口，然后每天计算一次输出结果即可。

但是，在实际生产环境中，对于大窗口的计算，一般都会设置触发器，以一定的频率输出中间结果，而不是等到一天结束时仅仅触发一次。

这里我们就需要设置数据流的 Trigger 属性。

**触发器设置**

窗口的计算是依赖触发器进行的，每种类型的窗口都有自己的触发器机制，如果用户没有指定，那么会使用默认的触发器。例如 TumblingEventTimeWindows 中自带的触发器如下：

@Override 
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {  
       return EventTimeTrigger.create(); 
    }

可以看到源码中包含了一个 DefaultTrigger：EventTimeTrigger，而 EventTimeTrigger 的实现如下：

public class EventTimeTrigger extends Trigger<Object, TimeWindow> {  
       private static final long serialVersionUID = 1L; 
       private EventTimeTrigger() { } 
       @Override 
       public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {  
          if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {  
             return TriggerResult.FIRE; 
          } else {  
             ctx.registerEventTimeTimer(window.maxTimestamp()); 
             return TriggerResult.CONTINUE; 
          } 
       } 
       @Override 
       public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {  
          return time == window.maxTimestamp() ? 
             TriggerResult.FIRE : 
             TriggerResult.CONTINUE; 
       } 
       @Override 
       public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {  
          return TriggerResult.CONTINUE; 
       } 
       @Override 
       public void clear(TimeWindow window, TriggerContext ctx) throws Exception {  
          ctx.deleteEventTimeTimer(window.maxTimestamp()); 
       } 
       @Override 
       public boolean canMerge() {  
          return true; 
       } 
       @Override 
       public void onMerge(TimeWindow window, 
             OnMergeContext ctx) {  
          long windowMaxTimestamp = window.maxTimestamp(); 
          if (windowMaxTimestamp > ctx.getCurrentWatermark()) {  
             ctx.registerEventTimeTimer(windowMaxTimestamp); 
          } 
       } 
       @Override 
       public String toString() {  
          return "EventTimeTrigger()"; 
       } 
       public static EventTimeTrigger create() {  
          return new EventTimeTrigger(); 
       } 
    }

我们可以看到，EventTimeTrigger 触发器的工作原理是 判断当前的水印是否超过了窗口的结束时间，如果超过则触发对窗口内数据的计算，否则不触发计算。  
Flink 本身提供了不同种类的触发器供我们使用，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 8]  
触发器的类图如上图所示，它们的实际含义如下：

*  EventTimeTrigger：通过对比 Watermark 和窗口的 Endtime 确定是否触发窗口计算，如果 Watermark 大于 Window EndTime 则触发，否则不触发，窗口将继续等待。
 *  ProcessTimeTrigger：通过对比 ProcessTime 和窗口 EndTime 确定是否触发窗口，如果 ProcessTime 大于 EndTime 则触发计算，否则窗口继续等待。
 *  ContinuousEventTimeTrigger：根据间隔时间周期性触发窗口或者 Window 的结束时间小于当前 EndTime 触发窗口计算。
 *  ContinuousProcessingTimeTrigger：根据间隔时间周期性触发窗口或者 Window 的结束时间小于当前 ProcessTime 触发窗口计算。
 *  CountTrigger：根据接入数据量是否超过设定的阈值判断是否触发窗口计算。
 *  DeltaTrigger：根据接入数据计算出来的 Delta 指标是否超过指定的 Threshold 去判断是否触发窗口计算。
 *  PurgingTrigger：可以将任意触发器作为参数转换为 Purge 类型的触发器，计算完成后数据将被清理。

我们在这里可以选择使用：ContinuousProcessingTimeTrigger 来周期性的触发窗口阶段性计算。

实现代码如下：

dataStream.windowAll(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))) 
    		  .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20)))

我们使用 ContinuousProcessingTimeTrigger 每隔 20 秒触发计算，输出中间结果。

到目前为止，完成了除计算 PV 和 UV 的所有前置条件的开发，我们实现了：清洗原始数据、写入新的 Kafka Topic、消费新的 Topic 信息、自定义水印和时间戳、自定义窗口和触发器。

# 第三十三讲 Flink 计算 PV、UV 代码实现 #

上一课时我们学习了 Flink 消费 Kafka 数据计算 PV 和 UV 的水印和窗口设计，并且定义了窗口计算的触发器，完成了计算 PV 和 UV 前的所有准备工作。

接下来就需要计算 PV 和 UV 了。在当前业务场景下，根据 userId 进行统计，PV 需要对 userId 进行统计，而 UV 则需要对 userId 进行去重统计。

下面我们使用不同的方法来统计 PV 和 UV。

**单窗口内存统计**

这种方法需要把一天内所有的数据进行缓存，然后在内存中遍历接收的数据，进行 PV 和 UV 的叠加统计。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 
    env.setStateBackend(new MemoryStateBackend(true)); 
    Properties properties = new Properties(); 
    properties.setProperty("bootstrap.servers", "127.0.0.1:9092"); 
    properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "10"); 
    FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("log_user_action", new SimpleStringSchema(), properties); 
    //设置从最早的offset消费 
    consumer.setStartFromEarliest(); 
    DataStream<UserClick> dataStream = env 
            .addSource(consumer) 
            .name("log_user_action") 
            .map(message -> {  
                JSONObject record = JSON.parseObject(message); 
                return new UserClick( 
                        record.getString("user_id"), 
                        record.getLong("timestamp"), 
                        record.getString("action") 
                ); 
            }) 
            .returns(TypeInformation.of(UserClick.class)); 
    SingleOutputStreamOperator<UserClick> userClickSingleOutputStreamOperator = dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<UserClick>(Time.seconds(30)) {  
        @Override 
        public long extractTimestamp(UserClick element) {  
            return element.getTimestamp(); 
        } 
    }); 
    userClickSingleOutputStreamOperator 
            .windowAll(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))) 
            .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20))) 
    ...

在上一课时中我们已经定义了全局的窗口，并且自定义触发器，每 20 秒触发一次计算输出中间结果。

我们在后面可以继续调用 process 方法，自定义 ProcessFunction 如下：

public class MyProcessFunction extends ProcessAllWindowFunction<UserClick,Tuple2<String,Integer>,TimeWindow> {  
        @Override 
        public void process(Context context, Iterable<UserClick> elements, Collector<Tuple2<String, Integer>> out) throws Exception {  
            HashSet<String> uv = Sets.newHashSet(); 
            Integer pv = 0; 
            Iterator<UserClick> iterator = elements.iterator(); 
            while (iterator.hasNext()){  
                String userId = iterator.next().getUserId(); 
                uv.add(userId); 
                pv = pv + 1; 
            } 
            out.collect(Tuple2.of("pv",pv)); 
            out.collect(Tuple2.of("uv",uv.size())); 
        } 
    }

我们自定义的 ProcessFunction 继承了 ProcessAllWindowFunction 并且覆写了 process 方法。在 process 方法中，新建了一个 HashSet，利用 HashSet 对 userId 去重，即是我们需要的 UV。PV 初始化为 0 并且每来一条数据则加1，最后得到的就是 PV。

TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)) 方法每天从 0 点开始计算并且每天都会清空数据。

这种方法代码简单清晰，但是有很严重的内存占用问题。如果我们的数据量很大，那么所定义的 TumblingProcessingTimeWindows 窗口会缓存一整天的数据，内存消耗非常大。

**分组窗口 + 过期数据剔除**

为了减少窗口内缓存的数据量，我们可以根据用户的访问时间戳所在天进行分组，然后将数据分散在各个窗口内进行计算，接着在 State 中进行汇总。

首先，我们把 DataStream 按照用户的访问时间所在天进行分组：

userClickSingleOutputStreamOperator 
            .keyBy(new KeySelector<UserClick, String>() {  
                @Override 
                public String getKey(UserClick value) throws Exception {  
                    return DateUtil.timeStampToDate(value.getTimestamp()); 
                } 
            }) 
            .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))) 
            .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20))) 
            .evictor(TimeEvictor.of(Time.seconds(0), true)) 
            ...

然后根据用户的访问时间所在天进行分组并且调用了 evictor 来剔除已经计算过的数据。

其中的 DateUtil 是获取时间戳的年月日：

public class DateUtil { 
        public static String timeStampToDate(Long timestamp){ 
            ThreadLocal<SimpleDateFormat> threadLocal
                    = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
            String format = threadLocal.get().format(new Date(timestamp));
            return format.substring(0,10);
        }
    }

接下来看看 Flink 中的剔除器原理和使用方法。

**剔除器**

Flink 中的剔除器可以在 Window Function 执行前或后使用，用来从 Window 中剔除元素。目前 Flink 支持了三种类型的剔除器，具体如下。

*  CountEvictor：数量剔除器。在 Window 中保留指定数量的元素，并从窗口头部开始丢弃其余元素。
 *  DeltaEvictor：阈值剔除器。计算 Window 中最后一个元素与其余每个元素之间的增量，丢弃增量大于或等于阈值的元素。
 *  TimeEvictor：时间剔除器。保留 Window 中最近一段时间内的元素，并丢弃其余元素。

**剔除器源码**

我们在这里使用了 TimeEvictor，来看看该剔除器的源码实现：

private void evict(Iterable<TimestampedValue<Object>> elements, int size, EvictorContext ctx) {  
       if (!hasTimestamp(elements)) {  
          return; 
       } 
       long currentTime = getMaxTimestamp(elements); 
       long evictCutoff = currentTime - windowSize; 
       for (Iterator<TimestampedValue<Object>> iterator = elements.iterator(); iterator.hasNext(); ) {  
          TimestampedValue<Object> record = iterator.next(); 
          if (record.getTimestamp() <= evictCutoff) {  
             iterator.remove(); 
          } 
       } 
    } 
    ...

可以看到，该剔除器首先会找到窗口中元素的最大时间戳，并且找到当前窗口的截断点：最大的时间戳减去要保留的时间段；然后遍历窗口中每一个元素，如果当前元素的时间戳小于等于截断点，则剔除该元素。

所以，我们定义的 TimeEvictor.of(Time.seconds(0), true) 则会剔除每条计算过的元素。

接下来我们实现自己的 ProcessFunction：

public class MyProcessWindowFunction extends ProcessWindowFunction<UserClick,Tuple3<String,String, Integer>,String,TimeWindow>{  
        private transient MapState<String, String> uvState; 
        private transient ValueState<Integer> pvState; 
        @Override 
        public void open(Configuration parameters) throws Exception {  
            super.open(parameters); 
            uvState = this.getRuntimeContext().getMapState(new MapStateDescriptor<>("uv", String.class, String.class)); 
            pvState = this.getRuntimeContext().getState(new ValueStateDescriptor<Integer>("pv", Integer.class)); 
        } 
        @Override 
        public void process(String s, Context context, Iterable<UserClick> elements, Collector<Tuple3<String, String, Integer>> out) throws Exception {  
            Integer pv = 0; 
            Iterator<UserClick> iterator = elements.iterator(); 
            while (iterator.hasNext()){  
                pv = pv + 1; 
                String userId = iterator.next().getUserId(); 
                uvState.put(userId,null); 
            } 
            pvState.update(pvState.value() + pv); 
            Integer uv = 0; 
            Iterator<String> uvIterator = uvState.keys().iterator(); 
            while (uvIterator.hasNext()){  
                String next = uvIterator.next(); 
                uv = uv + 1; 
            } 
            Integer value = pvState.value(); 
            if(null == value){  
                pvState.update(pv); 
            }else {  
                pvState.update(value + pv); 
            } 
            out.collect(Tuple3.of(s,"uv",uv)); 
            out.collect(Tuple3.of(s,"pv",pvState.value())); 
        } 
    }

在 State 中进行 PV 和 UV 的迭代计算，并且将计算过的数据剔除出窗口。保证窗口中的数据不会占用太多内存。

我们在主程序中可以直接使用：

userClickSingleOutputStreamOperator 
            .keyBy(new KeySelector<UserClick, String>() {  
                @Override 
                public String getKey(UserClick value) throws Exception {  
                    return value.getUserId(); 
                } 
            }) 
            .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))) 
            .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20))) 
            .evictor(TimeEvictor.of(Time.seconds(0), true)) 
            .process(new MyProcessWindowFunction());

**使用 BitMap / 布隆过滤器**

我们在计算 UV 时需要进行全局去重，在第 20 课时“Flink 高级应用之海量数据高效去重”中介绍过大数量下的去重方法，这里我们就可以使用 BitMap 或者布隆过滤器进行去重。

假如用户的 ID 可以转化为 Long 型，可以使用 BitMap 进行去重计算 UV：

public class MyProcessWindowFunctionBitMap extends ProcessWindowFunction<UserClick,Tuple3<String,String, Integer>,String,TimeWindow>{  
        private transient ValueState<Integer> uvState; 
        private transient ValueState<Integer> pvState; 
        private transient ValueState<Roaring64NavigableMap> bitMapState; 
    
        @Override 
        public void open(Configuration parameters) throws Exception {  
            super.open(parameters); 
            uvState = this.getRuntimeContext().getState(new ValueStateDescriptor<Integer>("uv", Integer.class)); 
            pvState = this.getRuntimeContext().getState(new ValueStateDescriptor<Integer>("pv", Integer.class)); 
            bitMapState = this.getRuntimeContext().getState(new ValueStateDescriptor("bitMap", TypeInformation.of(new TypeHint<Roaring64NavigableMap>() {  
            }))); 
        } 
        @Override 
        public void process(String s, Context context, Iterable<UserClick> elements, Collector<Tuple3<String, String, Integer>> out) throws Exception {  
            Integer uv = uvState.value(); 
            Integer pv = pvState.value(); 
            Roaring64NavigableMap bitMap = bitMapState.value(); 
            if(bitMap == null){  
                bitMap = new Roaring64NavigableMap(); 
                uv = 0; 
                pv = 0; 
            } 
            Iterator<UserClick> iterator = elements.iterator(); 
            while (iterator.hasNext()){  
                pv = pv + 1; 
                String userId = iterator.next().getUserId(); 
                //如果userId可以转成long 
                bitMap.add(Long.valueOf(userId)); 
            } 
            out.collect(Tuple3.of(s,"uv",bitMap.getIntCardinality())); 
            out.collect(Tuple3.of(s,"pv",pv)); 
    
        } 
    }

上面定义了一个 Roaring64NavigableMap 用来存储用户 ID，最后只需要调用 bitMap.getIntCardinality() 即可求得去重后的 UV 值。

当然了，如果业务中的用户 ID 是字符串类型，不能被转化为 Long 型，那么你可以使用布隆过滤器进行去重。

# 第三十四讲 Flink 和 Redis 整合以及 Redis Sink 实现 #

上一课时我们使用了 3 种方法进行了 PV 和 UV 的计算，分别是全窗口内存统计、使用分组和过期数据剔除、使用 BitMap / 布隆过滤器。到此为止我们已经讲了从数据清洗到水印、窗口设计，PV 和 UV 的计算，接下来需要把结果写入不同的目标库供前端查询使用。

下面我们分别讲解 Flink 和 Redis/MySQL/HBase 是如何整合实现 Flink Sink 的。

**Flink Redis Sink**

我们在第 27 课时，详细讲解过 Flink 使用 Redis 作为 Sink 的设计和实现，分别使用自定义 Redis Sink、开源的 Redis Connector 实现了写入 Redis。

在这里我们直接使用开源的 Redis 实现，首先新增 Maven 依赖如下：

<dependency> 
        <groupId>org.apache.flink</groupId> 
        <artifactId>flink-connector-redis_2.11</artifactId> 
        <version>1.1.5</version> 
    </dependency>

可以通过实现 RedisMapper 来自定义 Redis Sink，在这里我们使用 Redis 中的 HASH 作为存储结构，Redis 中的 HASH 相当于 Java 语言里面的 HashMap：

public class MyRedisSink implements RedisMapper<Tuple3<String,String, Integer>>{  
        /** * 设置redis数据类型 */ 
        @Override 
        public RedisCommandDescription getCommandDescription() {  
            return new RedisCommandDescription(RedisCommand.HSET,"flink_pv_uv"); 
        } 
        //指定key 
        @Override 
        public String getKeyFromData(Tuple3<String, String, Integer> data) {  
            return data.f1; 
        } 
        //指定value 
        @Override 
        public String getValueFromData(Tuple3<String, String, Integer> data) {  
            return data.f2.toString(); 
        } 
    }

上面实现了 RedisMapper 并覆写了其中的 getCommandDescription、getKeyFromData、getValueFromData 3 种方法，其中 getCommandDescription 定义了存储到 Redis 中的数据格式。这里我们定义的 RedisCommand 为 HSET，使用 Redis 中的 HASH 作为数据结构；getKeyFromData 定义了 HASH 的 Key；getValueFromData 定义了 HASH 的值。

然后我们直接调用 addSink 函数即可：

... 
    userClickSingleOutputStreamOperator 
                .keyBy(new KeySelector<UserClick, String>() {  
                    @Override 
                    public String getKey(UserClick value) throws Exception {  
                        return value.getUserId(); 
                    } 
                }) 
                .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))) 
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20))) 
                .evictor(TimeEvictor.of(Time.seconds(0), true)) 
                .process(new MyProcessWindowFunction()) 
                .addSink(new RedisSink<>(conf,new MyRedisSink())); 
    ...

接下来讲解 Flink 和 MySQL 是如何整合实现 Flink Sink 的？

**Flink MySQL Sink**

Flink 在最新版本 1.11 中支持了新的 JDBC Connector，我们可以直接在 Maven 中新增依赖：

<dependency> 
      <groupId>org.apache.flink</groupId> 
      <artifactId>flink-connector-jdbc_2.11</artifactId> 
      <version>1.11.0</version> 
    </dependency>

可以直接使用 JdbcSink 如下：

String driverClass = "com.mysql.jdbc.Driver"; 
    String dbUrl = "jdbc:mysql://127.0.0.1:3306/test"; 
    String userNmae = "root"; 
    String passWord = "123456"; 
    
    userClickSingleOutputStreamOperator 
            .keyBy(new KeySelector<UserClick, String>() {  
                @Override 
                public String getKey(UserClick value) throws Exception {  
                    return value.getUserId(); 
                } 
            }) 
            .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))) 
            .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20))) 
            .evictor(TimeEvictor.of(Time.seconds(0), true)) 
            .process(new MyProcessWindowFunction()) 
            .addSink( 
                    JdbcSink.sink( 
                            "replace into pvuv_result (type,value) values (?,?)", 
                            (ps, value) -> {  
                                ps.setString(1, value.f1); 
                                ps.setInt(2,value.f2); 
                            }, 
                            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() 
                                    .withUrl(dbUrl) 
                                    .withDriverName(driverClass) 
                                    .withUsername(userNmae) 
                                    .withPassword(passWord) 
                                    .build()) 
                    );

JDBC Sink 可以保证 “at-least-once” 语义保障，可通过实现“有则更新、无则写入”来实现写入 MySQL 的幂等性来实现 “exactly-once” 语义。

当然我们也可以自定义 MySQL Sink，直接继承 RichSinkFunction ：

public class MyMysqlSink extends RichSinkFunction<Person> {  
        private PreparedStatement ps = null; 
        private Connection connection = null; 
        String driver = "com.mysql.jdbc.Driver"; 
        String url = "jdbc:mysql://127.0.0.1:3306/test"; 
        String username = "root"; 
        String password = "123456"; 
        // 初始化方法 
        @Override 
        public void open(Configuration parameters) throws Exception {  
            super.open(parameters); 
            // 获取连接 
            connection = getConn(); 
            connection.setAutoCommit(false); 
        } 
        private Connection getConn() {  
            try {  
                Class.forName(driver); 
                connection = DriverManager.getConnection(url, username, password); 
            } catch (Exception e) {  
                e.printStackTrace(); 
            } 
            return connection; 
        } 
        //每一个元素的插入，都会调用一次 
        @Override 
        public void invoke(Tuple3<String,String,Integer> data, Context context) throws Exception {  
            ps.prepareStatement("replace into pvuv_result (type,value) values (?,?)") 
            ps.setString(1,data.f1); 
            ps.setInt(2,data.f2); 
            ps.execute(); 
            connection.commit(); 
        } 
        @Override 
        public void close() throws Exception {  
            super.close(); 
            if(connection != null){  
                connection.close(); 
            } 
            if (ps != null){  
                ps.close(); 
            } 
        } 
    }

我们通过重写 open、invoke、close 方法，数据写入 MySQL 时会首先调用 open 方法新建连接，然后调用 invoke 方法写入 MySQL，最后执行 close 方法关闭当前连接。

最后来讲讲 Flink 和 HBase 是如何整合实现 Flink Sink 的？

**Flink HBase Sink**

HBase 也是我们经常使用的存储系统之一。

> HBase 是一个分布式的、面向列的开源数据库，该技术来源于 Fay Chang 所撰写的 Google 论文“Bigtable：一个结构化数据的分布式存储系统”。就像 Bigtable 利用了 Google 文件系统（File System）所提供的分布式数据存储一样，HBase 在 Hadoop 之上提供了类似于 Bigtable 的能力。HBase 是 Apache 的 Hadoop 项目的子项目。HBase 不同于一般的关系数据库，它是一个适合于非结构化数据存储的数据库；另一个不同的是 HBase 基于列的而不是基于行的模式。

如果你对 HBase 不了解，可以参考官网给出的 [快速入门][Link 2]。

Flink 没有提供直接连接 HBase 的连接器，我们通过继承 RichSinkFunction 来实现 HBase Sink。

首先，我们在 Maven 中新增以下依赖：

<dependency> 
        <groupId>org.apache.hbase</groupId> 
        <artifactId>hbase-client</artifactId> 
        <version>1.2.6.1</version> 
    </dependency> 
    <dependency> 
        <groupId>org.apache.hadoop</groupId> 
        <artifactId>hadoop-common</artifactId> 
        <version>2.7.5</version> 
    </dependency>

接下来通过继承 RichSinkFunction 覆写其中的 open、invoke、close 方法。代码如下：

public class MyHbaseSink extends RichSinkFunction<Tuple3<String, String, Integer>> {  
        private transient Connection connection; 
        @Override 
        public void open(Configuration parameters) throws Exception {  
            super.open(parameters); 
            org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create(); 
            conf.set(HConstants.ZOOKEEPER_QUORUM, "localhost:2181"); 
            connection = ConnectionFactory.createConnection(conf); 
        } 
        @Override 
        public void invoke(Tuple3<String, String, Integer> value, Context context) throws Exception {  
            String tableName = "database:pvuv_result"; 
            String family = "f"; 
            Table table = connection.getTable(TableName.valueOf(tableName)); 
            Put put = new Put(value.f0.getBytes()); 
            put.addColumn(Bytes.toBytes(family),Bytes.toBytes(value.f1),Bytes.toBytes(value.f2)); 
            table.put(put); 
            table.close(); 
        } 
        @Override 
        public void close() throws Exception {  
            super.close(); 
            connection.close(); 
        } 
    }

因为我们的程序是每 20 秒计算一次，并且输出，所以在写入 HBase 时没有使用批量方式。在实际的业务中，如果你的输出写入 HBase 频繁，那么推荐使用批量提交的方式。我们只需要稍微修改一下代码实现即可：

public class MyHbaseSink extends RichSinkFunction<Tuple3<String, String, Integer>> {  
        private transient Connection connection; 
        private transient List<Put> puts = new ArrayList<>(100); 
        @Override 
        public void open(Configuration parameters) throws Exception {  
            super.open(parameters); 
            org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create(); 
            conf.set(HConstants.ZOOKEEPER_QUORUM, "localhost:2181"); 
            connection = ConnectionFactory.createConnection(conf); 
        } 
        @Override 
        public void invoke(Tuple3<String, String, Integer> value, Context context) throws Exception {  
            String tableName = "database:pvuv_result"; 
            String family = "f"; 
            Table table = connection.getTable(TableName.valueOf(tableName)); 
            Put put = new Put(value.f0.getBytes()); 
            put.addColumn(Bytes.toBytes(family),Bytes.toBytes(value.f1),Bytes.toBytes(value.f2)); 
            puts.add(put); 
            if(puts.size() == 100){  
                table.put(puts); 
                puts.clear(); 
            } 
            table.close(); 
        } 
        @Override 
        public void close() throws Exception {  
            super.close(); 
            connection.close(); 
        } 
    }

我们定义了一个容量为 100 的 List，每 100 条数据批量提交一次，可以大大提高写入效率。

# 第三十五讲 项目背景和 Flink CEP 简介 #

从这一课时开始我们将进入“Flink CEP 实时预警系统”的学习，本课时先介绍项目的背景、架构设计。

**背景**

我们在第 11 课时“Flink CEP 复杂事件处理”已经介绍了 Flink CEP 的原理，它是 Flink 提供的复杂事件处理库，也是 Flink 提供的一个非常亮眼的功能，当然更是 Flink 中最难以理解的部分之一。

Complex Event Processing（CEP）允许我们在源源不断的数据中通过自定义的模式（Pattern）检测并且获取需要的数据，还可以对这些数据做对应的处理。Flink 提供了非常丰富的 API 来帮助我们实现非常复杂的模式进行数据匹配。

**Flink CEP 应用场景**

CEP 在互联网各个行业都有应用，例如金融、物流、电商等行业，具体的作用如下。

*  **实时监控**：我们需要在大量的订单交易中发现那些虚假交易，在网站的访问日志中寻找那些使用脚本或者工具“爆破”登录的用户，或者在快递运输中发现那些滞留很久没有签收的包裹等。
 *  **风险控制**：比如金融行业可以用来进行风险控制和欺诈识别，从交易信息中寻找那些可能存在危险交易和非法交易。
 *  **营销广告**：跟踪用户的实时行为，指定对应的推广策略进行推送，提高广告的转化率。

当然了，还要很多其他的场景比如智能交通、物联网行业等，可以应用的场景不胜枚举。

**Flink CEP 的原理**

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

Flink CEP 在运行时会将用户提交的代码转化成 NFA Graph，Graph 中包含状态（Flink 中 State 对象），以及连接状态的边（Flink 中 StateTransition 对象）。

Flink 中的每个模式都包含多个状态，我们进行模式匹配的过程就是进行状态转换的过程，在实际应用 Flink CEP 时，首先需要创建一系列的 Pattern，然后利用 NFACompiler 将 Pattern 进行拆分并且创建出 NFA，NFA 包含了 Pattern 中的各个状态和各个状态间转换的表达式。

我们用官网中的一个案例来讲解 Flink CEP 的应用：

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 在输入流上进行匹配，一旦命中我们的模式，就发出一个报警。

**整体架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 9]  
我们在项目中定义特定事件附带各种上下文信息进入 Kafka，Flink 首先会消费这些信息过滤掉不需要的信息，然后会被我们定义好的模式进行处理，接着触发对应的规则；同时把触发规则的数据输出进行存储。

整个项目的设计可以分为下述几个部分：

*  Flink CEP 源码解析和自定义消息事件
 *  自定义 Pattern 和报警规则
 *  Flink 调用 CEP 实现报警功能

# 第三十六讲 自定义消息事件 #

我们在上一课时中讲了 CEP 的基本原理并且用官网的案例介绍了 CEP 的简单应用。在 Flink CEP 中存在多个比较晦涩的概念，如果你对于这些概念理解有困难，我们可以把：创建系列 Pattern，然后利用 NFACompiler 将 Pattern 进行拆分并且创建出 NFA，NFA 包含了 Pattern 中的各个状态和各个状态间转换的表达式。这整个过程我们可以把 Flink CEP 的使用类比为正则表达式的使用。CEP 中定义的 Pattern 就是正则表达式，而 DataStream 是需要进行匹配的原数据，Flink CEP 通过 DataStream 和 Pattern 进行匹配，然后生成一个经过正则过滤后的 DataStream。

**Flink CEP 源码解析**

Flink CEP 被设计成 Flink 中的算子，而不是单独的引擎。那么当一条数据到来时，Flink CEP 是如何工作的呢？

**Flink DataStream 和 PatternStream**

我们还是用官网中的案例来进行讲解：

我们知道，Flink 中的 DataStream 由相同类型的事件或者元素构成，可以经过复杂的算子比如 Map、Filter 等进行转换。

那么 CEP 是如何和 DataStream 无缝衔接的呢？我们注意到 CEP 的 pattern 方法：

public class CEP {  
        public CEP() {  
        } 
        public static <T> PatternStream<T> pattern(DataStream<T> input, Pattern<T, ?> pattern) {  
            return new PatternStream(input, pattern); 
        } 
        public static <T> PatternStream<T> pattern(DataStream<T> input, Pattern<T, ?> pattern, EventComparator<T> comparator) {  
            PatternStream<T> stream = new PatternStream(input, pattern); 
            return stream.withComparator(comparator); 
        } 
    }

PatternStream 是 Flink CEP 对模式匹配后流的抽象和定义，它把 DataStream 和 Pattern 组合到一起，并且基于 PatternStream 提供了一系列的方法，比如 select、process 等。

我们在 PatternStream 上调用 select 或者 process 方法时，会继续调用到下面的方法：

@Internal 
    final class PatternStreamBuilder<IN> {  
       ... 
        <OUT, K> SingleOutputStreamOperator<OUT> build(TypeInformation<OUT> outTypeInfo, PatternProcessFunction<IN, OUT> processFunction) {  
            Preconditions.checkNotNull(outTypeInfo); 
            Preconditions.checkNotNull(processFunction); 
            TypeSerializer<IN> inputSerializer = this.inputStream.getType().createSerializer(this.inputStream.getExecutionConfig()); 
            boolean isProcessingTime = this.inputStream.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime; 
            boolean timeoutHandling = processFunction instanceof TimedOutPartialMatchHandler; 
            NFAFactory<IN> nfaFactory = NFACompiler.compileFactory(this.pattern, timeoutHandling); 
            CepOperator<IN, K, OUT> operator = new CepOperator(inputSerializer, isProcessingTime, nfaFactory, this.comparator, this.pattern.getAfterMatchSkipStrategy(), processFunction, this.lateDataOutputTag); 
            SingleOutputStreamOperator patternStream; 
            if (this.inputStream instanceof KeyedStream) {  
                KeyedStream<IN, K> keyedStream = (KeyedStream)this.inputStream; 
                patternStream = keyedStream.transform("CepOperator", outTypeInfo, operator); 
            } else {  
                KeySelector<IN, Byte> keySelector = new NullByteKeySelector(); 
                patternStream = this.inputStream.keyBy(keySelector).transform("GlobalCepOperator", outTypeInfo, operator).forceNonParallel(); 
            } 
            return patternStream; 
        } 
        ... 
    }

这其中 NFACompiler.compileFactory，会帮我们完成 Pattern 到 State 的转换。

public static <T> NFACompiler.NFAFactory<T> compileFactory(Pattern<T, ?> pattern, boolean timeoutHandling) {  
        if (pattern == null) {  
            return new NFACompiler.NFAFactoryImpl(0L, Collections.emptyList(), timeoutHandling); 
        } else {  
            NFACompiler.NFAFactoryCompiler<T> nfaFactoryCompiler = new NFACompiler.NFAFactoryCompiler(pattern); 
            nfaFactoryCompiler.compileFactory(); 
            return new NFACompiler.NFAFactoryImpl(nfaFactoryCompiler.getWindowTime(), nfaFactoryCompiler.getStates(), timeoutHandling); 
        } 
    }

其中，compileFactory 方法会生成 State，也就是说把 Pattern 转化为 NFA 中的状态信息，状态会不断地向后追加，所以需要分别先后创建 EndState、MiddleState 和 StartState。这里 Pattern 中专门定义了一个 getPrevious 方法用来获取前一个状态。

void compileFactory() {  
        if (this.currentPattern.getQuantifier().getConsumingStrategy() == ConsumingStrategy.NOT_FOLLOW) {  
            throw new MalformedPatternException("NotFollowedBy is not supported as a last part of a Pattern!"); 
        } else {  
            this.checkPatternNameUniqueness(); 
            this.checkPatternSkipStrategy(); 
            State<T> sinkState = this.createEndingState(); 
            sinkState = this.createMiddleStates(sinkState); 
            this.createStartState(sinkState); 
        } 
    }

**Event 事件处理**

那么，当一条消息进入 Flink CEP 中，是如何被处理的呢？

当 Flink 的事件属性为 EventTime 时，关键代码如下：

public void processElement(StreamRecord<IN> element) throws Exception {  
        long currentTime; 
        if (this.isProcessingTime) {  
            if (this.comparator == null) {  
                NFAState nfaState = this.getNFAState(); 
                long timestamp = this.getProcessingTimeService().getCurrentProcessingTime(); 
                this.advanceTime(nfaState, timestamp); 
                this.processEvent(nfaState, element.getValue(), timestamp); 
                this.updateNFA(nfaState); 
            } else {  
                currentTime = this.timerService.currentProcessingTime(); 
                this.bufferEvent(element.getValue(), currentTime); 
                this.timerService.registerProcessingTimeTimer(VoidNamespace.INSTANCE, currentTime + 1L); 
            } 
        } else {  
            currentTime = element.getTimestamp(); 
            IN value = element.getValue(); 
            if (currentTime > this.lastWatermark) {  
                this.saveRegisterWatermarkTimer(); 
                this.bufferEvent(value, currentTime); 
            } else if (this.lateDataOutputTag != null) {  
                this.output.collect(this.lateDataOutputTag, element); 
            } 
        } 
    }

当 EventTime 大于上一次的 Watermark 时，会把当前的数据加入 elementQueueState 队列中，不符合条件的数据会直接丢弃，关键代码如下：

currentTime = element.getTimestamp(); 
    IN value = element.getValue(); 
    if (currentTime > this.lastWatermark) {  
        this.saveRegisterWatermarkTimer(); 
        this.bufferEvent(value, currentTime); 
    } else if (this.lateDataOutputTag != null) {  
        this.output.collect(this.lateDataOutputTag, element); 
    }

满足条件的数据加入队列后，会在 onEventTime 方法中判断是否触发计算：

public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {  
        PriorityQueue<Long> sortedTimestamps = this.getSortedTimestamps(); 
        NFAState nfaState; 
        long timestamp; 
        for(nfaState = this.getNFAState(); !sortedTimestamps.isEmpty() && (Long)sortedTimestamps.peek() <= this.timerService.currentWatermark(); this.elementQueueState.remove(timestamp)) {  
            timestamp = (Long)sortedTimestamps.poll(); 
            this.advanceTime(nfaState, timestamp); 
            Stream<IN> elements = this.sort((Collection)this.elementQueueState.get(timestamp)); 
            Throwable var7 = null; 
            try {  
                elements.forEachOrdered((event) -> {  
                    try {  
                        this.processEvent(nfaState, event, timestamp); 
                    } catch (Exception var6) {  
                        throw new RuntimeException(var6); 
                    } 
                }); 
            } catch (Throwable var16) {  
                var7 = var16; 
                throw var16; 
            } finally {  
                if (elements != null) {  
                    if (var7 != null) {  
                        try {  
                            elements.close(); 
                        } catch (Throwable var15) {  
                            var7.addSuppressed(var15); 
                        } 
                    } else {  
                        elements.close(); 
                    } 
                } 
            } 
        } 
        this.advanceTime(nfaState, this.timerService.currentWatermark()); 
        this.updateNFA(nfaState); 
        if (!sortedTimestamps.isEmpty() || !this.partialMatches.isEmpty()) {  
            this.saveRegisterWatermarkTimer(); 
        } 
        this.updateLastSeenWatermark(this.timerService.currentWatermark()); 
    }

到此为止，我们可以看到一条数据进入 Flink CEP 中处理的逻辑大概可以分为以下几个步骤：

*  DataSource 中的数据转换为 DataStream；
 *  定义 Pattern，并将 DataStream 和 Pattern 组合转换为 PatternStream；
 *  PatternStream 经过 select、process 等算子转换为 DataStream；
 *  再次转换的 DataStream 经过处理后，sink 到目标库。

**自定义消息事件**

我们在后面的案例中会分别举几个不同的场景，那么我们需要定义几个不同的消息源。

*  第一个场景，连续登录场景

在这个场景中，我们模拟生产环境中可能出现的“爆破登录”现象，模拟用户的登录请求信息：

public class LogInEvent {  
        private Long userId; 
        private String isSuccess; 
        private Long timeStamp; 
        public Long getUserId() {  
            return userId; 
        } 
        public void setUserId(Long userId) {  
            this.userId = userId; 
        } 
        public String getIsSuccess() {  
            return isSuccess; 
        } 
        public void setIsSuccess(String isSuccess) {  
            this.isSuccess = isSuccess; 
        } 
        public Long getTimeStamp() {  
            return timeStamp; 
        } 
        public void setTimeStamp(Long timeStamp) {  
            this.timeStamp = timeStamp; 
        } 
    }

其中 userId 为用户的 ID，isSuccess 表示用户本次登录是否成功，timeStamp 表示用户登录时间戳。

*  第二个场景，超时未支付

在这个场景中，我们要检测出那些用户下单后 5 分钟还没有支付的信息：

public class PayEvent {  
        private Long userId; 
        private String action; 
        private Long timeStamp; 
        public Long getUserId() {  
            return userId; 
        } 
        public void setUserId(Long userId) {  
            this.userId = userId; 
        } 
        public String getAction() {  
            return action; 
        } 
        public void setAction(String action) {  
            this.action = action; 
        } 
        public Long getTimeStamp() {  
            return timeStamp; 
        } 
        public void setTimeStamp(Long timeStamp) {  
            this.timeStamp = timeStamp; 
        } 
    }

同样的，其中 userId 为用户的 ID，action 表示用户的操作事件枚举比如下单、支付等，timeStamp 表示用户操作的时间戳。

*  高频交易，找出活跃账户

在这个场景中，我们模拟账户交易信息中，那些高频的转账支付信息，希望能发现其中的风险或者活跃的用户：

public class TransactionEvent {  
        private String accout; 
        private Double amount; 
        private Long timeStamp; 
        public String getAccout() {  
            return accout; 
        } 
        public void setAccout(String accout) {  
            this.accout = accout; 
        } 
        public Double getAmount() {  
            return amount; 
        } 
        public void setAmount(Double amount) {  
            this.amount = amount; 
        } 
        public Long getTimeStamp() {  
            return timeStamp; 
        } 
        public void setTimeStamp(Long timeStamp) {  
            this.timeStamp = timeStamp; 
        } 
    }

其中 account 表是账户信息，amount 为转账金额，timeStamp 是交易时的时间戳信息。

# 第三十七讲 自定义 Pattern 和报警规则 #

在上一课时提过，PatternStream 是 Flink CEP 对模式匹配后流的抽象和定义，它把 DataStream 和 Pattern 组合到一起，并且基于 PatternStream 提供了一系列的方法，比如 select、process 等。

Flink CEP 的核心在于模式匹配，对于不同模式匹配特性的支持，往往决定相应的 CEP 框架是否能够得到广泛应用。那么 Flink CEP 对模式提供了哪些支持呢？

**Pattern 分类**

Flink CEP 提供了 Pattern API 用于对输入流数据进行复杂事件规则的定义，用来提取符合规则的事件序列。

Flink 中的 Pattern 分为单个模式、组合模式、模式组 3 类。

**单个模式**

复杂规则中的每一个单独的模式定义，就是个体模式。我们既可以定义一个给定事件出现的次数（量词），也可以定义一个条件来决定一个进来的事件是否被接受进入这个模式（条件）。

例如，我们对一个命名为 start 的模式，可以定义如下量词：

// 期望出现4次 
    start.times(4); 
    // 期望出现0或者4次 
    start.times(4).optional(); 
    // 期望出现2、3或者4次 
    start.times(2, 4); 
    // 期望出现2、3或者4次，并且尽可能地重复次数多 
    start.times(2, 4).greedy(); 
    // 期望出现0、2、3或者4次 
    start.times(2, 4).optional(); 
    // 期望出现0、2、3或者4次，并且尽可能地重复次数多 
    start.times(2, 4).optional().greedy(); 
    // 期望出现1到多次 
    start.oneOrMore(); 
    // 期望出现1到多次，并且尽可能地重复次数多 
    start.oneOrMore().greedy(); 
    // 期望出现0到多次 
    start.oneOrMore().optional(); 
    // 期望出现0到多次，并且尽可能地重复次数多 
    start.oneOrMore().optional().greedy(); 
    // 期望出现2到多次 
    start.timesOrMore(2); 
    // 期望出现2到多次，并且尽可能地重复次数多 
    start.timesOrMore(2).greedy(); 
    // 期望出现0、2或多次 
    start.timesOrMore(2).optional(); 
    // 期望出现0、2或多次，并且尽可能地重复次数多 
    start.timesOrMore(2).optional().greedy();

我们还可以定义需要的匹配条件：

start.where(new SimpleCondition<Event>() {  
        @Override 
        public boolean filter(Event value) {  
            return value.getName().startsWith("foo"); 
        } 
    });

上面代码展示了定义一个以“foo”开头的事件。

当然我们还可以把条件组合到一起，例如，可以通过依次调用 where 来组合条件：

pattern.where(new SimpleCondition<Event>() {  
        @Override 
        public boolean filter(Event value) {  
            return ... // 一些判断条件 
        } 
    }).or(new SimpleCondition<Event>() {  
        @Override 
        public boolean filter(Event value) {  
            return ... // 一些判断条件 
        } 
    });

**组合模式**

我们把很多单个模式组合起来，就形成了组合模式。Flink CEP 支持事件之间如下形式的连续策略：

*  严格连续，期望所有匹配的事件严格的一个接一个出现，中间没有任何不匹配的事件；
 *  松散连续
 *  不确定的松散连续

我们直接参考官网给出的案例：

// 严格连续 
    Pattern<Event, ?> strict = start.next("middle").where(...); 
    // 松散连续 
    Pattern<Event, ?> relaxed = start.followedBy("middle").where(...); 
    // 不确定的松散连续 
    Pattern<Event, ?> nonDetermin = start.followedByAny("middle").where(...); 
    // 严格连续的NOT模式 
    Pattern<Event, ?> strictNot = start.notNext("not").where(...); 
    // 松散连续的NOT模式 
    Pattern<Event, ?> relaxedNot = start.notFollowedBy("not").where(...);

对于上述的松散连续和不确定的松散连续这里举个例子来说明一下，例如，我们定义了模式“a b”，对于一个事件序列“a,c,b1,b2”，那么会产生下面的结果：

*  “a”和“b”之间严格连续则返回： \{\} （没有匹配）
 *  “a”和“b”之间松散连续则返回： \{a b1\}
 *  “a”和“b”之间不确定的松散连续则返回： \{a b1\}、\{a b2\}

**模式组**

将一个模式作为条件嵌套在单个模式里，就是模式组。我们举例如下：

Pattern<Event, ?> start = Pattern.begin( 
        Pattern.<Event>begin("start").where(...).followedBy("start_middle").where(...) 
    ); 
    // 严格连续 
    Pattern<Event, ?> strict = start.next( 
        Pattern.<Event>begin("next_start").where(...).followedBy("next_middle").where(...) 
    ).times(3); 
    // 松散连续 
    Pattern<Event, ?> relaxed = start.followedBy( 
        Pattern.<Event>begin("followedby_start").where(...).followedBy("followedby_middle").where(...) 
    ).oneOrMore(); 
    // 不确定松散连续 
    Pattern<Event, ?> nonDetermin = start.followedByAny( 
        Pattern.<Event>begin("followedbyany_start").where(...).followedBy("followedbyany_middle").where(...) 
    ).optional();

**自定义 Pattern**

上一课时定义了几个不同的消息源，我们分别根据需求自定义匹配模式。

*  第一个场景，连续登录场景

在这个场景中，我们需要找出那些 5 秒钟内连续登录失败的账号，然后禁止用户再次尝试登录需要等待 1 分钟。

Pattern.<LogInEvent>begin("start").where(new IterativeCondition<LogInEvent>() {  
        @Override 
        public boolean filter(LogInEvent value, Context<LogInEvent> ctx) throws Exception {  
            return value.getIsSuccess().equals("fail"); 
        } 
    }).next("next").where(new IterativeCondition<LogInEvent>() {  
        @Override 
        public boolean filter(LogInEvent value, Context<LogInEvent> ctx) throws Exception {  
            return value.getIsSuccess().equals("fail"); 
        } 
    }).within(Time.seconds(5));

*  第二个场景，超时未支付

在这个场景中，我们需要找出那些下单后 10 分钟内没有支付的订单。

Pattern.<PayEvent> 
            begin("begin") 
            .where(new IterativeCondition<PayEvent>() {  
                @Override 
                public boolean filter(PayEvent payEvent, Context context) throws Exception {  
                    return payEvent.getAction().equals("create"); 
                } 
            }) 
            .next("next") 
            .where(new IterativeCondition<PayEvent>() {  
                @Override 
                public boolean filter(PayEvent payEvent, Context context) throws Exception {  
                    return payEvent.getAction().equals("pay"); 
                } 
            }) 
            .within(Time.seconds(600)); 
    OutputTag<PayEvent> orderTiemoutOutput = new OutputTag<PayEvent>("orderTimeout") { };

在这里我们使用了侧输出流，并且将正常的订单流和超时未支付的超时流分开：

SingleOutputStreamOperator selectResult = patternStream.select(orderTiemoutOutput, 
            (PatternTimeoutFunction<PayEvent, ResultPayEvent>) (map, l) -> new ResultPayEvent(map.get("begin").get(0).getUserId(), "timeout"), 
            (PatternSelectFunction<PayEvent, ResultPayEvent>) map -> new ResultPayEvent(map.get("next").get(0).getUserId(), "success") 
    ); 
    DataStream timeOutSideOutputStream = selectResult.getSideOutput(orderTiemoutOutput)

*  第三个场景，找出交易活跃用户

在这个场景下，我们需要找出那些 24 小时内至少 5 次有效交易的账户。

Pattern.<TransactionEvent>begin("start").where( 
            new SimpleCondition<TransactionEvent>() {  
                @Override 
                public boolean filter(TransactionEvent transactionEvent) {  
                    return transactionEvent.getAmount() > 0; 
                } 
            } 
    ).timesOrMore(5) 
     .within(Time.hours(24));

# 第三十八讲 Flink 调用 CEP 实现报警功能 #

在上一课时中，我们详细讲解了 Flink CEP 中 Pattern 的分类，需要根据实际生产环境来选择单个模式、组合模式或者模式组。

在前面的课程中我们提到的三种典型场景下，分别根据业务需要实现了 Pattern 的定义，也可以根据自定义的 Pattern 检测到异常事件。那么接下来就需要根据检测到的异常事件发送告警，这一课将从这三种场景入手，来讲解完整的代码实现逻辑。

**连续登录场景**

在这个场景中，我们需要找出那些 5 秒钟内连续登录失败的账号，然后禁止用户，再次尝试登录需要等待 1 分钟。

我们定义的 Pattern 规则如下：

Pattern.<LogInEvent>begin("start").where(new IterativeCondition<LogInEvent>() { 
        @Override
        public boolean filter(LogInEvent value, Context<LogInEvent> ctx) throws Exception { 
            return value.getIsSuccess().equals("fail");
        }
    }).next("next").where(new IterativeCondition<LogInEvent>() { 
        @Override
        public boolean filter(LogInEvent value, Context<LogInEvent> ctx) throws Exception { 
            return value.getIsSuccess().equals("fail");
        }
    }).within(Time.seconds(5));

从登录消息 LogInEvent 可以得到用户登录是否成功，当检测到 5 秒钟内用户连续两次登录失败，则会发出告警消息，提示用户 1 分钟以后再试，或者这时候就需要前端输入验证码才能继续尝试。

首先我们模拟读取 Kafka 中的消息事件：

DataStream<LogInEvent> source = env.fromElements(
            new LogInEvent(1L, "fail", 1597905234000L),
            new LogInEvent(1L, "success", 1597905235000L),
            new LogInEvent(2L, "fail", 1597905236000L),
            new LogInEvent(2L, "fail", 1597905237000L),
            new LogInEvent(2L, "fail", 1597905238000L),
            new LogInEvent(3L, "fail", 1597905239000L),
            new LogInEvent(3L, "success", 1597905240000L)
    ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator()).keyBy(new KeySelector<LogInEvent, Object>() { 
        @Override
        public Object getKey(LogInEvent value) throws Exception { 
            return value.getUserId();
        }
    });

我们模拟了用户的登录信息，其中可以看到 ID 为 2 的用户连续三次登录失败。  
时间戳和水印提取器的代码如下：

private static class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks<LogInEvent>{ 
            private final long maxOutOfOrderness = 5000L;
            private long currentTimeStamp;
            @Nullable
            @Override
            public Watermark getCurrentWatermark() { 
                return new Watermark(currentTimeStamp - maxOutOfOrderness);
            }
            @Override
            public long extractTimestamp(LogInEvent element, long previousElementTimestamp) { 
                Long timeStamp = element.getTimeStamp();
                currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
                System.err.println(element.toString() + ",EventTime:" + timeStamp + ",watermark:" + (currentTimeStamp - maxOutOfOrderness));
                return timeStamp;
            }
        }

我们调用 Pattern.CEP 方法将 Pattern 和 Stream 结合在一起，在匹配到事件后先在控制台打印，并且向外发送。

PatternStream<LogInEvent> patternStream = CEP.pattern(source, pattern);
    SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<LogInEvent, AlertEvent>() { 
        @Override
        public void processMatch(Map<String, List<LogInEvent>> match, Context ctx, Collector<AlertEvent> out) throws Exception { 
            List<LogInEvent> start = match.get("start");
            List<LogInEvent> next = match.get("next");
            System.err.println("start:" + start + ",next:" + next);
    
            out.collect(new AlertEvent(String.valueOf(start.get(0).getUserId()), "出现连续登陆失败"));
        }
    });
    process.printToErr();
    env.execute("execute cep");

我们右键运行，查看结果，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 10]  
可以看到报警事件的触发，其中 AlertEvent 是我们定义的报警事件。

public class AlertEvent { 
        private String id;
        private String message;
        public String getId() { 
            return id;
        }
        public void setId(String id) { 
            this.id = id;
        }
        public String getMessage() { 
            return message;
        }
        public void setMessage(String message) { 
            this.message = message;
        }
        public AlertEvent(String id, String message) { 
            this.id = id;
            this.message = message;
        }
        @Override
        public String toString() { 
            return "AlertEvent{" +
                    "id='" + id + '\'' +
                    ", message='" + message + '\'' +
                    '}';
        }
    }

对于 PatternProcessFunction 中的 processMatch 方法，第一个参数 Map<String, List> match，可以在源码中看到以下的注释：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 11]  
match 这个 Map 的 Key 则是我们在 Pattern 中定义的 start 和 next。

**交易活跃用户**

在这个场景下，我们需要找出那些 24 小时内至少 5 次有效交易的账户。

在上一课时中我们的定义的规则如下：

Pattern.<TransactionEvent>begin("start").where(
            new SimpleCondition<TransactionEvent>() { 
                @Override
                public boolean filter(TransactionEvent transactionEvent) { 
                    return transactionEvent.getAmount() > 0;
                }
            }
    ).timesOrMore(5)
     .within(Time.hours(24));

首先，我们模拟读取 Kafka 中的消息事件：

DataStream<TransactionEvent> source = env.fromElements(
            new TransactionEvent("100XX", 0.0D, 1597905234000L),
            new TransactionEvent("100XX", 100.0D, 1597905235000L),
            new TransactionEvent("100XX", 200.0D, 1597905236000L),
            new TransactionEvent("100XX", 300.0D, 1597905237000L),
            new TransactionEvent("100XX", 400.0D, 1597905238000L),
            new TransactionEvent("100XX", 500.0D, 1597905239000L),
            new TransactionEvent("101XX", 0.0D, 1597905240000L),
            new TransactionEvent("101XX", 100.0D, 1597905241000L)
    ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator()).keyBy(new KeySelector<TransactionEvent, Object>() { 
        @Override
        public Object getKey(TransactionEvent value) throws Exception { 
            return value.getAccout();
        }
    });

然后调用 Pattern.CEP 方法将 Pattern 和 Stream 结合在一起，我们在匹配到事件后先在控制台打印，并且向外发送。

PatternStream<TransactionEvent> patternStream = CEP.pattern(source, pattern);
        SingleOutputStreamOperator<AlertEvent> process = patternStream.process(new PatternProcessFunction<TransactionEvent, AlertEvent>() { 
            @Override
            public void processMatch(Map<String, List<TransactionEvent>> match, Context ctx, Collector<AlertEvent> out) throws Exception { 
                List<TransactionEvent> start = match.get("start");
                List<TransactionEvent> next = match.get("next");
                System.err.println("start:" + start + ",next:" + next);
                out.collect(new AlertEvent(start.get(0).getAccout(), "连续有效交易！"));
            }
        });
        process.printToErr();
        env.execute("execute cep");
    }

最后右键运行查看结果，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 12]  
**超时未支付**

在这个场景中，我们需要找出那些下单后 10 分钟内没有支付的订单。

在上一课时中我们的定义的规则如下：

Pattern.<PayEvent>
            begin("begin")
            .where(new IterativeCondition<PayEvent>() { 
                @Override
                public boolean filter(PayEvent payEvent, Context context) throws Exception { 
                    return payEvent.getAction().equals("create");
                }
            })
            .next("next")
            .where(new IterativeCondition<PayEvent>() { 
                @Override
                public boolean filter(PayEvent payEvent, Context context) throws Exception { 
                    return payEvent.getAction().equals("pay");
                }
            })
            .within(Time.seconds(600));
    OutputTag<PayEvent> orderTiemoutOutput = new OutputTag<PayEvent>("orderTimeout") { };

首先我们模拟读取 Kafka 中的消息事件：

DataStream<PayEvent> source = env.fromElements(
            new PayEvent(1L, "create", 1597905234000L),
            new PayEvent(1L, "pay", 1597905235000L),
            new PayEvent(2L, "create", 1597905236000L),
            new PayEvent(2L, "pay", 1597905237000L),
            new PayEvent(3L, "create", 1597905239000L)
    
    ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator()).keyBy(new KeySelector<PayEvent, Object>() { 
        @Override
        public Object getKey(PayEvent value) throws Exception { 
            return value.getUserId();
        }
    });

然后对匹配的结果进行分流，select 在这里有三个参数，第一个是超时消息的侧输出 Tag，第二个参数是超时消息的处理逻辑，第三个参数是正常的订单消息。

OutputTag<PayEvent> orderTimeoutOutput = new OutputTag<PayEvent>("orderTimeout") { };
    Pattern<PayEvent, PayEvent> pattern = Pattern.<PayEvent>
            begin("begin")
            .where(new IterativeCondition<PayEvent>() { 
                @Override
                public boolean filter(PayEvent payEvent, Context context) throws Exception { 
                    return payEvent.getAction().equals("create");
                }
            })
            .next("next")
            .where(new IterativeCondition<PayEvent>() { 
                @Override
                public boolean filter(PayEvent payEvent, Context context) throws Exception { 
                    return payEvent.getAction().equals("pay");
                }
            })
            .within(Time.seconds(600));
    PatternStream<PayEvent> patternStream = CEP.pattern(source, pattern);
    SingleOutputStreamOperator<PayEvent> result = patternStream.select(orderTimeoutOutput, new PatternTimeoutFunction<PayEvent, PayEvent>() { 
        @Override
        public PayEvent timeout(Map<String, List<PayEvent>> map, long l) throws Exception { 
            return map.get("begin").get(0);
        }
    }, new PatternSelectFunction<PayEvent, PayEvent>() { 
        @Override
        public PayEvent select(Map<String, List<PayEvent>> map) throws Exception { 
            return map.get("next").get(0);
        }
    });
    DataStream<PayEvent> sideOutput = result.getSideOutput(orderTimeoutOutput);
    sideOutput.printToErr();
    env.execute("execute cep");

# 第三十九讲 Flink 面试-基础篇 #

最后一个面试模块分为了 4 个课时：

*  Flink 面试基础篇，包含了 Flink 的整体介绍、核心概念、算子等考察点；
 *  Flink 面试进阶篇，包含了 Flink 中的数据传输、容错机制、序列化、数据热点、反压等实际生产环境中遇到的问题等考察点；
 *  Flink 面试源码篇，包含了 Flink 的核心代码实现、Job 提交流程、数据交换、分布式快照机制、Flink SQL 的原理等考察点；
 *  Flink 面试方案设计篇，主要是生产环境中常见业务场景下的解决方案设计。

本课时将分析 Flink 面试基础相关的面试题，总结出了经常面试的 12 道题。

**面试题 1：请介绍一下 Flink**

这道题是一道很简单的入门题，考察我们队 Flink 整体的掌握情况，我们应该从以下几个基本的概念入手。

Flink 是大数据领域的分布式实时和离线计算引擎，其程序的基础构建模块是流（Streams）和转换（Transformations），每一个数据流起始于一个或多个 Source，并终止于一个或多个 Sink。数据流类似于有向无环图（DAG）。

Flink 提供了诸多高抽象层的 API 以便用户编写分布式任务：

*  DataSet API，对静态数据进行批处理操作，将静态数据抽象成分布式的数据集，用户可以方便地使用 Flink 提供的各种操作符对分布式数据集进行处理，支持 Java、Scala 和 Python；
 *  DataStream API，对数据流进行流处理操作，将流式的数据抽象成分布式的数据流，用户可以方便地对分布式数据流进行各种操作，支持 Java 和 Scala；
 *  Table API，对结构化数据进行查询操作，将结构化数据抽象成关系表，并通过类 SQL 的 DSL 对关系表进行各种查询操作，支持 Java 和 Scala。

此外，Flink 还针对特定的应用领域提供了领域库，例如，Flink ML，Flink 的机器学习库提供了机器学习 Pipelines API 并实现了多种机器学习算法；Gelly、Flink 的图计算库提供了图计算的相关 API 及多种图计算算法的实现。

**面试题 2：Flink 的主要特性是什么？**

这道题考察我们在使用 Flink 的过程中用到的哪些强大的特性。我们在开篇词和第 01 课时“Flink 的应用场景和架构模型”中提到过，Flink 的主要特性包括：批流一体、Exactly-Once、强大的状态管理等。

同时，Flink 还支持运行在包括 Yarn、Mesos、Kubernetes 在内的多种资源管理框架上。Flink 可以扩展到数千核心，其状态可以达到 TB 级别，且仍能保持高吞吐、低延迟的特性。

**面试题 3：Flink 的适用场景有哪些？**

这道题是一道很典型的考察我们掌握 Flink 的广度的问题，要求我们开发者不要局限在自己的一小块业务中，能主动地去学习其他人是如何玩转 Flink 的。我们可以从以下几个方面回答：

Flink 从诞生之初就因为其独特的流式计算特性迅速被各个公司采用到各行各业中，主要的应用场景包括：

*  实时数据计算
 *  实时数据仓库和 ETL
 *  事件驱动型场景，如告警、监控

此外，随着 Flink 对机器学习的支持越来越完善，还可以被用作机器学习和人工智能领域的引擎。

**面试题 4：Flink 和 Spark Streaming 的异同点有哪些？**

分析：这是一个非常宏观的面试题，因为两个框架的不同点非常之多。但是在面试时有非常重要的一点一定要回答出来：Flink 是标准的实时处理引擎，基于事件驱动；而 Spark Streaming 是微批（Micro-Batch）的模型。

在“03 | Flink 的编程模型与其他框架比较”课时中也讲过 Flink 的优势及与其他框架的区别。

*  架构

Spark Streaming 的架构是基于 Spark 的，它的本质是微批处理，每个 batch 都依赖 Driver，我们可以把 Spark Streaming 理解为时间维度上的 Spark DAG。

Flink 也采用了经典的主从模式，DataFlow Graph 与 Storm 形成的拓扑 Topology 结构类似，Flink 程序启动后，会根据用户的代码处理成 Stream Graph，然后优化成为 JobGraph，JobManager 会根据 JobGraph 生成 ExecutionGraph。ExecutionGraph 才是 Flink 真正能执行的数据结构，当很多个 ExecutionGraph 分布在集群中，就会形成一张网状的拓扑结构。

*  容错

针对 Spark Streaming 任务，我们可以配置对应的 Checkpoint，也就是保存点。当任务出现 Failover 的时候，会从 Checkpoint 重新加载，使得数据不丢失。但是这个过程会导致原来的数据重复处理，不能做到“只处理一次”语义。

Flink 基于两阶段提交实现了精确的一次处理语义，我们在第 14 课时 “Flink Exactly-once 实现原理解析” 中有过详细的讲解。

*  反压（BackPressure）

反压是分布式处理系统中经常遇到的问题，当消费者速度低于生产者的速度时，则需要消费者将信息反馈给生产者使得生产者的速度能和消费者的速度进行匹配。

Spark Streaming 为了实现反压这个功能，在原来的架构基础上构造了一个“速率控制器”，“速率控制器”会根据几个属性，比如任务的结束时间、处理时长、处理消息的条数等计算一个速率。在实现控制数据的接收速率中用到了一个经典的算法，即“PID 算法”。

Flink 没有使用任何复杂的机制来解决反压问题，Flink 在数据传输过程中使用了分布式阻塞队列。我们知道在一个阻塞队列中，当队列满了以后发送者会被天然阻塞住，这种阻塞功能相当于给这个阻塞队列提供了反压的能力。

*  时间机制

Spark Streaming 支持的时间机制有限，只支持处理时间。 Flink 支持了流处理程序在时间上的三个定义：处理时间、事件时间、注入时间；同时也支持 watermark 机制来处理滞后数据。

**面试题 5：请谈谈 Flink 的组件栈和数据流模型**

Flink 的组件栈和数据流模型是我们使用 Flink 的基础，只有了解这些组件栈和数据模型才能入手进行 Flink 项目的开发。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 14]  
Flink 自身提供了不同级别的抽象来支持我们开发流式或者批量处理程序，上图描述了 Flink 支持的 4 种不同级别的抽象。

对于我们开发者来说，大多数应用程序不需要上图中的最低级别的 Low-level 抽象，而是针对 Core API 编程，比如DataStream API（有界/无界流）和DataSet API（有界数据集）。这些流畅的 API 提供了用于数据处理的通用构建块，比如各种形式用户指定的转换、连接、聚合、窗口、状态等。

Table API 和 SQL是 Flink 提供的更为高级的 API 操作，Flink SQL 是 Flink 实时计算为简化计算模型，降低了用户使用实时计算门槛而设计的一套符合标准 SQL 语义的开发语言。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 15]  
Flink 程序的基本构建是数据输入来自一个 Source，Source 代表数据的输入端，经过 Transformation 进行转换，然后在一个或者多个 Sink 接收器中结束。数据流（Stream）就是一组永远不会停止的数据记录流，而转换（Transformation）是将一个或多个流作为输入，并生成一个或多个输出流的操作。在执行时，Flink 程序映射到 Streaming Dataflows，由流（Streams）和转换操作（Transformation Operators）组成。

**面试题 6：请谈谈 Flink 中的角色和作用各是什么？**

这道题考察我们生产环境中 Flink 的集群是如何工作的，开发者有没有主动去研究过生产环境 Flink 集群的部署过程，任务的启动流程等。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 16]  
Flink 程序在运行时主要有 TaskManager、JobManager、Client 三种角色。其中 JobManager 扮演着集群中的管理者 Master 的角色，它是整个集群的协调者，负责接收 Flink Job、协调检查点、Failover 故障恢复等，同时管理 Flink 集群中从节点 TaskManager。

TaskManager 是实际负责执行计算的 Worker，在其上执行 Flink Job 的一组 Task，每个 TaskManager 负责管理其所在节点上的资源信息，比如内存、磁盘、网络，在启动的时候将资源的状态向 JobManager 汇报。

Client 是 Flink 程序提交的客户端，当用户提交一个 Flink 程序时，会先创建一个 Client，该 Client 首先会对用户提交的 Flink 程序进行预处理，然后提交到 Flink 集群中处理，所以 Client 需要从用户提交的 Flink 程序配置中获取 JobManager 的地址，并建立到 JobManager 的连接，将 Flink Job 提交给 JobManager。

**面试题 7：请谈谈 Flink 中的计算资源 Task Slot**

Flink 中的计算资源是非常容易被忽略的一个知识点，它直接关系着我们线上 Flink 任务调优的原理和过程，变相的考察我们对于大数量下的 Flink 资源分配和调优是否熟悉。

在 Flink 集群中，一个 TaskManger 就是一个 JVM 进程，并且会用独立的线程来执行 Task，为了控制一个 TaskManger 能接受多少个 Task，Flink 提出了 Task Slot 的概念。

我们可以简单地把 Task Slot 理解为 TaskManager 的计算资源子集。假如一个 TaskManager 拥有 5 个 Slot，那么该 TaskManager 的计算资源会被平均分为 5 份，不同的 Task 在不同的 Slot 中执行，避免资源竞争。但需要注意的是，Slot 仅仅用来做内存的隔离，对 CPU 不起作用。那么运行在同一个 JVM 的 Task 可以共享 TCP 连接，以减少网络传输，在一定程度上提高了程序的运行效率，降低了资源消耗。

**面试题 8：请谈谈 Flink 中的并行度设置**

Flink 使用并行度来定义某一个算子被切分成多少个子任务。我们的 Flink 代码会被转换成逻辑视图，在实际运行时，根据用户的并行度配置会被转换成对应的子任务进行执行。

Flink 本身支持不同级别来设置我们任务并行度的方法，它们分别是：

*  算子级别
 *  环境级别
 *  客户端级别
 *  集群配置级别

需要特别指出的是，设置并行度的优先级依次是：算子级别 > 环境级别 > 客户端级别 > 集群配置级别。

**面试题 9：请谈谈 Flink 支持的集群规模情况**

这个面试题需要我们把生产环境中使用的集群规模、配置情况、Flink 版本等介绍清楚；同时说明部署模式（一般是 Flink on Yarn），除此之外，我们也可以同时在小集群（少于 5 个节点）和拥有 TB 级别状态的上千个节点上运行 Flink 任务。

**面试题 10：请谈谈 Flink 中的时间分类**

这道题考察我们开发者对于 Flink 中的时间类型的掌握程度，Flink 中的时间是最基本的概念，我们任何一个实时计算任务都需要进行实践特性的选择。

我们在 “08 | Flink 窗口、时间和水印”课时中提到过 Flink 中的时间分为三种：

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

**面试题 11：请谈谈 Flink 中的水印**

Flink 中的水印是难度较高的一个知识点，这个问题考察开发者在实际生产中如何处理乱序问题的，是基于什么样的考虑选择不同的水印生成方式。

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

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

Flink 中的水印有两种：

*  周期生成水印 AssignerWithPeriodicWatermarks，周期默认的时间是 200ms；
 *  按需要生成水印 PunctuatedWatermark，它适用于根据接收到的消息判断是否需要产生水印。

**面试题 12：请谈谈 Flink 中的窗口**

这道题考察的是开发者在不同的业务场景中是如何选择时间窗口的，当然变相的就会和你的业务关联，为什么这样选择窗口？他有什么特点？

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

# 第四十讲 Flink 面试-进阶篇 #

Flink 面试进阶篇主要包含了 Flink 中的数据传输、容错机制、序列化、数据热点、反压等实际生产环境中遇到的问题等考察点。这一阶段主要考察我们对 Flink 掌握的深度，也是给面试官留下好印象的关键环节。

**面试题 1：请谈谈你对 Flink Table & SQL 的了解情况？以及 TableEnvironment 这个类有什么样的作用？**

这道题考察的是对 Flink Table & SQL 的掌握情况，要知道 Flink 中的 TableEnvironment 有 5 个之多，那么对于 TableEnvironment 的掌握就显得特别重要。

TableEnvironment 是 Table API 和 SQL 集成的核心概念，它主要被用来创建 Table & SQL 程序的上下文执行环境，这个类主要用来：

*  在内部 catalog 中注册表
 *  注册外部 catalog
 *  执行 SQL 查询
 *  注册用户定义（标量、表或聚合）函数
 *  将 DataStream 或 DataSet 转换为表
 *  持有对 ExecutionEnvironment 或 StreamExecutionEnvironment 的引用

**面试题 2：Flink SQL 的原理是什么？请谈谈整个 SQL 的解析过程**

我们在第 05 课时“Flink Table & SQL 编程和案例” 中提过 Flink 把 SQL 的解析、优化和执行教给了 Calcite。

这道题比较深入，我们在使用 Flink Table & SQL 时要知道它的运行原理，才能针对性地做一些调优。

基于此，一次完整的 SQL 解析过程如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 17]

*  用户使用对外提供的 Stream SQL 语法来开发业务应用；
 *  用 calcite 对 StreamSQL 进行语法检验，语法检验通过后，转换成 calcite 的逻辑树节点，最终形成 calcite 的逻辑计划；
 *  采用 Flink 自定义的优化规则和 calcite 火山模型、启发式模型共同对逻辑树进行优化，生成最优的 Flink 物理计划；
 *  对物理计划采用 janino codegen 生成代码，生成用低阶 API DataStream 描述的流应用，并提交到 Flink 平台执行。

**面试题 3：请谈谈你对 Flink 容错的理解**

Flink 的容错机制是非常重要的一个特性，我们在回答这个问题时，要把 CheckPoint 和 State 都答出来，并且阐述一下原理。

Flink 实现容错主要靠强大的 CheckPoint 和 State 机制。Checkpoint 负责定时制作分布式快照、对程序中的状态进行备份；State 用来存储计算过程中的中间状态。

Flink 提供了三种可用的状态后端用于在不同情况下进行状态后端的保存：

*  MemoryStateBackend
 *  FsStateBackend
 *  RocksDBStateBackend

**面试题 4 ：Flink 是如何保证 Exactly-once 语义的**

这道题在面试时一定会遇到，Flink 的“精确一次”语义的支持是区别于其他框架最显著的特性之一。

Flink 通过实现两阶段提交和状态保存来实现端到端的一致性语义，分为以下几个步骤：

*  开始事务（beginTransaction），创建一个临时文件夹，然后把数据写入这个文件夹里面；
 *  预提交（preCommit），将内存中缓存的数据写入文件并关闭；
 *  正式提交（Commit），将之前写完的临时文件放入目标目录下，这代表着最终的数据会有一些延迟；
 *  丢弃（Abort），丢弃临时文件。

我们在第 14 课时“Flink Exactly-once 实现原理解析”中还对 Flink 分布式快照的核心元素 Barrier 和两阶段提交的原理做过详细讲解，可以参考。

**面试题 5：请谈谈 Flink 中的内存管理**

Flink 的内存管理一般会在面试者与面试官讨论反压问题解决方案时被提及，对于内存管理的准确掌握也是我们进行内存调优和配置的前提。

Flink 并不是将大量对象存在堆上，而是将对象都序列化到一个预分配的内存块上，此外，Flink 大量使用了堆外内存。如果需要处理的数据超出了内存限制，则会将部分数据存储到硬盘上。

Flink 为了直接操作二进制数据实现了自己的序列化框架。

理论上 Flink 的内存管理分为以下 3 部分。

*  Network Buffers：这个是在 TaskManager 启动的时候分配的，这是一组用于缓存网络数据的内存，每个块是 32K，默认分配 2048 个，可以通过“taskmanager.network.numberOfBuffers”修改。
 *  Memory Manage pool：大量的 Memory Segment 块，用于运行时的算法（Sort/Join/Shuffle 等），这部分启动时会被分配。根据配置文件中的各种参数来计算内存的分配方法，并且内存的分配支持预分配和 lazy load，默认懒加载的方式。
 *  User Code，这个是除了 Memory Manager 之外的内存用于 User Code 和 TaskManager 本身的数据结构。

**面试题 6：请谈谈你遇到的数据倾斜问题？Flink 中的 Window 出现了数据倾斜，你有什么解决办法？**

这道题也是面试中一定会被提及的问题，数据倾斜是大数据领域最常见的问题之一。Flink 中对于数据倾斜的调优极其重要，我们一定要掌握。

我们在第 16 课时“如何处理生产环境中的数据倾斜问题”中对于这个问题有过一整节的讲解。

产生数据倾斜的原因主要有 2 个方面：

*  业务上有严重的数据热点，比如滴滴打车的订单数据中北京、上海等几个城市的订单量远远超过其他地区；
 *  技术上大量使用了 KeyBy、GroupBy 等操作，错误的使用了分组 Key，人为产生数据热点。

因此解决问题的思路也很清晰：

*  业务上要尽量避免热点 key 的设计，例如我们可以把北京、上海等热点城市分成不同的区域，并进行单独处理；
 *  技术上出现热点时，要调整方案打散原来的 key，避免直接聚合；此外 Flink 还提供了大量的功能可以避免数据倾斜。

另外我们还可以结合实际工作中出现的问题做举例说明。

**面试题 7：Flink 任务出现很高的延迟，你会如何入手解决类似问题？**

这是一个实操性很强的问题，我们在回答这类问题一定要结合实际情况来说。在生产环境中处理这类问题会从以下方面入手：

*  在 Flink 的后台任务管理中，可以看到 Flink 的哪个算子和 task 出现了反压；
 *  资源调优和算子调优，资源调优即对作业中的 Operator 并发数（Parallelism）、CPU（Core）、堆内存（Heap\_memory）等参数进行调优；
 *  作业参数调优，并行度的设置、State 的设置、Checkpoint 的设置。

**面试题 8： 请谈谈你们是如何处理脏数据的？**

这也是一个开放性的面试题，建议你结合自己的实际业务来谈。比如可以通过一个 fliter 算子将不符合规则的数据过滤出去。当然了，我们也可以在数据源头就将一些不合理的数据抛弃，不允许进入 Flink 系统参与计算。

**面试题 9：请谈谈 Operator Chain 这个概念？**

算子链是我们进行任务调优一定会遇到的问题，主要考察我们对于概念是否正确理解，实际操作中有怎样的指导作用。

为了更高效地分布式执行，Flink 会尽可能地将 Operator 的 Subtask 链接（Chain）在一起形成 Task，每个 Task 在一个线程中执行。将 Operators 链接成 Task 是非常有效的优化，它能减少：

*  线程之间的切换；
 *  消息的序列化/反序列化；
 *  数据在缓冲区的交换；
 *  延迟的同时提高整体的吞吐量。

这就是我们所说的算子链。

**面试题 10：Flink 在什么情况下才会把 Operator Chain 在一起形成算子链？**

这道题是有标准答案的，截止 1.11 版本，Flink 的算子之间形成算子链需要以下条件：

*  上下游的并行度一致
 *  下游节点的入度为 1（即下游节点没有来自其他节点的输入）
 *  上下游节点都在同一个 Slot Group 中
 *  下游节点的 Chain 策略为 ALWAYS（可以与上下游链接，Map、Flatmap、Filter 等默认是 ALWAYS）
 *  上游节点的 Chain 策略为 ALWAYS 或 HEAD（只能与下游链接，不能与上游链接，Source 默认是 HEAD）
 *  两个节点间数据分区方式是 Forward
 *  用户没有禁用 Chain

**面试题 11：请谈谈 Flink 中的分布式快照机制是如何实现的？**

这道题也是考察我们对于 Flink 掌握深度的一个问题，需要你完全理解 Flink 中的分布式快照机制的实现原理。

Flink 容错机制的核心部分是制作分布式数据流和操作算子状态的一致性快照，这些快照充当一致性 Checkpoint，系统可以在发生故障时回滚。Flink 用于制作这些快照的机制在“分布式数据流的轻量级异步快照”中进行了描述，它受到分布式快照的标准 Chandy-Lamport 算法的启发，专门针对 Flink 的执行模型而定制。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 18]  
barrier 在数据流源处被注入并行数据流中。快照 n 的 barrier 被插入的位置（我们称为 Sn）是快照所包含的数据在数据源中最大位置。例如，在 Apache Kafka 中，此位置将是分区中最后一条记录的偏移量，将该位置 Sn 报告给 Checkpoint 协调器（Flink 的 JobManager）。

接着 barrier 向下游流动。当一个中间操作算子从其所有输入流中收到快照 n 的 barrier 时，它会为快照 n 发出 barrier 进入其所有输出流中。 一旦 sink 操作算子（流式 DAG 的末端）从其所有输入流接收到 barrier n，它就向 checkpoint 协调器确认快照 n 完成。在所有 sink 确认快照后，意味着快照已完成。

一旦完成快照 n，job 将永远不再向数据源请求 Sn 之前的记录，因为此时这些记录（及其后续记录）将已经通过整个数据流拓扑，也即已经被处理结束。

**总结**

这一课时我们主要介绍了 Flink 面试中的一些有深度的问题，这些问题是我们在技术面试中能否通过的关键部分，也是考察我们对于 Flink 掌握深度的部分。由于篇幅的原因我们不能 100% 覆盖所有高级进阶的面试题目，建议你参考我之前的文章[《全网第一份 | Flink 学习面试灵魂40问，看看你能答上来几个》][_ Flink _40]和[《Flink 面试通关手册》][Flink]。

# 第四十一讲 Flink 面试-源码篇 #

Flink 的源码篇包含了 Flink 的核心代码实现、Job 提交流程、数据交换、分布式快照机制、Flink SQL 的原理等考察点。你应该记得，我们在前面近 40个课时中几乎每一课时都有一定的篇幅是源码阅读，源码部分的考察是面试时十分重要的一关，如果你对 Flink 的源码有一定的研究而不仅仅停留在使用阶段，那么你的面试成功率将得到大幅提升。

我们本课时将把 Flink 面试中源码部分最高频的考点列举出来。

**面试题 1：请从源码层面谈谈 Flink Job 的提交流程**

用户提交的 Flink Job 会被转化成一个 DAG 任务运行。Flink Job 的提交涉及的类主要包括：StreamGraph、JobGraph、ExecutionGraph、TaskManager、JobManager、ResourceManager。

JobManager 会把接收到的需要执行的应用程序进行打包，然后把 JobGraph 转换成可以执行的ExecutionGraph，接着向 ResourceManager 请求执行任务所需要的资源，也就是我们之前课程中提到的 Slot，如果资源获取成功 JobManager 会负责所有的任务调度，比如 Checkpoint，并且将任务派发给 TaskManager 去执行。

**面试题 2：我们通常说的 Flink 的“三层图”结构是什么意思？**

这道题要求面试者掌握Flink 框架引擎划分执行计划的详细过程。

一个 Flink 任务的 DAG 生成计算图大致经历以下3个过程。

*  首先，StreamGraph 的拓扑结构最接近代码层面，主要由 StreamNode 和 StreamEdge 构成，其中 StreamNode 对应着 Operator，它们通过 StreamEdge 进行链接。
 *  其次，JobGraph 是能被 Flink 引擎识别的数据结构，由JobVertex、JobEdge和 IntermediateDataSet3个元素组成。我们可以把JobGraph 形象地比喻为一个抽水系统，JobVertex 是水泵，JobEdge 是水管，而 IntermediateDataSet 则是中间的蓄水池。
 *  最后，ExecutionGraph 由 JobGraph 转换而来，包含了任务具体执行所需的内容，是最贴近底层实现的执行图。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 19]  
上图清晰展现了 Flink 各个图之间的转换过程和工作原理。

**面试题 3：请从源码层面谈谈 JobManger 在集群中扮演了什么角色？**

我们在第03课时“Flink 的编程模型与其他框架比较” 中专门用了一小节的内容来讲解 Flink 中的角色和作用，其中 JobManager 是 Flink 集群中非常重要的一个角色。

JobManager 负责整个 Flink 集群任务的调度及资源的管理，从客户端中获取提交的应用，然后根据集群中 TaskManager 上 TaskSlot 的使用情况，为提交的应用分配相应的 TaskSlot 资源并命令 TaskManager 启动从客户端中获取的应用。

JobManager 相当于整个集群的 Master 节点，且整个集群有且只有一个活跃的 JobManager，负责整个集群的任务管理和资源管理。

JobManager 和 TaskManager 之间通过 Actor System 进行通信，获取任务执行的情况并通过 Actor System 将应用的任务执行情况发送给客户端。

同时在任务执行的过程中，Flink JobManager 会触发 Checkpoint 操作，每个 TaskManager 节点收到 Checkpoint 触发指令后，完成 Checkpoint 操作，所有的 Checkpoint 协调过程都是在 Fink JobManager 中完成。

当任务完成后，Flink 会将任务执行的信息反馈给客户端，并且释放掉 TaskManager 中的资源以供下一次提交任务使用。

**面试题 4：JobManger 在集群启动过程中起到什么作用？**

一定要注意，这道题目和面试题 3 有一定的相似性，但是这里会要求我们从源码层面介绍 JobManager 在启动过程中涉及的主要类和方法。

首先，我们要回答出 JobManager 的主要职责，主要包括负责整个 Flink 集群任务调度和资源的管理，并且负责接收 Flink 作业、调度 Task、收集作业状态和管理 TaskManager。

然后，如果开发者能从源码层面回答出涉及的关键方法，会大大增加面试官的印象。

*  RegisterTaskManager：它由想要注册到 JobManager 的 TaskManager 发送，注册成功则通过 AcknowledgeRegistration 消息进行 Ack。
 *  SubmitJob：将 Job 提交给 Client，提交的信息是 JobGraph 形式的作业描述信息。
 *  CancelJob：请求取消指定ID的作业，成功会返回 CancellationSuccess，否则返回 CancellationFailure。
 *  UpdateTaskExecutionState：由 TaskManager 发送，用来更新执行节点（ExecutionVertex）的状态；成功则返回 true，否则返回 false。
 *  RequestNextInputSplit：TaskManager 上的 Task 请求下一个输入 split，成功则返回 NextInputSplit，否则返回 null。
 *  JobStatusChanged：它意味着作业的状态（RUNNING、CANCELING、FINISHED等）发生变化，这个消息由 ExecutionGraph 发送。

**面试题 5：TaskManager 在集群中扮演了什么角色？**

这道题和面试题 3 类似，TaskManager 是集群中的 Worker 角色，我们在第 03 课时“Flink 的编程模型与其他框架比较”中也对 TaskManager 做过详细讲解。

TaskManager 相当于整个集群的 Slave 节点，负责具体的任务执行和对应任务在每个节点上的资源申请和管理。

客户端通过将编写好的 Flink 应用编译打包，提交到 JobManager，然后 JobManager 会根据已注册在 JobManager 中 TaskManager 的资源情况，将任务分配给有资源的 TaskManager 节点，然后启动并运行任务。

TaskManager 从 JobManager 接收需要部署的任务，然后使用 Slot 资源启动 Task，建立数据接入的网络连接，接收数据并开始数据处理。同时 TaskManager 之间的数据交互都是通过数据流的方式进行的。

可以看出，Flink 的任务运行其实是采用多线程的方式，这和 MapReduce 多 JVM 并行的方式有很大的区别，Flink 能够极大提高 CPU 使用效率，在多个任务和 Task 之间通过 TaskSlot 方式共享系统资源，每个 TaskManager 中通过管理多个 TaskSlot 资源池对资源进行有效管理。

**面试题 6：TaskManger 的启动过程是怎样的？**

相比 JobManager而言，TaskManager 的启动流程较为简单，启动类入口为org.apache.flink.runtime.taskexecutor.TaskManagerRunner。核心的入口如下所示：

public static void runTaskManager(Configuration configuration, ResourceID resourceId) throws Exception { 
            //主要初始化一些service,并新建一个TaskExecutor
    		final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration,resourceId);
    		//调用TaskExecutor的start()方法
            taskManagerRunner.start();
    }

启动过程中主要进行 Slot 资源的分配、RPC 服务的初始化，以及JobManager 进行通信等。

**面试题 7：Flink 计算资源的调度是如何实现的？**

这类问题的难度都非常高，考察开发人员在进行资源配置和调优的过程中是否对源码有足够的了解。我们在第 17 课时“生产环境中的并行度和资源配置”中对 Flink 中的资源有过详细讲解。

TaskManager 中最细粒度的资源是 Task slot，代表了一个固定大小的资源子集，每个 TaskManager 会将其所占有的资源平分给它的 slot。

通过调整 task slot 的数量，用户可以定义 task 之间是如何相互隔离的。每个 TaskManager 有一个 slot，也就意味着每个 task 运行在独立的 JVM 中；每个 TaskManager中有多个 slot 的话，也就意味着多个 task 运行在同一个 JVM 中。

而在同一个 JVM 进程中的 task，可以共享 TCP 连接（基于多路复用）和心跳消息，可以减少数据的网络传输，也能共享一些数据结构，在一定程度上减少了每个 task 的消耗。 每个 slot 可以接受单个 task，也可以接受多个连续 task 组成的 pipeline，如下图所示。FlatMap 函数占用一个 taskslot，而 key Agg 函数和 sink 函数共用一个 taskslot：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 20]  
**面试题 8：请你谈谈 Flink 的数据抽象及数据交换过程？**

这个问题涉及 Flink 中的内存设计和定义，需要你将核心的类答出来。

我们知道 JVM 在对象序列化上有一些固有的缺陷，主要体现在存储对象的密度较低，含有大量不需要的信息，并且 FGC 还会对整体的吞吐和响应有严重影响。

为了降低这些影响，Flink 实现了自己的内存管理。主要体现在，Flink 定义了自己的内存抽象：MemorySegment。

public abstract class MemorySegment { 
        protected static final Unsafe UNSAFE;
        protected static final long BYTE_ARRAY_BASE_OFFSET;
        private static final boolean LITTLE_ENDIAN;
        protected byte[] heapMemory;
        protected long address;
        protected long addressLimit;
        protected int size;
        protected Object owner;
        MemorySegment(byte[] buffer, Object owner) { 
            if (buffer == null) { 
                throw new NullPointerException("buffer");
            } else { 
                this.heapMemory = buffer;
                this.address = BYTE_ARRAY_BASE_OFFSET;
                this.size = buffer.length;
                this.addressLimit = this.address + (long)this.size;
                this.owner = owner;
            }
        }
        ...

我们可以把MemorySegment 看作是一个 32KB大的内存块的抽象，这块内存既可以是 JVM 里的一个 byte\[\]，也可以是堆外内存（DirectByteBuffer）。

在 MemorySegment 这个抽象之上，Flink 抽象出了两个关键的进行数据转换的类：

*  Buffer，用于各个 TaskManager 进行数据传输；
 *  StreamRecord，用于 Java 对象和 Buffer 对象互相转换。

**面试题9：谈谈 Flink 的 SQL 部分是如何实现的？**

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

我们在第 5 课时“Flink Table & SQL 编程和案例”中进行过非常详细的讲解。

Flink 把 SQL 的解析、优化和执行教给了[Calcite][]。从图中可以看到，无论是批查询 SQL 还是流式查询 SQL，都会经过对应的转换器 Parser 转换成为节点树 SQLNode tree，然后生成逻辑执行计划 Logical Plan，逻辑执行计划在经过优化后生成真正可以执行的物理执行计划，交给 DataSet 或者 DataStream 的 API 去执行。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 21]  
上图清晰地展示了 SQL 的整个解析和转换过程，当 SQL 被转换为可以执行的物理执行计划后会被分发到各个 TaskManager 节点中去，并且最后翻译成代码运行。

**总结**

这一课时我们介绍了 Flink 源码面试中的一些经典题目，在之前的课程中都会花一定的篇幅去介绍源码和实现。可以这么说，源码的阅读决定了我们是否真正掌握了一个框架，建议你在学习 Flink 的每个模块时都要养成阅读源码的习惯。另外，抽空也可以参考我之前的面试相关文章[《全网第一份 | Flink学习面试灵魂40问，看看你能答上来几个》][_ Flink _40]和[《Flink面试通关手册》][Flink]。

# 第四十二讲 Flink 面试-方案设计篇 #

Flink 的方案设计面试题目在面试中，是面试官了解我们项目的最直接的问题，它通常体现在面试者回答自己的项目整体是如何设计的？Flink 在你的项目中起到什么作用？有没有在应用过程中对 Flink 有一些定制开发等。

如何介绍自己的项目，为什么技术选型 Flink 也代表我们对于 Flink 框架的了解程度，我们本课时将介绍 Flink 典型的应用场景和整体的方案设计。

**面试题1：基于 Flink 的实时数据仓库是如何做的？**

我们在第 01 课时“Flink 的应用场景和架构模型”和第 21 课时“Flink 在实时计算平台和实时数据仓库中的应用”中对基于 Flink 的实时数仓做过详细介绍。

在回答这类问题时，我们要从 Flink 的优势开始入手，介绍基于 Flink 的实时数仓建设的关键技术选型和整体设计。

传统的离线数据仓库将业务数据集中进行存储后，以固定的计算逻辑定时进行ETL和其他建模后产出报表等应用。离线数据仓库主要是构建 T+1 的离线数据，通过定时任务每天拉取增量数据，然后创建各个业务相关的主题维度数据，对外提供 T+1 的数据查询接口。计算和数据的实时性均较差，业务人员无法根据自己的即时性需要获取几分钟之前的实时数据。数据本身的价值随着时间的流逝会逐步减弱，因此数据发生后必须尽快地达到用户的手中，实时数仓的构建需求也应运而生。

总之就是一句话：时效性的要求。

Flink 在实时数仓和实时 ETL 中有天然的优势：

*  状态管理，实时数仓里面会进行很多的聚合计算，这些都需要对状态进行访问和管理，Flink 支持强大的状态管理；
 *  丰富的 API，Flink 提供了极为丰富的多层次 API，包括 Stream API、Table API 及 Flink SQL；
 *  生态完善，实时数仓的用途广泛，Flink 支持多种存储（HDFS、ES 等）；
 *  批流一体，Flink 已经在将流计算和批计算的 API 进行统一。

我们在进行实时数仓的设计时，一般也是分为 ODS 源数据接入层、DWD 明细层、DWS 汇总层、ADM 应用层。

这其中比较关键的技术点：实时数仓的明细层的汇总一般是基于 Flink 等接入 Kafka 消息进行关联的，维度表的数据一般会放在 HDFS、HBase 中作为明细层的补充。另外，在实时数据仓库中要选择不同的 OLAP 库来满足即席查询。OLAP 数据库的选择至关重要，你可以参考我之前写的文章[《你需要的不是实时数仓 | 你需要的是一款强大的OLAP数据库》][_ _OLAP]。

此外，我们在和面试官交流的过程中，可以将自己的实时数据的架构图完整画出来，这样会给人很深刻的印象，你可以参考下面两张架构设计图。

网易严选的实时数据仓库设计图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 22]  
美团的实时数仓分层架构模型图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 23]  
**面试题2：基于 Flink 的实时计算平台是如何做的？**

实时计算平台的搭建是随着越来越多的业务场景需要实时计算而产生的。由于离线计算天然时效性不强，一般都是隔天级别的滞后，业务数据随着实践的推移，本身的价值就会逐渐减少。

我们在第 03 课时“Flink 的编程模型与其他框架比较”中，提到过 Flink 自身独有的优势。由于 Flink 在架构、容错、反压上表现出来的优势和特性，使得 Flink 在实时计算平台的搭建上占有一席之地。

我们在第 21 课时“Flink 在实时计算平台和实时数据仓库中的应用”中提到过一般的实时计算平台的构成大都是由以下几部分构成。

*  实时数据收集层

在实际业务中，大量的实时计算都是基于消息系统进行的数据收集和投递，这都离不开强大的消息中间件。目前业界使用最广的是 Kafka，另外一些重要的业务数据还会用到其他消息系统比如 RocketMQ 等。Kafka 因为高吞吐、低延迟的特性，特别适合大数量量、高 QPS 下的业务场景，而 RocketMQ 则在事务消息、一致性上有独特的优势。

*  实时计算层

Flink 在计算层同时支持流式及批量分析应用，这就是我们所说的批流一体，Flink 承担了数据的实时采集、实时计算和下游发送的角色。随着 Blink 的开源和一些其他实时产品的开源，支持可视化、SQL 化的开发模式已经越来越普及。

*  数据存储层

这里是我们的实时数据存储层，存储层除了传统 MySQL 等存储引擎以外，还会根据场景数据的不同存储在 Redis、HBase、OLAP 中。而这一层我个人认为最重要的技术选型则是 OLAP。OLAP 的技术选型直接制约着数据存储层和数据服务层的能力。关于 OLAP 的技术选型，可以参考[这里][_ _OLAP]。

*  数据服务层

数据服务层会提供统一的对外查询、多维度的实时汇总，加上完善的租户和权限设计，能够支持多部门、多业务的数据需求。另外，基于数据服务层还会有数据的展示、大屏、指标可视化等。

我们在回答这个面试问题时，可以结合自己的实际项目回答，例如，在实时数据收集层中的技术选型是什么、支持了多大的数据量以及取得的业务成果；并且可以把自己的架构图完整地画出来。你也可以参考下面的架构图来画出自己的整体架构图。

美团的实时计算平台图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 24]  
微博的实时计算平台图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 25]  
**面试题3：有没有做过基于 Flink SQL 的计算平台？你有什么思路？**

基于 Flink SQL 的实时计算平台是很多大公司提高开发效率、降低开发成本、降低维护成本作出的选择。一般的 SQL 开发平台都会有以下几个模块：

*  SQL 开发和提交模块
 *  完善的权限管理体系
 *  UDF 管理模块
 *  资源调度模块
 *  日志和监控模块

这其中，SQL 的开发和提交是核心，成熟的 SQL 开发平台应该是和公司常用的组件栈打通，基于 Flink 原生的 SQL 模块支持更为丰富的 Source 和 Sink。并且支持丰富的作业配置，如作业 Checkpoint、重启策略等。

权限管理模块中，我们的 SQL 应该在作业级别进行多租户的权限管控，对于原表和目标表更应该精确到开发者级别。

UDF 模块应该支持用户自定义基于业务的复杂 UDF，并且可以快捷方便地更新版本。

资源和调度模块可以支持算子级别的并行度和内存设置，并且应该支持作业在不同集群间进行切换。

日志和监控模块中是我们进行作业调优、异常监控非常重要的模块，我们应该有专门的服务进行日志收集和查询，异常监控方面可以做到指定到人且不同渠道（邮件、电话、钉钉等）的报警。

**总结**

本课时我们讲解了 Flink 面试中非常重要的一个模块：方案设计篇。本文介绍了 Flink 在实时数据仓库、实时计算平台、SQL 平台三种业务场景中的方案设计，大家在面试的过程中要结合实际的业务背景进行讲解。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/32c970711b824ce4bf88b46933de9d2d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/29e89e6da3a1474ab6f2e568f4271be2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/3db4b8e8b7884bed8ba8b6c0d8dd309f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/55106e313b74477cbcd76bba2da7faad.png
[20200929143252973.png_pic_center]: /images/20221123/91a100ea151d4b5da24a1244cef926df.png
[Link 1]: http://archive.apache.org/dist/flume/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/41ffe4c9318d452483e6f233afdd9532.png
[20200929144545885.png_pic_center]: /images/20221123/53b70d37f1d9402badb102e063a6238c.png
[20200929144604436.png_pic_center]: /images/20221123/92a765bd438146a0a145febf4b633b1f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/df1a7c30cac942f3ac1b802194e64a4a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/c5c041b52e3343f28383a3b56123bc0e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/94cadb21913c4dbbbce9ae98a9076578.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/fb5b3e419d5541eaa31e281a8d6c3e36.png
[Link 2]: http://hbase.apache.org/book.html#quickstart
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/518d2f8a01a844e198d8a22ffacabb19.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/bb7e3e9a893349e1944dd59f3685d697.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221123/e99ad47bfd5a4d20827fbef372484362.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221123/09e705c47a31445db1bd9b1738603c19.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 13]: /images/20221123/53ab63128a4f477bb85e085271ef46c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 14]: /images/20221123/4f691c0571094b39b33ad74715903ec9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 15]: /images/20221123/aace3e1eff5f44878f22f38be2cf6e40.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 16]: /images/20221123/96768e5bbb874b6882e663453613d49d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 17]: /images/20221123/a72a3db76e714ba99af05de36087f6f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 18]: /images/20221123/1df2bcfe1ade4dcba0ea16928f5865e5.png
[_ Flink _40]: https://mp.weixin.qq.com/s/-J-UZ6vs8BD9sYjdeMOmTQ
[Flink]: https://mp.weixin.qq.com/s/xRqrojjFITuhswtjNJo7OQ
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 19]: /images/20221123/3324256c482841ba93a120866fa60194.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 20]: /images/20221123/d3d836656ddf42fd8b25011509c38e4b.png
[Calcite]: https://calcite.apache.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 21]: /images/20221123/97167e35c11a497c848768115dc7a19d.png
[_ _OLAP]: https://mp.weixin.qq.com/s/9MZ9ztr8fYJTl1HchqtQqA
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 22]: /images/20221123/68cc6b8e37e84d10a9fd55b0583be504.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 23]: /images/20221123/0eaf6767c67d4a3e9426fc8b6a7eb2d4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 24]: /images/20221123/9d6f4797dc9844718cbd3f16362314c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 25]: /images/20221123/8dce56482ab24b379b12cda3ea8290e0.png