江,阴宜,兴

本次，我们从 0 开始逐步剖析 Flink sql 的来龙去脉以及核心概念，并附带完整的示例程序，希望对大家有帮助！

本文大纲

一、快速体验 Flink sql

为了快速搭建环境体验 Flink sql，我们使用 Docker 来安装一些基础组件，包括 zk 和 kafka，如果你有这个环境，可以略过了。

在 Centos 7 上安装 Docker 环境，具体见这个链接，此处就不细说了：
https://blog.csdn.net/qq_24434251/article/details/105712044

1、拉取安装并执行 zookeeper 镜像

docker pull debezium/zookeeper
docker run -d -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper

2、拉取安装并执行 kafka 镜像

docker pull debezium/kafka
docker run -d -it --rm --name kafka -p 9092:9092 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.56.10:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 --link zookeeper:zookeeper debezium/kafka

3、进入 kafka 容器内的命令行

docker exec -it kafka /bin/bash

4、创建一个 topic

/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.56.10:2181 --topic user_log --partitions 1 --replication-factor 1

5、在 IDEA 中启动程序

这里不贴代码太长了，具体程序可以参见我的 github：
https://github.com/nicekk/Flink-Practice

6、写入数据

/kafka/bin/kafka-console-producer.sh --broker-list  192.168.56.10:9092 --topic user_log

数据样例：

{"user_id":123,"item_id":345,"ts":"2021-01-05 23:04:00"}
{"user_id":345,"item_id":345,"ts":"2021-01-05 23:04:00"}

7、结果输出：

二、数据类型系统

继续说明 Flink sql 使用之前，我们还需要谈一谈 Flink 的数据类型系统。

Flink 作为一款高性能的计算框架，必然绕不开分布式计算、数据传输和持久化这些问题。

在数据传输过程中，要对数据进行序列化和反序列化：序列化就是将一个内存对象转换成二进制串，形成网络传输或者持久化的数据流；反序列化将二进制串转换为内存对象，这样就可以直接在编程语言中读写这个对象。

Flink 是运行在 JVM 上的，计算过程中会有大量的数据存储在内存中，这就会面临一些问题，如 Java 对象存储密度较低等。

针对这些问题，最常用的方案就是自己实现一个显示的内存管理，用自定义的内存池来进行内存的分配回收，接着将序列化后的对象存储到内存块中。

所以，Flink 对数据类型推断越准确，越能更早的完成数据类型检查，帮助 Flink 更好的规划内存，节省存储空间。

比如下面这个类，Tuple3 <Integer，Double，Person> ，包含三种数据类型。

其中 Person 包含两个字段，分别是 id 和 name。

如图，int 占四个字节，通过 IntSerializer 序列化操作之后，给它分配 4 个字节就行了。对象之间可以紧凑的在一起存储，不像 Java 的序列化会有更多的存储损耗。

（数据类型系统，是 Flink 一个非常大的领域，我们会单开一篇文章来详细说明，此处只想说明一下数据类型的重要作用）

三、在无界数据流上怎么执行 sql

在有界的数据集上执行 sql ，相信大家每天都深有体会，每天都会做。有界的数据集是静止的，离线模式下，sql 可以访问完整的数据集，查询产生结果后就终止了。

而数据流是无限的，意味着程序需要一直运行，等待数据进入并进行处理，这样的一种模式如何和 sql 关联起来呢？

这里我们要引入两个概念：动态表（Dynamic Table）和持续查询（Continuous Queries ）。

（1）动态表

如果想用 sql 去分析一个数据流，那第一件事就是要把流转换成表。

如下图，左边是一个点击的事件流，有姓名，事件时间，点击的 url 信息。右边是一张表，也有这三个字段。

从左边的流到右边的表，是一个逻辑上的映射过程，并没有将数据持久化。

随着左边流事件源源不断的到来，右边的表的记录也会一直追加更新。

这样一直变化的表，就称为「动态表」。

（2）连续查询

对于动态表的查询就被称为是连续查询。

如下图，将下面的 sql 作用在动态表上，就产生了一个持续查询，因为这个查询一直不会终止掉，并且每个事件到来时，都会产生一次查询。

查询的结果，会生成一个新的动态表。

select 
  user,
  count(url) as cnt
  from clicks
 group by user;

Mary，./home）这条数据到来，产生查询的结果：【Mary，1】

（Bob，./cart） 这条数据到来，会在动态表上追加一条 Bob 的记录，最终的结果为：【Mary，1】【Bob，1】

（Mary，./prod?id=1） 这条数据到来，会更新动态表的 Mary 的记录，最终结果为：【Mary，2】【Bob，1】

（Liz，./home） 这条数据到来，会在动态表上追加一个记录，最终结果为：【Mary，2】【Bob，1】【Liz，1】

这样的话，我们就可以使用 sql 在动态表上连续查询，产生新的动态表。（实际上，在上一篇中，我们已经知道，sql 最终是会变成程序执行的）。

（3）查询限制

由于流是无限的，我们不得不思考一个问题，那就是所有的查询语句都能在流上执行吗？

答案是否定的，主要是两点原因，一是维护的状态比较大，二是计算更新的成本高。

由于连续查询会一直运行，为了更新之前产生的结果，需要维护所有的输出行，这样的话，内存中存储的数据会越来越大。

然后有时候，即使只来了一条记录，也需要重新计算和更新之前大部分的结果行，这样的查询也不适合作为连续查询。

比如下面的 sql，求排名，每次来数据之后，都需要计算大量数据的排名：

SELECT user, RANK() OVER (ORDER BY lastLogin)
  FROM ( 
          SELECT user, 
                 MAX(cTime) AS lastAction 
            FROM clicks GROUP BY user
  );

（4）结果输出

最后一个问题，Flink 是一个计算引擎，自身不存储数据，那么它是如何表示更新数据并更新到外部存储？
这里我们举两个例子来说明

1、目标表是控制台

我们可以回到上面的那个例子，例子中，由于目标是控制台，可以任意打印结果。

-- 源表，连接 kafka，从最新的地方开始消费
CREATE TABLE user_log (
  user_id bigint,
  item_id bigint,
  ts TIMESTAMP
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_log',
  'scan.startup.mode' = 'latest-offset',
  'properties.bootstrap.servers' = '192.168.56.10:9092',
  'format' = 'json',
  'json.fail-on-missing-field' = 'false'
)

-- 目标表是控制台，直接打印
CREATE TABLE user_log_result(
  user_id bigint,
  cnt bigint
) WITH (
  'connector' = 'print'
)

-- 查询的 sql，一个简单的 group by ，统计源表的 user_id 数量，写到目标表
insert into user_log_result select user_id,count(1) cnt from user_log group by user_id

当我们第一次输入一条数据时： {"user_id":123,"item_id":345,"ts":"2021-01-05 23:04:00"}
控制台上打印：

3> +I(123,1)

当我们再次输入一条数据时：{"user_id":123,"item_id":123,"ts":"2021-01-05 23:04:00"}
控制台上打印了两条数据：

3> -U(123,1)

3> +U(123,2)

+I，-U，+U 表示一行数据的 changelog，+I 表示是新增的数据，-U 表示之前的记录已经被更新，之前的记录要回撤，+U 表示本次更新的数据。

可以看到，输出结果是以对于每行产生 changelog 的形式来表示的。

如果 sink 阶段要使用 DataStream Api，可以把动态表变成流，继续 sink 到下游节点。如果使用 sql，则直接可以发送到下游。

具体程序见：

2、目标表是 Kafka 的时候

-- 源表，连接 kafka，从最新的地方开始消费
CREATE TABLE user_log (
  user_id bigint,
  item_id bigint,
  ts TIMESTAMP
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_log',
  'scan.startup.mode' = 'latest-offset',
  'properties.bootstrap.servers' = '192.168.56.10:9092',
  'format' = 'json',
  'json.fail-on-missing-field' = 'false'
)

-- 目标表是 Kafka
CREATE TABLE user_log_result (
  user_id bigint,
  cnt bigint
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_log_result',
  'scan.startup.mode' = 'latest-offset',
  'properties.bootstrap.servers' = '192.168.56.10:9092',
  'format' = 'json'
)

-- 查询的 sql，一个简单的 group by ，统计源表的 user_id 数量，写到目标表
insert into user_log_result select user_id,count(1) cnt from user_log group by user_id

此时再运行，直接就报错了，提示信息如下：

Exception in thread "main" org.apache.flink.table.api.TableException: 
Table sink 'default_catalog.default_database.user_log_result' doesn't support consuming update changes 
which is produced by node GroupAggregate(groupBy=[user_id], select=[user_id, COUNT(*) AS cnt])

大意是：这是一个 Group 的聚合，而目标表 user_log_result （kafka）不支持更新的数据。kafka 只能支持一直新增的数据。

如果我们换成下面的 sql，数据只有新增不会更新，就可以运行了。当然也可以把目标表换成其他可以更新的介质，如 MysqL ，hbase 等等。

insert into user_log_result select user_id,count(1) cnt from user_log group by user_id

具体程序见：

四、时间、INTERVAL 与 窗口计算

窗口计算永远是流计算的核心，窗口将无限流切分为有限大小的数据集，可以对这个有限数据集进行计算。

在谈到窗口的时候，总是会情不自禁冒出 N 多的概念，比如：事件时间，处理时间，窗口开始时间，窗口结束时间，滑动窗口，滚动窗口，窗口大小，水印 .......

在最新的 Flink sql 中，已经可以在 DDL 中定义所有的这一切了，让我们各个击破他们。

1. INTERVAL

Interval 这个东西，并不是 Flink sql 中特有的，在 ANSI sql 中就有，下面我们以 Oracle 举例来说明。

首先得有 Oracle 环境，这里我们使用 Docker 来搭建，具体教程见这个链接：

https://blog.csdn.net/qq_24434251/article/details/112341197

INTERVAL 表示一段时间差，直接建表体验一下

create table INTERVAL_TAB
(
    DURATION INTERVAL DAY (2) TO SECOND (5)
)

表示建一个表，字段 duration 表示 天 到 秒，括号的数字表示精度。

insert into interval_tab (duration) values (interval '3 12:32' day(3) to minute );

插入的这条数据表示一段时间：3天12小时32分钟

可能感觉这个没啥用，比如我问你在公司入职几年了，你可以轻松说出来，但是如果我问你在公司入职多少天了，这就很复杂了，中间的闰年，2 月都要考虑，有了这样的表示方法就很方便了。

比如可以很轻易的算出今天之前100天，是哪一天：

select sysdate,sysdate - interval '100' day(3)  as "当前时间-100天" from dual;

有了 INTERVAL ，我们就可以轻松表示窗口的时间长短了。

2. 窗口计算

滚动窗口 - 使用ProcessingTime

-- 源表，user_name 用户名，data 数据
CREATE TABLE user_actions (
	user_name string,
	data string,
	user_action_time as PROCTIME()
   ) WITH (
	'connector' = 'kafka',
	'topic' = 'user_log',
	'scan.startup.mode' = 'latest-offset',
	'properties.bootstrap.servers' = '192.168.56.10:9092',
	'format' = 'json',
	'json.fail-on-missing-field' = 'false'
)

-- 结果表
CREATE TABLE user_action_result(
  window_start TIMESTAMP(3),
  cnt bigint
) WITH (
  'connector' = 'print'
)
				
-- 窗口计算
INSERT INTO user_action_result
select * from (
	SELECT TUMBLE_START(user_action_time, INTERVAL '10' SECOND) window_start, COUNT(disTINCT user_name) cnt
	FROM user_actions
	GROUP BY TUMBLE(user_action_time, INTERVAL '10' SECOND)
)

-- 测试数据
{"user_name":"zhangsan","data":"browse"}
{"user_name":"lisi","data":"browse"}

首先源表上，我们使用了 processing time，加载了字段 user_action_time 上，这并不是我们数据中的字段，而是程序自动给我们加上的，是一个虚拟字段作为时间属性。

然后是查询 sql， group by 后面的 TUMBLE(user_action_time, INTERVAL '10' SECOND)，表示这是一个滚动窗口，使用 user_action_time 作为时间字段，并且窗口大小为 INTERVAL '10' SECOND ，表示 10 s，就是刚刚讲到的 INTERVAL 的语法。

select 中的 TUMBLE_START(user_action_time, INTERVAL '10' SECOND) 是窗口的开始时间，COUNT(disTINCT user_name) 表示统计每个窗口中的 user_name 去重值。

具体程序见：

滚动窗口 - 使用 EventTime

首先仍然需要在执行环境中声明使用 EventTime：

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

修改一下源表的定义

CREATE TABLE user_actions (
	user_name string,
	data string,
	user_action_time TIMESTAMP(3),
	WATERMARK FOR user_action_time as user_action_time - INTERVAL '5' SECOND
   ) WITH (
	'connector' = 'kafka',
	'topic' = 'user_log',
	'scan.startup.mode' = 'latest-offset',
	'properties.bootstrap.servers' = '192.168.56.10:9092',
	'format' = 'json',
	'json.fail-on-missing-field' = 'false'
)

可以看到，有一个时间字段是 user_action_time，然后 使用 WATERMARK FOR user_action_time as user_action_time - INTERVAL '5' SECOND ，来表示把 user_action_time 作为时间字段，并且声明一个 5s 延迟的 watermark。只用一句 sql 就定义好了 event_time 和 水位。

具体程序可以去我的 github 上下载：

https://github.com/nicekk/Flink-Practice

五、总结

假设你之前没有接触过 Flink sql，看完本文相信你已经对 Flink sql 有了初步的认识，再打开 IDEA，亲自动手操作一遍就会有更加深刻的认识，这也就达到了本文的目的了。