引言

前面群里面同学说对flink感兴趣，特别邀请资深流专家张如聪给大家深入分析下Flink里面最重要部分：Flink SQL。

本文对Flink SQL深入浅出，相当有深度的技术分析文章，希望大家会喜欢，对Flink技术上有疑问的也可以联系专家帮忙解答。

一、Flink SQL简介

Flink SQL 是Fllink提供的SQL的SDK API。SQL是比Table更高阶的API，集成在Table library中提供，在流和批上都可以用此API开发业务。本文主要侧重于SQL在Stream上的能力，也就是介绍StreamSQL的能力。

二、 StreamSQL能力概述

Flink SQL的语法采用Apache Calcite的语法.很多开源组件如Samza、Storm、Apex都使用Calcite的语法作为其SQL的语法。

在Flink 1.3.0版本中流上的操作支持SELECT, FROM, WHERE，UNION、聚合和自定义能力，join能力预计在Flink 1.4.0(预计在9月份发布)版本中提供。详细的语法能力，请参见第6章节。

三、用户使用StreamSQL开发业务流应用流程

在使用StreamSQL开发业务流应用前，需要在pom.xml增加引用flink-table lib的依赖，具体如下：

<dependency>

使用StreamSQL开发流应用的过程如下：

1、获取table stream 环境；

val env = StreamExecutionEnvironment.getExecutionEnvironment

2、从source算子获取流式数据，并定义数据的schema：

val orderA: DataStream[Order] = env.fromCollection(Seq(

其中order类的定义为：

case class Order(user: Long, product: String, amount: Int)

当前StreamSQL支持的source算子包含公共的kafka source、CSV file source，以及自定义的source。 3、用StreamSQL书写数据处理逻辑，即定义业务应用：

val result = tEnv.sql(

4、定义处理结果的输出，即sink算子：

result.toAppendStream[Order].print()

5、提交到Flink系统执行：

env.execute()

四、StreamSQL 执行原理介绍

如上图所示，StreamSQL API的执行原理如下：

1、用户使用对外提供Stream SQL的语法开发业务应用；

2、用calcite对StreamSQL进行语法检验，语法检验通过后，转换成calcite的逻辑树节点；最终形成calcite的逻辑计划；

3、采用Flink自定义的优化规则和calcite火山模型、启发式模型共同对逻辑树进行优化，生成最优的Flink物理计划；

4、对物理计划采用janino codegen生成代码，生成用低阶API DataStream 描述的流应用，提交到Flink平台执行；

五、StreamSQL 编译执行流程介绍

1、使用calcite对Sql进行编译，在编译的过程中对语法进行校验，如果语法符合要求，编译结果为calcite的一个逻辑树，如：第四章节中的SQL“SELECT * FROM OrderA WHERE amount > 2 + 3”编译后为：

LogicalFilter(condition=[>($2, 2+3)])

2、首先用如下规则集将table scan节点转化为关系表达式。

val TABLE_CONV_RULES: RuleSet = RuleSets.ofList(

3、再用如下规则集，将逻辑计划树标准化。

 val DATASET_NORM_RULES: RuleSet = RuleSets.ofList(

4、再使用如下规则，将逻辑树进行优化，生成最优的逻辑计划树.

val LOGICAL_OPT_RULES: RuleSet = RuleSets.ofList(

如逻辑计划

LogicalFilter(condition=[>($2, 2+3)])

优化为：

LogicalProject.NONE(input=rel#19:LogicalFilter.NONE(input=rel#10:LogicalTableScan.NONE(table=[OrderA]),condition=>($2, 5)),user=$0,product=$1,amount=$2)

主要做了如下几点优化： 1）过滤下压到tablescan节点，直接在输入的时候进行判断； 2）表达式提前预计算，如“2+3”之间计算出结果5； 5、再使用如下规则并使用janino codegen生成用DataStream API描述的物理执行计划；

val DATASTREAM_OPT_RULES: RuleSet = RuleSets.ofList(

6、最后使用如下规则对物理执行计划进行优化。主要是对聚合操作的优化。

val DATASTREAM_DECO_RULES: RuleSet = RuleSets.ofList(

7、最后提交到Flink平台执行；

六、StreamSQL 语法介绍

整体语法范式如下：

query:

语法操作

1）Scan、Projection、Filter以及自定义函数

对于自定义函数，需要注册到table 的stream环境中。如：

object TimestampModifier extends ScalarFunction {

2）Aggregations及用户自定义的聚合函数

说明： a）window当前支持group window和row window。group window就是所谓的跳跃窗口，固定周期触发输出。row window就是所谓的滑动窗口，每个数据流过来，都会触发输出；group window当前支持：

b）聚合自定义函数和普通自定义函数类似。 3）Set

StreamSQL支持的数据类型

StreamSQL支持的数据类型如下：

版权归微信公众号/头条号《大数据和云计算技术》所有，转载请联系本人授权；欢迎关注，获取最新技术资讯和深度分析。