Flink SQL

Flink SQL/Table 如何转化为Flink graph?
Blink 进行了什么优化?

本节将主要从 SQL/Table API 如何转化为真正的 Job Graph 的流程开始，让大家对 Blink Planner 有一个比较清晰的认识，希望对大家阅读 Blink 代码，或者使用 Blink 方面有所帮助。然后介绍 Blink Planner 的改进及优化。

从上图可以很清楚的看到，解析的过程涉及到了三层：Table API/SQL，Blink Planner，Runtime，下面将对主要的步骤进行讲解：

Table API&SQL 解析验证：在 Flink 1.9 中，Table API 进行了大量的重构，引入了一套新的 Operation，这套 Operation 主要是用来描述任务的 Logic Tree。

当 SQL 传输进来后，首先会去做 SQL 解析，SQL 解析完成之后，会得到 SqlNode Tree(抽象语法树)，然后会紧接着去做 Validator（验证），验证时会去访问 FunctionManger 和 CatalogManger，FunctionManger 主要是查询用户定义的 UDF，以及检查 UDF 是否合法，CatalogManger 主要是检查这个 Table 或者 Database 是否存在，如果验证都通过，就会生成一个 Operation DAG（有向无环图）。

从这一步可以看出，Table API 和 SQL 在 Flink 中最终都会转化为统一的结构，即 Operation DAG。

生成RelNode：Operation DAG 会被转化为 RelNode(关系表达式) DAG。

优化：优化器会对 RelNode 做各种优化，优化器的输入是各种优化的规则，以及各种统计信息。当前，在 Blink Planner 里面，绝大部分的优化规则，Stream 和 Batch 是共享的。差异在于，对 Batch 而言，它没有 state 的概念，而对于 Stream 而言，它是不支持 sort 的，所以目前 Blink Planner 中，还是运行了两套独立的规则集（Rule Set），然后定义了两套独立的 Physical Rel：BatchPhysical Rel 和 StreamPhysical Rel。优化器优化的结果，就是具体的 Physical Rel DAG。

转化：得到 Physical Rel Dag 后，继续会转化为 ExecNode，通过名字可以看出，ExecNode 已经属于执行层的概念了，但是这个执行层是 Blink 的执行层，在 ExecNode 中，会进行大量的 CodeGen 的操作，还有非 Code 的 Operator 操作，最后，将 ExecNode 转化为 Transformation DAG。

**生成可执行 Job Graph：**得到 Transformation DAG 后，最终会被转化成 Job Graph，完成 SQL 或者 Table API 的解析。

Blink Planner 改进及优化

Blink Planner 功能方面改进主要包含如下几个方面：

更完整的 SQL 语法支持：例如，IN，EXISTS，NOT EXISTS，子查询，完整的 Over 语句，Group Sets 等。而且已经跑通了所有的 TPCH，TPCDS 这两个测试集，性能还非常不错。
提供了更丰富，高效的算子。
提供了非常完善的 cost 模型，同时能够对接 Catalog 中的统计信息，使 cost 根据统计信息得到更优的执行计划。
支持 join reorder。
shuffle service：对 Batch 而言，Blink Planner 还支持 shuffle service，这对 Batch 作业的稳定性有非常大的帮助，如果遇到 Batch 作业失败，通过 shuffle service 能够很快的进行恢复。

性能方面，主要包括以下部分：

分段优化。
Sub-Plan Reuse。
更丰富的优化 Rule：共一百多个 Rule ，并且绝大多数 Rule 是 Stream 和 Batch 共享的。
更高效的数据结构 BinaryRow：能够节省序列化和反序列化的操作。
mini-batch 支持（仅 Stream）：节省 state 的访问的操作。
节省多余的 Shuffle 和 Sort（Batch 模式）：两个算子之间，如果已经按 A 做 Shuffle，紧接着他下的下游也是需要按 A Shuffle 的数据，那中间的这一层 Shuffle，就可以省略，这样就可以省很多网络的开销，Sort 的情况也是类似。Sort 和 Shuffle 如果在整个计算里面是占大头，对整个性能是有很大的提升的。

深入性能优化及实践

本节中，将使用具体的示例进行讲解，让你深入理解 Blink Planner 性能优化的设计。

分段优化

示例 5

1
2
3

create view MyView as select word, count(1) as freq from SourceTable group by word; insert into SinkTable1 select \* from MyView where freq >10;

insert into SinkTable2 select count(word) as freq2, freq from MyView group by freq;

上面的这几个 SQL，转化为 RelNode DAG，大致图形如下：

图5 示例5 RelNode DAG

如果是使用 Flink Planner，经过优化层后，会生成如下执行层的 DAG：

图6 示例 5 Flink Planner DAG

可以看到，Flink Planner 只是简单的从 Sink 出发，反向的遍历到 Source，从而形成两个独立的执行链路，从上图也可以清楚的看到，Scan 和第一层 Aggregate 是有重复计算的。

在 Blink Planner 中，经过优化层之后，会生成如下执行层的 DAG：

图7 示例 5 Blink Planner DAG

Blink Planner 不是在每次调用 insert into 的时候就开始优化，而是先将所有的 insert into 操作缓存起来，等到执行前才进行优化，这样就可以看到完整的执行图，可以知道哪些部分是重复计算的。Blink Planner 通过寻找可以优化的最大公共子图，找到这些重复计算的部分。经过优化后，Blink Planner 会将最大公共子图的部分当做一个临时表，供其他部分直接使用。

这样，上面的图可以分为三部分，最大公共子图部分（临时表），临时表与 Filter 和 SinkTable1 优化，临时表与第二个 Aggregate 和 SinkTable 2 优化。

Blink Planner 其实是通过声明的 View 找到最大公共子图的，因此在开发过程中，如果需要复用某段逻辑，就将其定义为 View，这样就可以充分利用 Blink Planner 的分段优化功能，减少重复计算。

当然，当前的优化也不是最完美的，因为提前对图进行了切割，可能会导致一些优化丢失，今后会持续地对这部分算法进行改进。

总结一下，Blink Planner 的分段优化，其实解的是多 Sink 优化问题（DAG 优化），单 Sink 不是分段优化关心的问题，单 Sink 可以在所有节点上优化，不需要分段。

Sub-Plan Reuse

示例 6

insert into SinkTabl

select freq from (select word, count(1) as freq from SourceTable group by word) t where word like 'T%'

union all

select count(word) as freq2 from (select word, count(1) as freq from SourceTable group by word) t group by freq;

这个示例的 SQL 和分段优化的 SQL 其实是类似的，不同的是，没有将结果 Sink 到两个 Table 里面，而是将结果 Union 起来，Sink 到一个结果表里面。

下面看一下转化为 RelNode 的 DAG 图：

图 8 示例 6 RelNode DAG

从上图可以看出，Scan 和第一层的 Aggregate 也是有重复计算的，Blink Planner 其实也会将其找出来，变成下面的图：

图9 示例 6 Blink Planner DAG

Sub-Plan 优化的启用，有两个相关的配置：

table.optimizer.reuse-sub-plan-enabled （默认开启）
table.optimizer.reuse-source-enabled（默认开启）

这两个配置，默认都是开启的，用户可以根据自己的需求进行关闭。这里主要说明一下 table.optimizer.reuse-source-enabled 这个参数。在 Batch 模式下，join 操作可能会导致死锁，具体场景是在执行 hash-join 或者 nested-loop-join 时一定是先读 build 端，然后再读 probe 端，如果启用 reuse-source-enabled，当数据源是同一个 Source 的时候，Source 的数据会同时发送给 build 和 probe 端。这时候，build 端的数据将不会被消费，导致 join 操作无法完成，整个 join 就被卡住了。

为了解决死锁问题，Blink Planner 会先将 probe 端的数据落盘，这样 build 端读数据的操作才会正常，等 build 端的数据全部读完之后，再从磁盘中拉取 probe 端的数据，从而解决死锁问题。但是，落盘会有额外的开销，会多一次写的操作；有时候，读两次 Source 的开销，可能比一次写的操作更快，这时候，可以关闭 reuse-source，性能会更好。当然，如果读两次 Source 的开销，远大于一次落盘的开销，可以保持 reuse-source 开启。需要说明的是，Stream 模式是不存在死锁问题的，因为 Stream 模式 join 不会有选边的问题。

总结而言，sub-plan reuse 解的问题是优化结果的子图复用问题，它和分段优化类似，但他们是一个互补的过程。

注：Hash Join：对于两张待 join 的表 t1, t2。选取其中的一张表按照 join 条件给的列建立hash 表。然后扫描另外一张表，一行一行去建好的 hash 表判断是否有对应相等的行来完成 join 操作，这个操作称之为 probe (探测)。前一张表叫做 build 表，后一张表的叫做 probe 表。

Agg 分类优化

Blink 中的 Aggregate 操作是非常丰富的：

group agg，例如：select count(a) from t group by b
over agg，例如：select count(a) over (partition by b order by c) from t
window agg，例如：select count(a) from t group by tumble(ts, interval ‘10’ second), b
table agg ，例如：tEnv.scan(‘t’).groupBy(‘a’).flatAggregate(flatAggFunc(‘b’ as (‘c’, ‘d’)))

下面主要对 Group Agg 优化进行讲解，主要是两类优化。

1. Local/Global Agg 优化

Local/Global Agg 主要是为了减少网络 Shuffle。要运用 Local/Global 的优化，必要条件如下：

Aggregate 的所有 Agg Function 都是 mergeable 的，每个 Aggregate 需要实现 merge 方法，例如 SUM，COUNT，AVG，这些都是可以分多阶段完成，最终将结果合并；但是求中位数，计算 95% 这种类似的问题，无法拆分为多阶段，因此，无法运用 Local/Global 的优化。
table.optimizer.agg-phase-strategy 设置为 AUTO 或者 TWO_PHASE。
Stream 模式下，mini-batch 开启；Batch 模式下 AUTO 会根据 cost 模型加上统计数据，选择是否进行 Local/Global 优化。

示例 7

select count(*) from t group by color

没有优化的情况下，下面的这个 Aggregate 会产生 10 次的 Shuffle 操作。

图 10 示例 7 未做优化的 Count 操作

使用 Local/Global 优化后，会转化为下面的操作，会在本地先进行聚合，然后再进行 Shuffle 操作，整个 Shuffle 的数据剩下 6 条。在 Stream 模式下，Blink 其实会以 mini-batch 的维度对结果进行预聚合，然后将结果发送给 Global Agg 进行汇总。

图 11 示例 7 经过 Local/Global 优化的 Count 操作

2. Distinct Agg 优化

Distinct Agg 进行优化，主要是对 SQL 语句进行改写，达到优化的目的。但 Batch 模式和 Stream 模式解决的问题是不同的：

Batch 模式下的 Distinct Agg，需要先做 Distinct，再做 Agg，逻辑上需要两步才能实现，直接实现 Distinct Agg 开销太大。
Stream 模式下，主要是解决热点问题，因为 Stream 需要将所有的输入数据放在 State 里面，如果数据有热点，State 操作会很频繁，这将影响性能。

Batch 模式

第一层，求 distinct 的值和非 distinct agg function 的值，第二层求 distinct agg function 的值

示例 8

select color, count(distinct id), count(*) from t group by color

手工改写成：

select color, count(id), min(cnt) from (

select color, id, count(*) filter (where $e=2) as cnt from (

select color, id, 1 as $e from t --for distinct id

union all

select color, null as id, 2 as $e from t -- for count(\*)

) group by color, id, $e

) group by color

转化的逻辑过程，如下图所示：

图 12 示例 8 Batch 模式 Distinct 改写逻辑

Stream 模式

Stream 模式的启用有一些必要条件：

必须是支持的 agg function：avg/count/min/max/sum/first_value/concat_agg/single_value；
table.optimizer.distinct-agg.split.enabled（默认关闭）

示例 9

select color, count(distinct id), count(*) from t group by color

手工改写成：

select color, sum(dcnt), sum(cnt) from (

select color, count(distinct id) as dcnt, count(*) as cnt from t

group by color, mod(hash_code(id), 1024)

) group by color

改写前，逻辑图大概如下：

图 13 示例 9 Stream 模式未优化 Distinct

改写后，逻辑图就会变为下面这样，热点数据被打散到多个中间节点上。

图14 示例 9 Stream 模式优化 Distinct

需要注意的是，示例 5 的 SQL 中 mod(hash_code(id),1024)中的这个 1024 为打散的维度，这个值建议设置大一些，设置太小产生的效果可能不好。

总结

本文首先对新的 TableEnvironment 的整体设计进行了介绍，并且列举了各种模式下TableEnvironment 的选择，然后通过具体的示例，展示了各种模式下代码的写法，以及需要注意的事项。

在新的 Catalog 和 DDL 部分，对 Catalog 的整体设计、DDL 的使用部分也都以实例进行拆分讲解。最后，对 Blink Planner 解析 SQL/Table API 的流程、Blink Planner 的改进以及优化的原理进行了讲解，希望对大家探索和使用 Flink SQL 有所帮助。

SQL解析工具

hive使用了antlr3实现了自己的HQL,
Flink使用Apache Calcite,
而Calcite的解析器是使用JavaCC实现的,
Spark2.x以后采用了antlr4实现自己的解析器,
Presto也是使用antlr4。

Window

在Apache Flink中有2种类型的Window，一种是OverWindow，即传统数据库的标准开窗，每一个元素都对应一个窗口。一种是GroupWindow，目前在SQL中GroupWindow都是基于时间进行窗口划分的。

Over Window

Apache Flink中对OVER Window的定义遵循标准SQL的定义语法。
按ROWS和RANGE分类是传统数据库的标准分类方法，在Apache Flink中还可以根据时间类型(ProcTime/EventTime)和窗口的有限和无限(Bounded/UnBounded)进行分类，共计8种类型。为了避免大家对过细分类造成困扰，我们按照确定当前行的不同方式将OVER Window分成两大类进行介绍，如下:

ROWS OVER Window - 每一行元素都视为新的计算行，即，每一行都是一个新的窗口。
RANGE OVER Window - 具有相同时间值的所有元素行视为同一计算行，即，具有相同时间值的所有行都是同一个窗口。

Bounded ROWS OVER Window

Bounded ROWS OVER Window 每一行元素都视为新的计算行，即，每一行都是一个新的窗口。

语义

我们以3个元素(2 PRECEDING)的窗口为例，如下图:

上图所示窗口 user 1 的 w5和w6， user 2的窗口 w2 和 w3，虽然有元素都是同一时刻到达，但是他们仍然是在不同的窗口，这一点有别于RANGE OVER Window。

语法

Bounded ROWS OVER Window 语法如下：

SELECT 
    agg1(col1) OVER(
     [PARTITION BY (value_expression1,..., value_expressionN)] 
     ORDER BY timeCol
     ROWS 
     BETWEEN (UNBOUNDED | rowCount) PRECEDING AND CURRENT ROW) AS colName, 
... 
FROM Tab1

value_expression - 进行分区的字表达式；
timeCol - 用于元素排序的时间字段；
rowCount - 是定义根据当前行开始向前追溯几行元素。

SQL 示例

利用item_tab测试数据，我们统计同类商品中当前和当前商品之前2个商品中的最高价格。

SELECT  
    itemID,
    itemType, 
    onSellTime, 
    price,  
    MAX(price) OVER (
        PARTITION BY itemType 
        ORDER BY onSellTime 
        ROWS BETWEEN 2 preceding AND CURRENT ROW) AS maxPrice
  FROM item_tab

Result

itemID	itemType	onSellTime	price	maxPrice
ITEM001	Electronic	2017-11-11 10:01:00	20	20
ITEM002	Electronic	2017-11-11 10:02:00	50	50
ITEM003	Electronic	2017-11-11 10:03:00	30	50
ITEM004	Electronic	2017-11-11 10:03:00	60	60
ITEM005	Electronic	2017-11-11 10:05:00	40	60
ITEM006	Electronic	2017-11-11 10:06:00	20	60
ITEM007	Electronic	2017-11-11 10:07:00	70	70
ITEM008	Clothes	2017-11-11 10:08:00	20	20

Bounded RANGE OVER Window

Bounded RANGE OVER Window 具有相同时间值的所有元素行视为同一计算行，即，具有相同时间值的所有行都是同一个窗口。

语义

我们以3秒中数据(INTERVAL ‘2’ SECOND)的窗口为例，如下图：

注意: 上图所示窗口 user 1 的 w6， user 2的窗口 w3，元素都是同一时刻到达,他们是在同一个窗口，这一点有别于ROWS OVER Window。

语法

Bounded RANGE OVER Window的语法如下：

SELECT 
    agg1(col1) OVER(
     [PARTITION BY (value_expression1,..., value_expressionN)] 
     ORDER BY timeCol
     RANGE 
     BETWEEN (UNBOUNDED | timeInterval) PRECEDING AND CURRENT ROW) AS colName, 
... 
FROM Tab1

value_expression - 进行分区的字表达式；
timeCol - 用于元素排序的时间字段；
timeInterval - 是定义根据当前行开始向前追溯指定时间的元素行；

SQL 示例

我们统计同类商品中当前和当前商品之前2分钟商品中的最高价格。

SELECT  
    itemID,
    itemType, 
    onSellTime, 
    price,  
    MAX(price) OVER (
        PARTITION BY itemType 
        ORDER BY rowtime 
        RANGE BETWEEN INTERVAL '2' MINUTE preceding AND CURRENT ROW) AS maxPrice
  FROM item_tab

Result（Bounded RANGE OVER Window）

itemID	itemType	onSellTime	price	maxPrice
ITEM001	Electronic	2017-11-11 10:01:00	20	20
ITEM002	Electronic	2017-11-11 10:02:00	50	50
ITEM003	Electronic	2017-11-11 10:03:00	30	60
ITEM004	Electronic	2017-11-11 10:03:00	60	60
ITEM005	Electronic	2017-11-11 10:05:00	40	60
ITEM006	Electronic	2017-11-11 10:06:00	20	40
ITEM007	Electronic	2017-11-11 10:07:00	70	70
ITEM008	Clothes	2017-11-11 10:08:00	20	20

特别说明

OverWindow最重要是要理解每一行数据都确定一个窗口，同时目前在Apache Flink中只支持按时间字段排序。并且OverWindow开窗与GroupBy方式数据分组最大的不同在于，GroupBy数据分组统计时候，在SELECT中除了GROUP BY的key，不能直接选择其他非key的字段，但是OverWindow没有这个限制，SELECT可以选择任何字段。比如一张表table(a,b,c,d)4个字段，如果按d分组求c的最大值，两种写完如下:

GROUP BY - SELECT d, MAX(c) FROM table GROUP BY d
OVER Window = SELECT a, b, c, d, MAX(c) OVER(PARTITION BY d, ORDER BY ProcTime())
如上 OVER Window 虽然PARTITION BY d,但SELECT 中仍然可以选择 a,b,c字段。但在GROUPBY中，SELECT 只能选择 d 字段。

Group Window

根据窗口数据划分的不同，目前Apache Flink有如下3种Bounded Winodw:

Tumble - 滚动窗口，窗口数据有固定的大小，窗口数据无叠加；
Hop - 滑动窗口，窗口数据有固定大小，并且有固定的窗口重建频率，窗口数据有叠加；
Session - 会话窗口，窗口数据没有固定的大小，根据窗口数据活跃程度划分窗口，窗口数据无叠加。

说明： Aapche Flink 还支持UnBounded的 Group Window，也就是全局Window，流上所有数据都在一个窗口里面，语义非常简单，这里不做详细介绍了。

Tumble

语义

Tumble 滚动窗口有固定size，窗口数据不重叠,具体语义如下：

语法

Tumble 滚动窗口对应的语法如下：

SELECT 
    [gk],
    [TUMBLE_START(timeCol, size)], 
    [TUMBLE_END(timeCol, size)], 
    agg1(col1), 
    ... 
    aggn(colN)
FROM Tab1
GROUP BY [gk], TUMBLE(timeCol, size)

[gk] - 决定了流是Keyed还是/Non-Keyed;
TUMBLE_START - 窗口开始时间;
TUMBLE_END - 窗口结束时间;
timeCol - 是流表中表示时间字段；
size - 表示窗口的大小，如秒，分钟，小时，天。

SQL 示例

利用pageAccess_tab测试数据，我们需要按不同地域统计每2分钟的淘宝首页的访问量(PV)。

SELECT  
    region,
    TUMBLE_START(rowtime, INTERVAL '2' MINUTE) AS winStart,  
    TUMBLE_END(rowtime, INTERVAL '2' MINUTE) AS winEnd,  
    COUNT(region) AS pv
FROM pageAccess_tab 
GROUP BY region, TUMBLE(rowtime, INTERVAL '2' MINUTE)

Result

region	winStart	winEnd	pv
BeiJing	2017-11-11 02:00:00.0	2017-11-11 02:02:00.0	1
BeiJing	2017-11-11 02:10:00.0	2017-11-11 02:12:00.0	2
ShangHai	2017-11-11 02:00:00.0	2017-11-11 02:02:00.0	1
ShangHai	2017-11-11 04:10:00.0	2017-11-11 04:12:00.0	1

Hop

Hop 滑动窗口和滚动窗口类似，窗口有固定的size，与滚动窗口不同的是滑动窗口可以通过slide参数控制滑动窗口的新建频率。因此当slide值小于窗口size的值的时候多个滑动窗口会重叠。

语义

Hop 滑动窗口语义如下所示：

语法

Hop 滑动窗口对应语法如下：

SELECT 
    [gk], 
    [HOP_START(timeCol, slide, size)] ,  
    [HOP_END(timeCol, slide, size)],
    agg1(col1), 
    ... 
    aggN(colN) 
FROM Tab1
GROUP BY [gk], HOP(timeCol, slide, size)

[gk] 决定了流是Keyed还是/Non-Keyed;
HOP_START - 窗口开始时间;
HOP_END - 窗口结束时间;
timeCol - 是流表中表示时间字段；
slide - 是滑动步伐的大小；
size - 是窗口的大小，如秒，分钟，小时，天；

SQL 示例

利用pageAccessCount_tab测试数据，我们需要每5分钟统计近10分钟的页面访问量(PV).

SELECT  
  HOP_START(rowtime, INTERVAL '5' MINUTE, INTERVAL '10' MINUTE) AS winStart,  
  HOP_END(rowtime, INTERVAL '5' MINUTE, INTERVAL '10' MINUTE) AS winEnd,  
  SUM(accessCount) AS accessCount  
FROM pageAccessCount_tab 
GROUP BY HOP(rowtime, INTERVAL '5' MINUTE, INTERVAL '10' MINUTE)

Result

winStart	winEnd	accessCount
2017-11-11 01:55:00.0	2017-11-11 02:05:00.0	186
2017-11-11 02:00:00.0	2017-11-11 02:10:00.0	396
2017-11-11 02:05:00.0	2017-11-11 02:15:00.0	243
2017-11-11 02:10:00.0	2017-11-11 02:20:00.0	33
2017-11-11 04:05:00.0	2017-11-11 04:15:00.0	129
2017-11-11 04:10:00.0	2017-11-11 04:20:00.0	129

Session

Seeeion 会话窗口是没有固定大小的窗口，通过session的活跃度分组元素。不同于滚动窗口和滑动窗口，会话窗口不重叠,也没有固定的起止时间。一个会话窗口在一段时间内没有接收到元素时，即当出现非活跃间隙时关闭。一个会话窗口分配器通过配置session gap来指定非活跃周期的时长.

语义

Session 会话窗口语义如下所示：

语法

Seeeion 会话窗口对应语法如下：

SELECT 
    [gk], 
    SESSION_START(timeCol, gap) AS winStart,  
    SESSION_END(timeCol, gap) AS winEnd,
    agg1(col1),
     ... 
    aggn(colN)
FROM Tab1
GROUP BY [gk], SESSION(timeCol, gap)

[gk] 决定了流是Keyed还是/Non-Keyed;
SESSION_START - 窗口开始时间；
SESSION_END - 窗口结束时间；
timeCol - 是流表中表示时间字段；
gap - 是窗口数据非活跃周期的时长；

SQL 示例

利用pageAccessSession_tab测试数据，我们按地域统计连续的两个访问用户之间的访问时间间隔不超过3分钟的的页面访问量(PV).

SELECT  
    region, 
    SESSION_START(rowtime, INTERVAL '3' MINUTE) AS winStart,  
    SESSION_END(rowtime, INTERVAL '3' MINUTE) AS winEnd, 
    COUNT(region) AS pv  
FROM pageAccessSession_tab
GROUP BY region, SESSION(rowtime, INTERVAL '3' MINUTE)

Result

region	winStart	winEnd	pv
BeiJing	2017-11-11 02:10:00.0	2017-11-11 02:13:00.0	1
ShangHai	2017-11-11 02:01:00.0	2017-11-11 02:08:00.0	4
ShangHai	2017-11-11 02:10:00.0	2017-11-11 02:14:00.0	2
ShangHai	2017-11-11 04:16:00.0	2017-11-11 04:19:00.0	1

UDX

Apache Flink 除了提供了大部分ANSI-SQL的核心算子，也为用户提供了自己编写业务代码的机会，那就是User-Defined Function,目前支持如下三种 User-Defined Function：

UDF - User-Defined Scalar Function
UDTF - User-Defined Table Function
UDAF - User-Defined Aggregate Funciton

UDX都是用户自定义的函数，那么Apache Flink框架为啥将自定义的函数分成三类呢？是根据什么划分的呢？Apache Flink对自定义函数进行分类的依据是根据函数语义的不同，函数的输入和输出不同来分类的，具体如下：

UDX	INPUT	OUTPUT	INPUT:OUTPUT
UDF	单行中的N(N>=0)列	单行中的1列	1:1
UDTF	单行中的N(N>=0)列	M(M>=0)行	1:N(N>=0)
UDAF	M(M>=0)行中的每行的N(N>=0)列	单行中的1列	M：1(M>=0)

UDF

定义
用户想自己编写一个字符串联接的UDF，我们只需要实现ScalarFunction#eval()方法即可，简单实现如下：

object MyConnect extends ScalarFunction {
  @varargs
  def eval(args: String*): String = {
    val sb = new StringBuilder
    var i = 0
    while (i < args.length) {
      if (args(i) == null) {
        return null
      }
      sb.append(args(i))
      i += 1
    }
    sb.toString
  }
}

使用

1
2
3

val fun = MyConnect
tEnv.registerFunction("myConnect", fun)
val sql = "SELECT myConnect(a, b) as str FROM tab"

UDTF

定义
用户想自己编写一个字符串切分的UDTF，我们只需要实现TableFunction#eval()方法即可，简单实现如下：

ScalarFunction#eval()`

class MySplit extends TableFunction[String] {
  def eval(str: String): Unit = {
    if (str.contains("#")){
      str.split("#").foreach(collect)
    }
  }

  def eval(str: String, prefix: String): Unit = {
    if (str.contains("#")) {
      str.split("#").foreach(s => collect(prefix + s))
    }
  }
}

使用

1
2
3

val fun = new MySplit()
tEnv.registerFunction("mySplit", fun)
val sql = "SELECT c, s FROM MyTable, LATERAL TABLE(mySplit(c)) AS T(s)"

UDAF

定义
UDAF 要实现的接口比较多，我们以一个简单的CountAGG为例，做简单实现如下：

class CountAccumulator extends JTuple1[Long] {
  f0 = 0L 
}


class MyCount
  extends AggregateFunction[JLong, CountAccumulator] {



  def accumulate(acc: CountAccumulator): Unit = {
    acc.f0 += 1L
  }



  def retract(acc: CountAccumulator): Unit = {
    acc.f0 -= 1L
  }

  def accumulate(acc: CountAccumulator, value: Any): Unit = {
    if (value != null) {
      acc.f0 += 1L
    }
  }

  def retract(acc: CountAccumulator, value: Any): Unit = {
    if (value != null) {
      acc.f0 -= 1L
    }
  }

  override def getValue(acc: CountAccumulator): JLong = {
    acc.f0
  }

  def merge(acc: CountAccumulator, its: JIterable[CountAccumulator]): Unit = {
    val iter = its.iterator()
    while (iter.hasNext) {
      acc.f0 += iter.next().f0
    }
  }

  override def createAccumulator(): CountAccumulator = {
    new CountAccumulator
  }

  def resetAccumulator(acc: CountAccumulator): Unit = {
    acc.f0 = 0L
  }

  override def getAccumulatorType: TypeInformation[CountAccumulator] = {
    new TupleTypeInfo(classOf[CountAccumulator], BasicTypeInfo.LONG_TYPE_INFO)
  }

  override def getResultType: TypeInformation[JLong] =
    BasicTypeInfo.LONG_TYPE_INFO
}

使用

1
2
3

val fun = new MyCount()
tEnv.registerFunction("myCount", fun)
val sql = "SELECT myCount(c) FROM MyTable GROUP BY  a"

上面我们介绍了Apache Flink SQL核心算子的语法及语义，这部分将选取Bounded EventTime Tumble Window为例为大家编写一个完整的包括Source和Sink定义的Apache Flink SQL Job。假设有一张淘宝页面访问表(PageAccess_tab)，有地域，用户ID和访问时间。我们需要按不同地域统计每2分钟的淘宝首页的访问量(PV). 具体数据如下：

region	userId	accessTime
ShangHai	U0010	2017-11-11 10:01:00
BeiJing	U1001	2017-11-11 10:01:00
BeiJing	U2032	2017-11-11 10:10:00
BeiJing	U1100	2017-11-11 10:11:00
ShangHai	U0011	2017-11-11 12:10:00

大家都知道，在 Flink 中，通过 Table API 和 SQL 实现的流处理逻辑，最终会翻译为基于 DataStreamAPI 实现的 DataStream 作业，返回这个作业输出的 DataStream (writeToSink 本质上也是先得到 DataStream 作业，再为其输出 DataStream 加上一个DataStreamSink) 。

从一段 SQL 到 DataStream 作业，其过程简单描述如下：

在 TableEnvironment，即“表环境”，将数据源注册为动态表。例如，通过表环境的接口`registerDataStream`, 作为源的DataStream，即数据流, 在表环境注册为动态表
通过表环境的接口 `sqlQuery`，将 SQL 构造为 Table 对象
通过toAppendStream/toRetractedStream接口，即翻译接口，将 Table 对象表达的作业逻辑，翻译为 DataStream 作业。

在调用翻译接口，将 Table 对象翻译为 DataStream 作业时，通过翻译接口传入的 TTL 配置，递归传递到各个计算节点的翻译、构造逻辑里，使得翻译出来的 DataStream 算子的内部状态按照该 TTL 配置及时清理。

【参考文献】