T O P

[资源分享]     Flink SQL 核心概念剖析与编程案例实战

  • By - 楼主

  • 2021-01-08 09:41:28
  • 本次,我们从 0 开始逐步剖析 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,亲自动手操作一遍就会有更加深刻的认识,这也就达到了本文的目的了。

    如果觉得有收获,可以关注我的公众号:「KK架构师」

    本帖子中包含资源

    您需要 登录 才可以下载,没有帐号?立即注册