摘要： Flink入坑指南系列文章，从实际例子入手，一步步引导用户零基础入门实时计算/Flink，并成长为使用Flink的高阶用户。本文属个人原创，仅做技术交流之用，笔者才疏学浅，如有错误，欢迎指正。转载请注明出处，侵权必究。

Flink接口

Flink支持三种接口，也就是三种写作业的方式：

• SQL：实时计算产品与开源Flink相比，提供更全面的语法支持，目前95%的用户都是使用SQL解决其问题
• TableAPI：表达能力与SQL相同
• DataStream API：底层API，需要一定基础
  本系列文章会从流式SQL开始介绍。重点是帮助用户理解批和流SQL的区别，使用户能快速上手，在Flink上写出正确的流SQL。

不会介绍实时计算/Flink SQL的语法细节，关于SQL语法或各内置函数的用法，请参考文档：帮助手册
有志于了解FlinkSQL实现原理或研究Flink代码的同学，可以参考《Apache Flink 漫谈系列 - SQL概览》

有问题？点我提问

明确需求

接上一章内容，本章计算一个指标：

1. 当天0点开始，全网的成交量
   使用系统：

• 上游（SOURCE）：Kafka
• 下游（SINK）：MySQL

开始写第一个作业

原始数据

主要逻辑

源表：

-- 源表DDL
create table src(
    ctime timestamp,       -- 交易时间戳
    category_id varchar,   -- 类目id
    shop_id varchar,       -- 店铺id
    item_id varchar,       -- 商品id
    price double           -- 价格
)

-- 结果表DDL
create table sink(
    cdate date,            -- 日期
    gmv_daily double       -- 从零点开始，每天的全网成交金额
)

批SQL写法

一般批的写法：

SELECT 
    date_format(ctime, '%Y%m%d') as cdate, -- 将数据从时间戳格式（2018-12-04 15:44:54），转换为date格式(20181204)
       SUM(price) AS gmv_daily
 FROM src
 GROUP BY date_format(ctime, '%Y%m%d') ; --按照天做聚合

结果：

特点：

• 每次执行前，数据库中都保存了当天已经入库的全量数据
• 每次执行SQL，都会拿到__一个__返回值，然后SQL执行结束

Flink SQL写法

SELECT 
    date_format(ctime, '%Y%m%d') as cdate, -- 将数据从时间戳格式（2018-12-04 15:44:54），转换为date格式(20181204)
       SUM(price) AS gmv_daily
 FROM src
 GROUP BY date_format(ctime, '%Y%m%d') ; --按照天做聚合

特点：

• Flink SQL是个常驻进程，一个SQL文件，就对应与一个Flink作业。如果用户不杀掉这个作业，这个SQL就会一直存在
• 每来一条数据，这个Flink作业都会输出一个值。如果MySQL结果表中没有加主键，那看到的结果如下：

如果把MySQL结果表中的cdate字段作为主键，那么每来一条数据，这个Flink作业都会输出一个值，三条数据的主键相同，因此会覆盖之前的结果，等三条数据都经过Flink计算后，得到的结果如下：

原理介绍

这个例子中，批和流的SQL相同，从最终结果看也相同。但是批引擎（比如MySQL/Hive等）的执行模式，和流引擎（如Flink）是完全不同的。这就导致同一个SQL在处理数据的行为上，会有很多区别。如果要深入使用Flink SQL，并且保证结果的正确性，成为Flink SQL调优专家，就需要对Flink底层实现有一定的了解。接下来每一章的例子之后，都会介绍一下本章所用的基础原理。但不会讲实现细节，需要了解实现细节的同学，可以follow flink源代码。

1. 从批和流SQL的行为开始 - 持续查询

从直观上看，批和流SQL行为非常不同：

• 批SQL每执行一次，只返回一次结果，针对计算时刻数据库中数据的快照做计算
• 流SQL只要启动，就会一直执行，在有数据的情况下会不停的产生结果
  这里涉及到的流计算中新增的一个非常重要且基础的概念持续查询。简单理解，持续查询的特点，Flink 作业会一直执行其SQL的逻辑，每到一条新数据，都会触发下游计算，从而源源不断的产生输出。

该SQL中有两个关键操作：

• Group By：分组操作，在批SQL和流SQL中，group by的行为是相同的，都是按某几个字段对数据进行分组。

• SUM：求和操作，在批SQL和流SQL中，SUM求和操作的语义上相同的，都是对每个分组的某个字段，做求和操作。但是批SQL和流SQL的实现方式是不同的：

  • 批：已经知道了所有数据，把每个分组的求和字段拿出来相加即可。
  • 流：数据是一条条进入系统的，并不知道全部数据，来一条加一条，产出一条结果。

2. group by + SUM（）- 状态

Flink SQL持续计算过程中，数据源源不断流入，以本文中例子来看，三条数据先后进入Flink，Flink中需要按cdate做一个全局group by，然后对每个cdate中所有数据的price做一个聚合运算（SUM），过程如下：

1. item_01: sum1=0+10
2. item_02: sum2=sum1+11.1=21.1
3. item_03: sum3=sum2+12.3=33.4

这就产生了个问题：每条数据的SUM计算，都要依赖上一条数据的计算结果。Flink在计算的时候，会保留这些中间结果么？答案是：会保存。而这些中间结果，就是一个作业的状态(state)的一部分。

关于state的几个关键问题：

• __是不是所有的作业都有状态？__是，但是只有聚合操作/JOIN等，state中才会保存中间结果。简单的运算，如filter等，state中不需要保存中间结果。State官方文档，请参阅：Apache Flink Doc，Flink Committer解析请参考 《Apache Flink 漫谈系列 - State》
• state会保存多长时间？ state会一直保存么？不会。流计算中state都是有过期时间的。实时计算产品中，默认是36小时。
• __state过期是什么意思？__state过期，是指36小时前的状态都会被删掉。这样做是为了节省系统存储空间，在大窗口join计算过程中，需要保存很多数据，如果都存下来，集群磁盘会满。

• __state过期规则是什么？__state过期规则

  • 不同group by分组的state互不相关
  • 与该group by分组上次state更新的时间有关，如果 __现在时间 - 某个key的state最近更新的时间>state过期时间__，则这个group by分组的state会被清理掉。
• state过期时间能调么？能，实时计算产品中，单作业可以配置这个参数：state.backend.rocksdb.ttl.ms=129600000，单位毫秒。
• __如果state过期会对作业造成什么影响__：
  以这个例子来说，极端一点，假设我们把state过期参数设成5分钟。如果3条原始数据进入Flink的时间__相差5分钟以上__，以ptime定义数据进入flink的时间，如下所示：

此时:

1. item_01(ptime=2018-12-04 15:45:00): sum1=0+10
2. item_02(ptime=__2018-12-04 15:45:10__): sum2=sum1+11.1=21.1
3. item_03(ptime=__2018-12-04 15:52:00__): sum3=0+12.3=12.3
   item_02和item_03之间超过5min，因此state中sum2的值被清掉，导致item_03到来时，sum3的值计算错误。

该例子中：

• ctime就是数据产生的时间，流计算中的术语叫event time（事件时间）
• ptime是数据进入Flink的事件，流计算中术语叫process time（处理时间）
  这两个时间域是流计算的基础概念。要正确使用流计算，还需要理解以下这两个概念，相关文章：

《Streaming System 第一章：Streaming 101》
《Streaming System 第二章：The What- Where- When- and How of Data Processing》

SQL优化

上述SQL中，每来一条数据就要就要计算一次，在输入数量大的情况下，容易产生性能瓶颈。每来一条数据，后端都会read和write一次state，发生序列化和反序列化操作，甚至是磁盘的 I/O 操作。对复杂场景，比如JOIN/TopN等，因此状态的相关操作通常都会成为整个任务的性能瓶颈。
如何避免这个问题呢？使用microbatch策略。microbatch顾名思义，就是攒批。不是来一条处理一条，而是攒一批再处理。相关配置如下：
​​

#攒批的间隔时间，使用 microbatch 策略时需要加上该配置，且建议和 blink.miniBatch.allowLatencyMs 保持一致
blink.microBatch.allowLatencyMs=5000
# 使用 microbatch 时需要保留以下两个 minibatch 配置
blink.miniBatch.allowLatencyMs=5000
# 防止OOM，每个批次最多缓存多少条数据
blink.miniBatch.size=20000。

相关知识点

• 持续查询

  • 《Apache Flink 漫谈系列 - 持续查询(Continuous Queries)》

• 状态(state):

  • Apache Flink Doc
  • Flink Committer解析请参考 《Apache Flink 漫谈系列 - State》

• 流计算中的时间域概念 - Process Time vs. Event Time

  • 《Streaming System 第一章：Streaming 101》
  • 《Streaming System 第二章：The What- Where- When- and How of Data Processing》

有问题？点我提问