关于flink:腾讯基于-Flink-SQL-的功能扩展与深度优化实践

整顿：戴季国（Flink 社区志愿者）

校对：苗文婷（Flink 社区志愿者）

摘要：本文由腾讯高级工程师杜立分享，次要介绍腾讯实时计算平台针对 Flink SQL 所做的优化，内容包含：

1. Flink SQL 现状
2. 窗口性能的扩大
3. 回撤流的优化
4. 将来的布局

一、背景及现状

1. 三种模式的剖析

Flink 作业目前有三种创立形式：JAR 模式、画布模式和 SQL 模式。不同的提交作业的形式针对的人群也是不一样的。

■ Jar 模式

Jar 模式基于 DataStream/DataSet API 开发，次要针对的是底层的开发人员。

• 长处：

· 性能灵便多变，因为它底层的 DataStream/DataSet API 是 Flink 的原生 API，你能够用它们开发任何你想要的算子性能或者 DAG 图；
· 性能优化不便，能够十分有针对性的去优化每一个算子的性能。

• 毛病：

· 依赖更新繁琐，无论扩大作业逻辑或是 Flink 版本的降级，都要去更新作业的代码以及依赖版本；
· 学习门槛较高。

■ 画布模式

所谓的画布模式，一般来讲会提供一个可视化的利落拽界面，让用户通过界面化的形式去进行利落拽操作，以实现 Flink 作业的编辑。它面向一些小白用户。

• 长处：

· 操作便捷，画布上能够很不便地定义 Flink 的作业所蕴含的各种算子；
· 性能较全，它基于 Table API 开发，性能笼罩比拟残缺；
· 易于了解，DAG 图比拟直观，用户可能非常容易的去了解整个作业的运行流程。

• 毛病：

· 配置简单：每一个算子都须要去一一的去配置，如果整个 DAG 图非常复杂，相应的配置工作也会十分大；
· 逻辑重用艰难：如果作业十分的多，不同的作业之间想去共享 DAG 逻辑的话十分艰难。

■ SQL 模式

SQL 语言曾经存在了很长时间了，它有本人的一套规范，次要面向数据分析人员。只有遵循既有的 SQL 规范，数据分析人员就能够在不同的平台和计算引擎之间进行切换。

• 长处：

· 清晰简洁，易于了解和浏览；
· 与计算引擎解耦，SQL 与计算引擎及其版本是解耦的，在不同的计算引擎之间迁徙业务逻辑不须要或极少须要去更改整段 SQL。同时，如果想降级 Flink 版本，也是不须要去更改 SQL；
· 逻辑重用不便，能够通过 create view 的形式去重用咱们的 SQL 逻辑。

• 毛病：

· 语法不对立，比如说流与维表 Join，Flink 1.9 之前应用 Lateral Table Join 语法，然而在 1.9 之后，更改成了 PERIOD FOR SYSTEM_TIME 语法，这种语法遵循了 SQL ANSI 2011 规范。语法的变动使得用户有肯定的学习老本；
· 性能笼罩不全：Flink SQL 这个模块存在的工夫不是很长，导致它的性能的一个笼罩不是很全。
· 性能调优艰难：一段 SQL 的执行效率次要由几个局部来决定，一个就是 SQL 自身所表白的业务逻辑；另一部分是翻译 SQL 所产生的执行打算的一个优化；第三局部的话，在产生最优的逻辑执行打算之后，翻译老本地的 native code 的时候计划也决定了 SQL 的执行效率；对于用户来讲的，他们所能优化的内容可能只局限于 SQL 所表白的业务逻辑。
· 问题定位艰难：SQL 是一个残缺的执行流程，如果咱们发现某些数据不对，想针对性地去排查到底是哪个算子出了问题，是比拟的艰难的。一般来讲，咱们想定位 Flink SQL 的问题，只能先一直的精简咱们的整个 SQL 逻辑，而后一直地去尝试输入，这个老本是十分高的。腾讯实时计算平台前期会针对这个问题，减少 trace 日志和 metrics 信息，输入到产品侧以帮忙用户定位 Flink SQL 应用上的问题。

2. 腾讯实时计算平台目前的工作

■ 扩大语法

定义了 window table-valued function 语法，以帮忙用户实现基于窗口的流 Join 和交并差操作。另外，实现了本人的流与维表 Join 的语法。

■ 新增性能

新增的一些性能，包含两个新的 Window 的类型，Incremental Window（增量窗口）和 Ehanced Tumble Window（加强窗口）。实现了 Eventtime Field 与 Table Source 的解耦，很多时候 Eventtime Field 并不能通过 Table Source 字段定义进去，比方 Table Source 是一个子查问或者某个工夫字段是由函数转换得出，想要用这些两头生成的工夫字段作为 Eventtime Field 目前是做不到的，咱们目前的计划是，让用户能够抉择物理表中任意的工夫字段来定义 Window 的工夫属性并输入 WaterMark。

■ 性能调优

• 回撤流优化；
• 内联 UDF，如果雷同的 UDF 既呈现在 LogicalProject 中，又呈现在 Where 条件中，那么 UDF 会进行屡次调用。将逻辑执行打算中反复调用的 UDF 提取进去，将该 UDF 的执行后果进行缓存，防止屡次调用；

■ Bucket Join

流表维表 Join 中存在数据冷启动问题，如果 Flink 工作在启动时大量加载内部数据，很容易造成反压。能够在启动时利用 State Processor API 等伎俩将全副数据预加载到内存中。但这种计划存在一种问题，维表数据加载到所有的 subtask 外面会造成较大的内存耗费。因而咱们的解决方案是，在维表的定义中指定一个 bucket 信息，流与维表进行 Join 的时候会基于 bucket 信息去加载维表中对应分片的数据，同时在翻译执行打算的时候流表拿到 bucket 信息，以保障流与维表的数据都会基于同一个 bucket 信息进行 Join。这种形式能大大减少全量维表数据预加载带来的内存耗费问题。

二、窗口性能扩大

腾讯实时计算平台基于现有 Flink SQL 语法进行了一些扩大，并另外定义了两种新的 Window 类型。

1. 新的窗口操作

现有如下需要，须要在两条流上针对某个工夫窗口做 Join 操作或者交并差操作。

应用 Flink SQL 基于某个 Window 去做双流 Join，现有的计划有两种，第一种计划就是先做 Join 再做 Group By，第二种就是 Interval Join。首先来剖析一下第一种计划是否满足需要。

■ 1.1 先 Join 再开窗

先 Join 再开窗的逻辑如上图所示，依据逻辑执行打算能够看到 Join 节点在 Window Aggregate 节点之下，所以会先进行流与流的 Join，Join 完了之后再去做 Window Aggregate。

图中右侧的流程图也能够看出，首先两边的流会做一个 Connect，而后基于 Join Key 做 Keyby 操作，以此保障两条流中领有雷同 Join Key 的数据可能 Shuffle 到同一个 task 上。左流会将数据存到本人的状态中，同时会去右流的状态中进行 Match，如果能 Match 上会将 Match 后的后果输入到上游。这种计划存在以下两个问题：

1. 状态无奈清理 ：因为 Join 在开窗之前，Join 外面并没有带 Window 的信息，即便上游的 Window 触发并实现计算，上游两条流的 Join 状态也无奈被清理掉，顶多只能应用基于 TTL 的形式去清理。
2. 语义无奈满足需要 ：原始的需要是想在两条流中基于雷同的工夫窗口去把数据进行切片后再 Join，然而以后计划并不能满足这样的需要，因为它先做 Join，应用 Join 后的数据再进行开窗，这种形式不能确保两条流中参加 Join 的数据是基于同一窗口的。

■ 1.2 Interval Join

Interval Join 绝对于后面一种写法，益处就是不存在状态无奈清理的问题，因为在扫描左右两条流的数据时能够基于某一确定的窗口，过了窗口工夫后，状态是能够被清理掉的。

然而这种计划绝对于第一种计划而言，数据准确性可能会更差一点，因为它对于窗口的划分不是基于一个确定窗口，而是基于数据进行驱动，即以后数据能够 Join 的另一条流上的数据的范畴是基于以后数据所携带的 Eventtime 的。这种窗口划分的语义与咱们的需要还是存在肯定差距的。

设想一下现有两条速率不统一的流，以 low 和 upper 两条边界来限定左流能够 Join 的右流的数据范畴，在如此死板的范畴束缚下，右流总会存在一些无效数据落在工夫窗口 [left + low, left + upper] 之外，导致计算不够精确。因而，最好还是依照窗口对齐的形式来划分工夫窗口，让两条流中 Eventtime 雷同的数据落在雷同的工夫窗口。

■ 1.3 Windowing Table-Valued Function

腾讯扩大出了 Windowing Table-Valued Function 语法，该语法能够满足“在两条流上针对某个工夫窗口做 Join 操作或者交并差操作”的需要。在 SQL 2016 规范中就有对于这一语法的形容，同时该语法在 Calcite1.23 外面就已存在。

Windowing Table-Valued Function 语法中的 Source 能够把它整个的语义形容分明，From 子句外面蕴含了 Window 定义所须要的所有信息，包含 Table Source、Eventtime Field、Window Size 等等。

从上图的逻辑打算能够看出，该语法在 LogiclTableScan 上加了一个叫 LogicalTableFunctionScan 的节点。另外，LogicalProject 节点（输入节点）多了两个字段叫作 WindowStart 和 WindowEnd，基于这两个字段能够把数据归纳到一个确定的窗口。基于以上原理，Windowing Table-Valued Function 语法能够做到上面这些事件：

• 在单流下面 ，能够像现有的 Group Window 语法一样去划分出一个工夫窗口。写法如上图，Window 信息全副放到 From 子句中，而后再进行 Group By。这种写法应该更合乎公众对于工夫窗口的了解，比以后 Flink SQL 中的 Group Window 的写法更加直观一点。咱们在翻译单流上的 Windowing Table-Valued Function 语法时做了一个讨巧，即在实现这段 SQL 的物理翻译时，并没有去翻译成具体的 DataStream API，而是将其逻辑执行打算间接变换到当初的 Group Window 的逻辑执行打算，也就是说共用了底层物理执行打算的代码，只是做了一个逻辑执行打算的等价。

另外，能够对 Window 外面的数据做一些 Sort 或者 TopN 的一些输入，因为 Windowing Table-Valued Function 语法曾经提前把数据划分进了一个个确定的窗口。如上图所示，首先在 From 子句外面把窗口划分好，而后 Order By 和 Limit 紧接其后，间接表白了排序和 TopN 语义。

• 在双流下面 ，能够满足“在两条流上针对某个工夫窗口做 Join 操作或者交并差操作”的原始需要。语法如上图，首先把两个窗口的 Window Table 结构好，而后利用 Join 关键字进行 Join 操作即可；交并差操作也一样，与传统数据库 SQL 的交并差操作无二。

■ 1.4 实现细节

上面简略介绍一下咱们在实现 Windowing Table-Valued Function 语法时的一些细节。

1.4.1 窗口的流传

原始的逻辑打算翻译形式，先基于 LogicalTableScan，而后再翻译到 Windowing Table-Valued Function，最初再翻译到 OrderBy Limit 子句。整个过程会存储很屡次状态，对于性能来讲会是比拟大的一个耗费，因而做了如下优化，把多个 Logical Relnode 合并在一起去翻译，这样能够缩小中间环节代码的产生，从而进步性能。

1.4.2 工夫属性字段

能够看到 Windowing Table-Valued Function 的语法：

SELECT * FROM TABLE(TUMBLE(TABLE <data>, DESCRIPTOR(<timecol>), <size> [, <offset>]))

table<data> 不仅仅能够是一张表，还能够是一个子查问。所以如果定义 Eventtime Field 的时候，把工夫属性和 Table Source 绑定，且 Table Source 恰好是一个子查问，此时就无奈满足咱们的需要。所以咱们在实现语法的时候，把工夫属性字段跟 Table Source 解耦，反之，用户应用物理表中的任意一个工夫字段来作为工夫属性，从而产生 watermark。

1.4.3 工夫水印

Watermark 的应用逻辑与在其余语法中一样，两条流的所有的 Input Task 的最小工夫水印，决定窗口的工夫水印，以此来触发窗口计算。

1.4.4 应用束缚

目前 Windowing Table-Valued Function 的应用存在一些束缚。首先，两条流的窗口类型必须是统一的，而且窗口大小也是一样的。而后，目前还没有实现 Session Window 相干的性能。

2. 新的窗口类型

接下来的介绍扩大出两个新的窗口类型。

■ 2.1 Incremental Window

有如下需要，用户心愿可能绘制一天内的 pv/uv 曲线，即在一天内或一个大的窗口内，输入屡次后果，而非等窗口完结之后对立输入一次后果。针对该需要，咱们扩大出了 Incremental Window。

2.1.1 屡次触发

基于 Tumble Window，自定义了 Incremental Trigger。该触发器确保，不仅仅是在 Windows 完结之后才去触发窗口计算，而是每个 SQL 中所定义的 Interval 周期都会触发一次窗口计算。

如上图中的 SQL 案例，总的窗口大小是一秒，且每 0.2 秒触发一次，所以在窗口内会触发 5 次窗口计算。且下一次的输入后果是基于上一次后果进行累计计算。

2.1.2 Lazy Trigger

针对 Incremental Window 做了一个名为 Lazy Trigger 的优化。在理论的生产过程中，一个窗口雷同 Key 值在屡次触发窗口计算后输入的后果是一样的。对于上游来讲，对于这种数据是没必要去反复接管的。因而，如果配置了 Lazy Trigger 的话，且在同一个窗口的同一个 Key 下，下一次输入的值跟上一次的是截然不同的，上游就不会接管到这次的更新数据，由此缩小上游的存储压力和并发压力。

■ 2.2 Enhanced Tumble Window

有如下需要，用户心愿在 Tumble Window 触发之后，不去抛弃早退的数据，而是再次触发窗口计算。如果应用 DataStream API，应用 SideOutput 就能够实现需要。然而对于 SQL，目前是没方法做到的。因而，扩大了现有的 Tumble Window，把早退的数据也收集起来，同时早退的数据并不是每来一条就从新触发窗口计算并向上游输入，而是会从新定义一个 Trigger，Trigger 的工夫距离应用 SQL 中定义的窗口大小，以此缩小向上游发送数据的频率。

同时，侧输入流在累计数据的时候也会应用 Window 的逻辑再做一次聚合。这里须要留神，如果上游是相似于 HBase 这样的数据源，对于雷同的 Window 雷同的 Key，前一条失常被窗口触发的数据会被早退的数据笼罩掉。实践上，早退的数据跟失常窗口触发的数据的重要性是一样的，不能互相笼罩。最初，上游会将收到的同一个窗口同一个 Key 下的失常数据和提早数据再做一次二次聚合。

三、回撤流优化

接下来介绍一下在回撤流上所做的一些优化。

1. 流表二义性

回顾一下对于在 Flink SQL 中对于回撤流的一些概念。

首先介绍一下继续查问（Continuous Query），绝对于批处理一次执行输入一次后果的特点，流的聚合是上游来一条数据，上游的话就会接管一条更新的数据，即后果是一直被上游的数据所更新的。因而，对于同一个 Key 上游可能接管到多条更新后果。

2. 回撤流

以上图的 SQL 为例，当第二条 Java 达到聚合算子时，会去更新第一条 Java 所产生的状态并把后果发送到上游。如果上游对于屡次更新的后果不做任何解决，就会产生谬误的后果。针对这种场景，Flink SQL 引入了回撤流的概念。

所谓回撤流的话，就是在原始数据前加了一个标识位，以 True/False 进行标识。如果标识位是 False，就示意这是一条回撤音讯，它告诉上游对这条数据做 Delete 操作；如果标识位是 True，上游间接会做 Insert 操作。

■ 2.1 什么时候产生回撤流

目前，Flink SQL 外面产生回撤流有以下四种场景：

• Aggregate Without Window（不带 Window 的聚合场景）
• Rank
• Over Window
• Left/Right/Full Outer Join

解释一下 Outer Join 为什么会产生回撤。以 Left Outer Join 为例，且假如左流的数据比右流的数据先到，左流的数据会去扫描右流数据的状态，如果找不到能够 Join 的数据，左流并不知道右流中是的确不存在这条数据还是说右流中的相应数据早退了。为了满足 Outer join 的语义的话，右边流数据还是会产生一条 Join 数据发送到上游，相似于 MySQL Left Join，左流的字段以失常的表字段值填充，右流的相应字段以 Null 填充，而后输入到上游，如下图所示：

（图片来源于云栖社区）

前期如果右流的相应数据达到，会去扫描左流的状态再次进行 Join，此时，为了保障语义的正确性，须要把后面曾经输入到上游的这条非凡的数据进行回撤，同时会把最新 Join 上的数据输入到上游。留神，对于雷同的 Key，如果产生了一次回撤，是不会再产生第二次回撤的，因为如果前期再有该 Key 的数据达到，是能够 Join 上另一条流上相应的数据的。

■ 2.2 如何解决回撤音讯

上面介绍 Flink 中解决回撤音讯的逻辑。

对于两头计算节点，通过上图中的 4 个标记位来管制，这些标识位示意以后节点是产生 Update 信息还是产生 Retract 信息，以及以后节点是否会生产这个 Retract 信息。这 4 个标识位可能决定整个对于 Retract 的产生和解决的逻辑。

对于 Sink 节点，目前 Flink 中有三种 sink 类型，AppendStreamTableSink、RetractStreamTableSink 和 UpsertStreamTableSink。AppendStreamTableSink 接管的上游数据是一条 Retract 信息的话会间接报错的，因为它只能形容 Append-Only 语义；RetractStreamTableSink 则能够解决 Retract 信息，如果上游算子发送一个 Retract 信息过去，它会对音讯做 Delete 操作，如果上游算子发送的是失常的更新信息，它会对音讯做 Insert 操作；UpsertStreamTableSink 能够了解为对于 RetractStreamTableSink 做了一些性能的优化。如果 Sink 数据源反对幂等操作，或者反对依照某 key 做 Update 操作，UpsertStreamTableSink 会在 SQL 翻译的时候把上游 Upsert Key 传到 Table Sink 外面，而后基于该 Key 去做 Update 操作。

■ 2.3 相干优化

咱们基于回撤流做以下优化。

2.3.1 两头节点的优化

产生回撤信息最基本的一个起因是一直地向上游屡次发送更新后果，因而，为了缩小更新的频率并升高并发，能够把更新后果累计一部分之后再发送进来。如上图所示：

• 第一个场景是一个嵌套 AGG 的场景（例如两次 Count 操作），在第一层 Group By 尝试将更新后果发送到上游时候会先做一个 Cache，从而缩小向上游发送数据频率。当达到了 Cache 的触发条件时，再把更新后果发送到上游。
• 第二个场景是 Outer Join，后面提到，Outer Join 产生回撤音讯是因为左右两边数据的速率不匹配。以 Left Outer Join 为例，能够把左流的数据进行 Cache。左流数据达到时会去右流的状态外面查找，如果能找到能够与之 Join 的数据则不作缓存；如果找不到相应数据，则对这条 Key 的数据先做缓存，当达到某些触发条件时，再去右流状态中查找一次，如果依然找不到相应数据，再去向上游发送一条蕴含 Null 值的 Join 数据，之后右流相应数据达到就会将 Cache 中该 Key 对应的缓存清空，并向上游发送一条回撤音讯。

以此来减小向上游发送回撤音讯的频率。

2.3.2 Sink 节点的优化

针对 Sink 节点做了一些优化，在 AGG 节点和 Sink 节点之间做了一个 Cache，以此加重 Sink 节点的压力。当回撤音讯在 Cache 中再做聚合，当达到 Cache 的触发条件时，对立将更新后的数据发送到 Sink 节点。以下图中的 SQL 为例：

参考优化前后的输入后果能够看到，优化后上游接管到的数据量是有缩小的，例如用户 Sam，当回撤音讯尝试发送到上游时，先做一层 Cache，上游接管到的数据量能够缩小很多。

四、将来布局

上面介绍一下咱们团队后续的工作布局：

• Cost-Based Optimization：当初 Flink SQL 的逻辑执行打算的优化还是基于 RBO（Rule Based Optimization）的形式。咱们团队想基于 CBO 所做一些事，次要的工作还是统计信息的收集。统计信息不仅仅来自 Flink SQL 自身，可能还会来自公司内其余产品，例如元数据，不同 Key 所对应的数据分布，或者其余数据分析后果。通过跟公司内其余产品买通，拿到最准的统计数据，产生最优的执行打算。
• More New Features（CEP Syntax etc.）：基于 Flink SQL 定义一些 CEP 的语法，以满足用户对于 CEP 的一些需要。
• Continuous Performance Optimization（Join Operator etc.）：咱们团队在做的不仅仅是执行打算层的优化，也在做 Join Operator 或者说数据 Shuffle 的一些细粒度的优化。
• Easier To Debug：最初是对于 Flink SQL 工作的调试和定位。目前 Flink SQL 在这方面是比拟欠缺的，特地是线上对于数据对不齐的问题，排查起来十分的辣手。咱们目前的思路是通过配置的形式，让 SQL 在执行的过程中吐出一些 Trace 信息或者一些 Metrics 信息，而后发送到其余平台。通过这些 Trace 信息和 Metric 信息，帮忙用户定位出问题的算子。