Flink 必知必会经典课程5:Flink SQL _ Table 介绍与实战-阿里云开发者社区

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

Flink 必知必会经典课程5:Flink SQL _ Table 介绍与实战

简介: 本文由 Apache Flink PMC,阿里巴巴技术专家伍翀(云邪)分享,主要介绍了Flink SQL和Table API的诞生背景、概念和功能,并通过三个实例演练让观众更直观地了解了Flink及其在Kibana上的具体操作流程。

作者:伍翀(云邪)

摘要: 本文由 Apache Flink PMC,阿里巴巴技术专家伍翀(云邪)分享,主要介绍了Flink SQL和Table API的诞生背景、概念和功能,并通过三个实例演练让观众更直观地了解了Flink及其在Kibana上的具体操作流程。内容如下:

  1. Flink SQL和Table API诞生的背景;
  2. Flink SQL和Table API的核心概念及功能;
  3. 结合Demo进行实战演练

Flink有非常强大的API抽象能力,它提供了三层的API,从底至上分别是Process Function, DataStream API以及SQL和Table API。这三层都有不同的用户群体,越低层灵活度越高,门槛也会越高,最高层门槛较低,但是会牺牲一些灵活度。

图片 1.png

为什么要花精力做SQL和Table API?

DataStream API非常好用,因为它的表达能力非常强,用户可以维护和更新应用状态,而且它对时间的控制力也非常灵活。但相对而言,它的复杂度和门槛也更高,并不适用于所有人,很多用户希望专注于业务逻辑。所以,要提供更加简单易懂的API,SQL是目前最佳的选择。

Flink SQL和Table API优势有很多。首先它非常易于理解,很多不同行业不同领域的人都懂SQL,它已经成为大数据处理生态圈的标准语言了;其次SQL是声明式的语言,用户只需要表达想要什么,而无需关心如何计算;然后SQL是会自动优化的,能生成最优的执行计划;同时SQL还是30多年的语言,非常稳定;最后,SQL可以更容易地统一流和批,用同一套系统就能同时处理,让用户只关注最核心的业务逻辑。

图片 2.png

SQL和Table API 简介

Flink关系式的API主要暴露两种,一种是SQL的API,还有一种是Table的API。SQL API完全遵循ANSI SQL的标准设计,所以如果有SQL基础,它的学习门槛是比较低的,而Table可以理解为类SQL的编程式的API。他们都是统一的批处理和流处理的API,不管输入是静态的批处理数据,还是无限的流处理数据,他的查询的结果都是相同的。总结而言,就是一份代码,一个结果,这也是流批统一的最重要的评价指标。

图片 3.png

Flink的工作流程

下面是比较高级的概览图,SQL和Table在进入Flink以后会转化成统一的数据结构表达形式,即Logical Plan。其中,Catalog会提供一些原数据信息,用于后续的优化。Logical Plan是优化的路口,经过一系列的优化规则后,Flink会把初始的Logical Plan优化为Physical Plan,并通过Code Generation机制翻译为Transformation,最后转换成JobGraph,用于提交到 Flink的集群做分布式的执行。可以看到,整个流程并没有单独的流处理和批处理的路径,因为这些优化的过程和扩建都是共享的。

图片 4.png

用实例理解流和批

比如一个点击的文件,有user、点击的时间和URL。如果我们要统计点击的次数,在选出user做统一的批处理的情况下,它的特点就是一次性读入和一次性输出。

图片 5.png

而如果Click是一个数据流,在这种情况下,输入一条数据后就能输出一个结果,比如Marry第一次点击会记录一次,第二次点击就会做增量计算。所以输入数据会持续读入,结果也会持续被更新。

可以看到,这里流和批的结果是一样的,所以我们可以把以前批处理的SQL迁移到Flink上做流处理,它的结果和语义应该和之前的批处理是一样的。

图片 6.png

Flink SQL和Table应用案例

典型的包括低延迟ETL处理,比如数据的预处理、清洗和过滤;还有数据管道,Flink可以做实时和离线的数据管道,可以构建低延时实时数仓,也可以实时数据同步,把数据从某一个数据系统同步到另一个数据系统;

第三种是流式和批式的数据分析,去计算和更新离线或实时的数据,并进行可视化,典型的比如阿里双11的大屏;

最后一种是模式识别,也就是实时地识别数据流中符合某种pattern的事件流,然后做相应的监控或者报警的服务,比如网约车的一些异常事件的监控服务。

图片 7.png

Flink的核心功能

下图包含了Flink的一些核心功能。第一个是SQL的DDL,DDL直接对接外部系统,它的强弱决定了Flink与外部系统的联通性,而作为一个计算引擎,与外部数据存储的联通性非常重要;第二是完整的类型系统,它支持多种数据类型,这对SQL引擎而言也是非常必要的;第三是高效流式TopN,有非常强大的流处理能力,用来实时计算排行旁,比如双11的销量排行榜;还有高效的流式去重对数据进行过滤,因为有时采集会包含重复的数据;还有维表关联、对接CDC等。

除此之外,Flink还有非常多的内置函数,支持MiniBatch,以及有多种解热点手段。它还支持完整的批处理,适用Python等语言,还有Hive的集成等功能,不仅能直接访问Hive的数据,还兼容了Hive的语法,让用户不必再频繁切换。

图片 8.png

示例

下面是一个电商的用户行为的实时分析。我们从Kafka中实时地消费用户的行为数据,并与MySQL中的数据进行关联,然后写入Elasticsearch的索引中,并用Kibana进行视觉化呈现。这是一个端到端的实时应用的构建。

image.png

下面是在Kibana上的最终展示成果,会有面板进行实时监控,显示出包括当前的独立用户数、类目排行、各时段购买量等数据。

image.png

下面是来自某宝的用户行为日志,我们只选取了11月27日当天的行为,它包含这几个字段,包括用户ID、商品ID、商品类目ID、行为类型和时间戳。行为类型中,pv代表点击,buy代表购买,cart代表加入购物车,fav代表收藏事件,而时间戳代表事件发生的时间。

image.png

实战演练

演练的示例代码已经传到了Github,大家如果有兴趣也可以按照这个文档一步一步做下去。我们准备一台装有Docker的Linux或者MacOS计算机即可,不需要下载额外的包。

首先,我们新建一个目录,比如叫flink-sql-demo,然后把docker-compose的demo文件下载下来,可以点进去看一下这个文件。

image.png

这里面有个dategen的数据源,我们可以去控制它的产生速度,比如把产生的速度从2000改成3000。

image.png

我们通过docker-compose up-d把docker中的容器都启动起来。容器包括Jobmanager、Taskmanager这两个Flink的集群,还有Kibana、Elasticsearch、Zookeeper、MySQL、Kafka等。

image.png

我们可以用Docker-compose的命令看一下Kafka中最新的10条数据。它有user ID,有商品ID,有类目ID,有用户的行为,还有一个TS代表这个行为当时发生的时间。

image.png

随后我们启动今天的主角,通过Docker-compose启动SQL-Client容器,当看到这个松鼠的时候,SQL Client就成功启动了,我们可以在里面运行SQL的命令。

image.png

第一步我们要用DDL创建数据源,把用户日志的数据源先创建起来。我们用Create Table这个DDL语法创建了一个user behavior的表,它里面有5个字段,包括user ID,商品ID,类目ID,用户行为和TS时间戳。With里面跟的是一些如何连接到外部系统的属性,比如用Kafka连接外部的topic。

image.png

另外我们也能通过show table看user behavior,用describe table看表的结构、字段、计算列、watermark策略等等。

image.png

我们也可以进到8081端口,这是Docker compose旗下Flink集群的一个Web UI界面,这里面各个栏目大家都可以去看看。

image.png

接下来我们用3个实战去画一些图表,深入了解Flink的一些功能。

首先是统计每小时的成交量。我们先用DDL创建Elasticsearch表,定义每小时的成交量,随后提交Query去做每小时成交量的统计分析。

我们需要做每小时的一个滑窗,用到Tumble Window语法。Tumble的第一个字段定义时间属性,也就是刚刚说的TS事件时间,然后开窗大小是一小时,也就是说我们每小时会滑一个窗口,然后对窗口内的数据做统计分析。

image.png

我们提交这个Query,然后通过5601端口访问Kibana对它进行可视化。刚进来的时候是空的,里面什么数据都没有,所以我们一般要先创建 create index pattern,通过页面的Management中的Index Pattern进入,找到我们的索引,点击进去创建。

image.png

创建了Index Pattern以后,我们才能在当中做一些Discovery或者可视化。可以看到,这些字段就是我们刚才DDL里定义的,并且它还有对应的值。

image.png

当然我们最终是要进行可视化,所以我们要创建一个Dashboard。在页面左上角点击Dashboard,然后点击Create New创建新的视图,随后就可以设置每小时的成交量了。

image.png

我们画一个面积图,在Y轴上选择购买量max,标签名字改为“成交量”,然后因为X轴展示时间,所以选择“hour-of-day”,order by字母序,改为24,随后点击播放键,面积图就画好了。我们也可以点击保存,这样这个面积图就会保存到Dashboard上。

image.png

随后我们再画一个图,统计一天每10分钟累计独立用户数。同样,我们需要现在SQL CLI中创建一个Elasticsearch表,用于存储结果汇总数据,字符段包括日期时间和累计uv数。

image.png

然后,我们在SQL CLI中执行Table。

image.png

这里面Query主要做了一件事,就是把日期和时间选出来。这里唯一有一点特殊的是,因为我们的需求是做每10分钟的点,所以用Substr的两个竖线做连接的函数来实现。随后,我们像之前一样把Query提交到SQL CLI里面运行。

image.png

和之前一样,我们创建新的视图,这里我们创建的是一个连线图。

image.png

我们在Y轴取uv的值,命名为“独立访问用户数”,X轴选terms,然后选time-str,order by Alphabetical,一天中改为150个点。然后点播放,独立用户数的曲线图就出现了。同样,我们可以点击保存,把这个图加到Dashboard上。

image.png

随后我们来画第三张图。第三张图是顶级类目排行榜,因为一个商品对应的类目太细分,比如对应非常细的第三四级类目,所以它对排行榜意义可能不大。但是我们希望归约到一个顶级类目做统计分析,所以开始之前准备了MySQL的容器,里面准备了子类目和顶级类目的映射关系。

我们首先在SQL CLI中创建MySQL表,后续用作维表查询,同时再创建一个Elasticsearch表,用于存储类目统计结果。而在Query这里,我们会用到Create View语法去注册一个临时的视图,简化写法,因为把两个Query写在一起可能会比较复杂。

image.png

我们同样在SQL CLI中运行代码,然后进入到Kibana页面建立索引和添加可视化图表。这里我们使用Horizontal Bar去画一个柱形图。

image.png

在Y轴上,我们统计类目的成交量,X轴用类目的名字,排序用倒序排列,随后点击播放键。

image.png

最后,我们同样点击保存,把类目排行榜添加到Dashboard上。加上之前我们做的两个,Dashboard上就有3个图表了,这里可以自己拖拽图表美化一下。

image.png

以上就是今天的课程,大家也可以到Github上的文档中再去学习和实践。

活动推荐:

仅需99元即可体验阿里云基于 Apache Flink 构建的企业级产品-实时计算 Flink 版!点击下方链接了解活动详情:https://www.aliyun.com/product/bigdata/sc?utm_content=g_1000250506

社区二维码.jpg

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

一套基于Apache Flink构建的一站式、高性能实时大数据处理平台,广泛适用于流式数据处理、离线数据处理、DataLake计算等场景。

官方博客
链接