Flink入坑指南 第三章:第一个作业-阿里云开发者社区

开发者社区> 阿里云实时计算Flink> 正文
登录阅读全文

Flink入坑指南 第三章:第一个作业

简介:

摘要: 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

开始写第一个作业

原始数据

ctime category_id shop_id item_id price
2018-12-04 15:44:54 cat_01 shop_01 item_01 10
2018-12-04 15:45:46 cat_02 shop_02 item_02 11.1
2018-12-04 15:46:11 cat_01 shop_03 item_03 12.4

主要逻辑

源表:

-- 源表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') ; --按照天做聚合

结果:

cdate gmv_daily
20181204 33.5

特点:

  • 每次执行前,数据库中都保存了当天已经入库的全量数据
  • 每次执行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结果表中没有加主键,那看到的结果如下:
cdate gmv_daily
20181204 10.0
20181204 21.1
20181204 33.5

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

cdate gmv_daily
20181204 33.5

原理介绍

这个例子中,批和流的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的时间,如下所示:
ctime category_id shop_id item_id price ptime
2018-12-04 15:44:54 cat_01 shop_01 item_01 10 2018-12-04 15:45:00
2018-12-04 15:45:46 cat_02 shop_02 item_02 11.1 2018-12-04 15:45:10
2018-12-04 15:46:11 cat_01 shop_03 item_03 12.4 2018-12-04 15:52:00

此时:

  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构建的一站式、高性能实时大数据处理平台,广泛适用于流式数据处理、离线数据处理、DataLake计算等场景。

官方博客
链接