Apache Flink-实时计算正当时

第2期

Phase2ApacheFlink

2

。

12实时计算正当时

版ApacheFUnk技术与实战精解

I目录

■技术篇

ApacheFlink1.13.0正式发布，流处理应用更加简单高效！1

深入解读FlinkSQL1.1319

ApacheFlink1.13,面向流批一体的运行时与DatastreamAPI优化41

ApacheFlink1.13,StateBackend优化及生产实践分享59

Flink+Iceberg全场景实时数仓的建设实践72

■实践篇

知乎的Flink数据集成平台建设实践94

腾讯游戏实时计算应用平台建设实践107

FUnkSQLCDC实践以及一致性分析128

ApacheFlink在bilibili的多元化探索与实践149

ApacheFlink1.13.0正式发布,流处理应用更加简单高效！

ApacheFlink1.13.0正式发布，

流处理应用更加简单高效！

来源IFlink中文社区

翻译|高赞

Review|朱翥、马国维

ApacheFlink1.13发布了！Flink1.13包括了超过200名贡献者所提交的1000多项修

复和优化。

这一版本中，Flink的一个主要目标取得了重要进展，即让流处理应用的使用像普通

应用一样简单和自然。Flink1.13新引入的被动扩缩容使得流作业的扩缩容和其它应

用一样简单，用户仅需要修改并发度即可。

这个版本还包括一系列重要改动使用户可以更好理解流作业的性能。当流作业的性

能不及预期的时候，这些改动可以使用户可以更好的分析原因。这些改动包括用于

识别瓶颈节点的负载和反压可视化、分析算子热点代码的CPU火焰图和分析State

Backend状态的State访问性能指标。

除了这些特性外，Flink社区还添加了大量的其它优化，本文后续会讨论其中的一些。

我们希望用户可以享受新的版本和特性带来的便利，在本文最后，我们还会介绍升级

Flink版本需要注意的一些变化。

我们鼓励用户下载试用新版Flink［1］并且通过邮件列表［2］和JIRA［3］来反馈遇到的问

题。

一、重要特性

被动扩缩容

Flink项目的一个初始目标，就是希望流处理应用可以像普通应用一样简单和自然，

被动扩缩容是Flink针对这一目标上的最新进展。

当考虑资源管理和部分的时候，Flink有两种可能的模式。用户可以将Flink应用部署

到k8s、yarn等资源管理系统之上，并且由Flink主动的来管理资源并按需分配和释

放资源。这一模式对于经常改变资源需求的作业和应用非常有用，比如批作业和实时

SQL查询。在这种模式下，Flink所启动的Worker数量是由应用设置的并发度决定的。

在FLink中我们将这一模式叫做主动扩缩容。

对于长时间运行的流处理应用，一种更适合的模型是用户只需要将作业像其它的长

期运行的服务一样启动起来，而不需要考虑是部署在k8s、yarn还是其它的资源管

理平台上，并且不需要考虑需要申请的资源的数量。相反，它的规模是由所分配的

worker数量来决定的。当worker数量发生变化时，Rink自动的改动应用的并发度。

在Flink中我们将这一模式叫做被动扩缩容。

Flink的Application部署模式［4］开启了使Flink作业更接近普通应用（即启动Flink

作业不需要执行两个独立的步骤来启动集群和提交应用）的努力，而被动扩缩容完成

了这一目标：用户不再需要使用额外的工具（如脚本、K8s算子）来让Worker的数

量与应用并发度设置保持一致。

用户现在可以将自动扩缩容的工具应用到Flink应用之上,就像普通的应用程序一样，

只要用户了解扩缩容的代价：有状态的流应用在扩缩容的时候需要将状态重新分发。

如果想要尝试被动扩缩容，用户可以增力口scheduler-mode:reactive这一酉d置项，然

后启动一个应用集群（Standalone［5］或者K8s⑹）。更多细节见被动扩缩容的文档［7］o

分析应用的性能

对所有应用程序来说，能够简单的分析和理解应用的性能是非常关键的功能。这一

功能对Flink更加重要，因为Flink应用一般是数据密集的（即需要处理大量的数据）

并且需要在（近）实时的延迟内给出结果。

3

当Flink应用处理的速度跟不上数据输入的速度时，或者当一个应用占用的资源超过

预期，下文介绍的这些工具可以帮你分析原因。

■瓶颈检测与反压监控

Flink性能分析首先要解决的问题经常是：哪个算子是瓶颈？

为了回答这一问题，Flink引入了描述作业繁忙（即在处理数据）与反压（由于下游

算子不能及时处理结果而无法继续输出）程度的指标。应用中可能的瓶颈是那些繁忙

并且上游被反压的算子。

Flink1.13优化了反压检测的逻辑（使用基于任务Mailbox计时，而不在再于堆栈采

样），并且重新实现了作业图的UI展示：Flink现在在UI上通过颜色和数值来展示繁

忙和反压的程度。

■WebUI中的CPU火焰图

Flink关于性能另一个经常需要回答的问题：瓶颈算子中的哪部分计算逻辑消耗巨大？

针对这一问题，一个有效的可视化工具是火焰图。它可以帮助回答以下问题：

•哪个方法调用现在在占用CPU?

•不同方法占用CPU的比例如何？

•一个方法被调用的栈是什么样子的？

火焰图是通过重复采样线程的堆栈来构建的。在火焰图中，每个方法调用被表示为一

个矩形，矩形的长度与这个方法出现在采样中的次数成正比。火焰图在UI上的一个

例子如下图所示。

termarfcsAccumulatorsBackpressureMet/icsHameGraph

On-CPUMixedMeasurement:14sago

DynamicRuleEvaluationFuncti

on->(AlertsDesenalizanon->■■■

Sink:AlertsJSONSink.RulesDljdd:l£6

eseriaNzation->Sink:Unname

6

Parallelism:1

Backpressureci(max)0%

Busy(max):45%

LowWatermark:16145301506$

火焰图的文档［8］包括启用这一功能的更多细节和指令。

■State访问延迟指标

另一个可能的性能瓶颈是statebackend,尤其是当作业的state超过内存容量而必

须使用RocksDBstatebackend［9］时。

这里并不是想说RocksDB性能不够好（我们非常喜欢RocksDB!）,但是它需要满

足一些条件才能达到最好的性能。例如，用户可能很容易遇到非故意的在云上由于

使用了错误的磁盘资源类型而不能满足RockDB的10性能需求［10］的问题。

基于CPU火焰图，新的StateBackend的延迟指标可以帮助用户更好的判断性能不

符合预期是否是由StateBackend导致的。例如，如果用户发现RocksDB的单次访

问需要几毫秒的时间，那么就需要查看内存和I/O的配置。这些指标可以通过设置

state.backend.rocksdb.latency-track-enabled这一选项来启用。这些指标是通过采

样的方式来监控性能的，所以它们对RocksDBStateBackend的性能影响是微不足道

的。

5

通过Savepoint来切换StateBackend

用户现在可以在从一个Savepoint重启时切换一个Flink应用的StateBackend。这

使得Flink应用不再被限制只能使用应用首次运行时选择的StateBackend。

基于这一功能，用户现在可以首先使用一个HashMapStateBackend（纯内存的

StateBackend）,如果后续状态变得过大的话，就切换到RocksDBStateBackend中。

在实现层，Flink现在统一了所有StateBackend的Savepoint格式来实现这一功能。

K8s部署时使用用户指定的Pod模式

原生kubernetes部署口1]（Flink主动要求K8s来启动Pod）中，现在可以使用自定

义的Pod模板。

使用这些模板，用户可以使用一种更符合K8s的方式来设置JM和TM的Pod,这种

方式比FlinkK8s集成内置的配置项更加灵活。

生产可用的UnalignedCheckpoint

UnalignedCheckpoint目前已达到了生产可用的状态，我们鼓励用户在存在反压的

情况下试用这一功能。

具体来说，ApacheFlink1.13中引入的这些功能使UnalignedCheckpoint更容易使

用：

•用户现在使用UnalignedCheckpoint时也可以扩缩容应用。如果用户需要因为性

能原因不能使用Savepoint而必须使用Retainedcheckpoint时，这一功能会非常

方便。

•对于没有反压的应用，启用UnalignedCheckpoint现在代价更小。Unaligned

Checkpoint现在可以通过超时来自动触发，即一个应用默认会使用Aligned

Checkpoint（不存储传输中的数据），而只在对齐超过一定时间范围时自动切换

至UUnalignedCheckpoint（存储传输中的数据）。

关于如何启用UnalignedCheckpoint可以参考相关文档口2]。

机器学习迁移到单独的仓库

为了加速Flink机器学习的进展（流批统一的机器学习），现在Flink机器学习开启了

新的仓库。我们采用类似于StatefulFunction项目的管理方式，通过使

用一个单独的仓库从而简化代码合并的流程并且可以进行单独的版本发布，从而提高

开发的效率。

用户可以关注FMk在机器学习方面的进展，比如与Alink口4]（Flink常用机器学习算

法套件）的互操作以及与的集成

FlinkTensorflow[15]o

二、SQL/TableAPI进展

与之前的版本类似，SQL和TableAPI仍然在所有开发中占用很大的比例。

通过Table-valued函数来定义时间窗口

在流式SQL查询中，一个最经常使用的是定义时间窗口。ApacheFlink1.13中引入

了一种新的定义窗口的方式：通过Table-valued函数。这一方式不仅有更强的表达

能力（允许用户定义新的窗口类型），并且与SQL标准更加一致。

ApacheFlink1.13在新的语法中支持TUMBLE和HOP窗口，在后续版本中也会支持

SESSION窗口。我们通过以下两个例子来展示这一方法的表达能力：

•例1：一个新引入的CUMULATE窗口函数，它可以支持按特定步长扩展的窗口，

直到达到最大窗口大小：

SELECTwindow_time,window_start,window_end,SUM(price)AStotal_price

FROMTABLE(CUMULATE(TABLEBid,DESCRIPTOR(bidtime),INTERVAL'2'MINUTES,INTERVAL'10'

MINUTES))

GROUPBYwindow_start?window_end?window_time;

•例2:用户在table-valued窗口函数中可以访问窗口的起始和终止时间，从而使

用户可以实现新的功能。例如，除了常规的基于窗口的聚合和Join之外,用户现

在也可以实现基于窗口的Top-K聚合：

SELECTwindow』me,…

FROM(

SELECT"ROW_NUMBER()OVER(PARTITIONBYwindow_start,window_endORDERBYtotal_price

DESC)

asrank

FROMt

)WHERErank<=100;

提高DatastreamAPI与TableAPI/SQL的互操作能力

这一版本极大的简化了DatastreamAPI与TableAPI混合的程序。

TableAPI是一种非常方便的应用开发接口,因为这仅支持表达式的程序编写并提供

了大量的内置函数。但是有时候用户也需要切换回Datastream,例如当用户存在表

达能力、灵活性或者State访问的需求时。

Flink新引入的StreamTableEnvironment.toDataStreaiTi()/.fromDataStream()可以

将一个DatastreamAPI声明的Source或者Sink当作Table的Source或者Sink来使

用。主要的优化包括：

•Datastream与TableAPI类型系统的自动转换。

•EventTime配置的无缝集成，Watermark行为的高度一致性。

•Row类型(即TableAPI中数据的表示)有了极大的增强，包括toString()/

hashCodeQ和equals。方法的优化，按名称访问字段值的支持与稀疏表示的支持。

Tabletable=tableEnv.fromDataStream(

datastream,

Schema.newBuilder()

n

.columnByMetadata(rowtime","TIMESTAMPtB)")

n

.watermark("rowtime","SOURCE_WATERMARK())

.buildQ)；

DataStream<Row>datastream=tableEnv.toDataStream(table)

.keyBy(r->r.getField("user"))

.window(...);

SQLClient:初始化脚本和语句集合(StatementSets)

SQLClient是一种直接运行和部署SQL流或批作业的简便方式，用户不需要编写代码

就可以从命令行调用SQL,或者作为Cl/CD流程的一部分。

这个版本极大的提高了SQLClient的功能。现在基于所有通过Java编程(即通过编

程的方式调用TableEnvironment来发起查询)可以支持的语法，现在SQLClient和

SQL脚本都可以支持。这意味着SQL用户不再需要添加胶水代码来部署他们的SQL

作业。

■配置简化和代码共享

Flink后续将不再支持通过Yami的方式来配置SQLClient（注：目前还在支持，但是

已经被标记为废弃）。作为替代，SQLClient现在支持使用一个初始化脚本在主SQL

脚本执行前来配置环境。

这些初始化脚本通常可以在不同团队/部署之间共享。它可以用来加载常用的

catalog,应用通用的配置或者定义标准的视图。

,/sql-client.sh-iinitl.sqlinit2.sql-fsqljob.sql

■更多的配置项

通过增加配置项，优化SET/RESET命令，用户可以更方便的在SQLClient和SQL脚

本内部来控制执行的流程。

■通过语句集合来支持多查询

多查询允许用户在一个Flink作业中执行多个SQL查询（或者语句）。这对于长期运

行的流式SQL查询非常有用。

语句集可以用来将一组查询合并为一组同时执行。

以下是一个可以通过SQLClient来执行的SQL脚本的例子。它初始化和配置了执行

多查询的环境。这一脚本包括了所有的查询和所有的环境初始化和配置的工作，从而

使它可以作为一个自包含的部署组件。

-setupacatalog

CREATECAIALOGhive_catalogWITH('type'='hive');

USECATALOGhive_catalog;

—orusetemporaryobjects

CREATETEMPORARYTABLEclicks(

userjdBIGINT

pagejdBIGINT

viewtimeTIMESTAMP

10

)WITH(

'connector'='kafka'

'topic'='clicks*

'properties.bootstrap.servers'=:;

'format'='avro*

)；

—settheexecutionmodeforjobs

SETexecution.runtime-mode=streaming;

-setthesync/asyncmodeforINSERTINTOs

SETtabLe.dmI-sync=faIse;

-setthejob'sparallelism

SETparallism.default=10;

—setthejobname

SET=my_flinkjob;

—restorestatefromthespecificsavepointpath

SETexecution.savepoint.path=/tmp/flink-savepoints/savepoint-bbOdab;

BEGINSTATEMENTSET;

INSERTINTOpageview_pv_sink

SELECTpagejd,count(l)FROMclicksGROUPBYpagejd;

INSERTINTOpageview_uv__sink

SELECTpagejd,count(distinctuserjd)FROMclicksGROUPBYpagejd;

END;

Hive查询语法兼容性

用户现在在Flink上也可以使用HiveSQL语法。除了HiveDDL方言之外，Flink现在

也支持常用的HiveDML和DQL方言。

为了使用HiveSQL方言，需要设置table.sql-dialect为hive并且加载HiveModule0

后者非常重要，因为必须要加载Hive的内置函数后才能正确实现对Hive语法和语义

的兼容性。例子如下：

CREATECATALOGmyhiveWITH('type'='hive');"setupHiveCatalog

USECATALOGmyhive;

LOADMODULEhive;-setupHiveModule

USEMODULEShive,core;

SETtable.sql-dialect;hive;-enableHivedialect

SELECTkeyvalueFROMsrcCLUSTERBYkey;—runsomeHivequeries

需要注意的是，Hive方言中不再支持Flink语法的DML和DQL语句。如果要使用

Flink语法，需要切换回default的方言配置。

优化的SQL时间函数

在数据处理中时间处理是一个重要的任务。但是与此同时，处理不同的时区、日期

和时间是一个日益复杂［16］的任务。

在Flink1.13中，我们投入了大量的精力来简化时间函数的使用。我们调整了时间

相关函数的返回类型使其更加精确，例如PROCTIMEQ,CURRENT_TIMESTAMP()和

N0W()o

其次，用户现在还可以基于一个TIMESTAMP_LTZ类型的列来定义EventTime属性，

从而可以优雅的在窗口处理中支持夏令时。

用户可以参考ReleaseNote来查看该部分的完整变更。

12

三、PyFlink核心优化

这个版本对PyFlink的改进主要是基于Python的DatastreamAPI和TableAPI还有

Java/seaLa版本的对应功能更加一致。

PythonDatastreamAPI中的有状态算子

在ApacheFlink1.13中,Python程序员可以享受到Flink状态处理API的所有能力。

在ApacheFlink1.12版本重构过的PythonDatastreamAPI现在已经拥有完整的状

态访问能力，从而使用户可以将数据的信息记录到state中并且在后续访问。

带状态的处理能力是许多依赖跨记录状态共享(例如WindowOperator)的复杂数

据处理场景的基础。

以下例子展示了一个自定义的计算窗口的实现：

classCountWindowAverage(FlatMapFunction):

def_init_(self?window_size):

self.window__size=window_size

defopen(selfzruntime_context:Runtimecontext):

descriptor=ValueStateDescriptor("average''Types.TUPLE([Types.LONG(),Types.LONG()]))

self.sum=runtime_context.get_state(descriptor)

defflat_map(self,value):

current_sum=self.sum.valueQ

ifcurrent_sumisNone:

current_sum=(0z0)

#updatethecount

current_sum=(current_sum[0]+1,current_sum[1]+value[1])

#ifthecountreacheswindow_sizexemittheaverageandclearthestate

ifcurrent_sum[0]>=self.window_size:

13

self.sum.clearQ

yieldvalue[O]/current_sum[l]//current_sum[O]

else:

self.sum.update(current_sum)

ds=...#type:Datastream

ds.key_by(lambdarow:row[0])\

.flat_map(CountWindowAverage(5))

PyFlinkDatastreamAPI中的用户自定义窗口

ApacheFlink1.13中PyFlinkDatastream接口增加了对用户自定义窗口的支持，现在

用户可以使用标准窗口之外的窗口定义。

由于窗口是处理无限数据流的核心机制(通过将流切分为多个有限的『桶』)，这

一功能极大的提高的API的表达能力。

PyFlinkTableAPI中基于行的操作

PythonTableAPI现在支持基于行的操作，例如用户对行数据的自定义函数。这一功

能使得用户可以使用非内置的数据处理函数。

一个使用map()操作的PythonTableAPI示例如下：

@udf(result_type=DataTypes.ROW(

[DataTypes.FIELDf'cl:DataTypes.BIGINT()).

DataTypes.FIELD("c2",DataTypes.STRING())]))

defincrement_column(r:Row)->Row:

returnRow(r[0]+1zr[1])

table=...#type:Table

mapped_result=table.map(increment_column)

14

除了map(),这一API还支持flat_map(),aggregate。，flat_aggregate()和其它基

于行的操作。这使PythonTableAPI的功能与JavaTableAPI的功能更加接近。

PyFlinkDatastreamAPI支持Batch执行模式

对于有限流，PyFlinkDatastreamAPI现在已经支持Flink1.12DatastreamAPI中引

入的Batch执行模式。

通过复用数据有限性来跳过Statebackend和Checkpoint的处理，Batch执行模式

可以简化运维，并且提高有限流处理的性能。

四、其它优化

基于Hugo的Flink文档

Flink文档从Jekyll迁移到了Hugo。如果您发现有问题，请务必通知我们，我们非常

期待用户对新的界面的感受。

WebUI支持历史异常

FlinkWebUI现在可以展示导致作业失败的n次历史异常，从而提升在一个异常导致

多个后续异常的场景下的调试体验。用户可以在异常历史中找到根异常。

优化失败Checkpoint的异常和失败原因的汇报

Flink现在提供了失败或被取消的Checkpoint的统计，从而使用户可以更简单的判断

Checkpoint失败的原因，而不需要去查看日志。

Flink之前的版本只有在Checkpoint成功的时候才会汇报指标(例如持久化数据的大

小、触发时间等)。

提供『恰好一次』一致性的JDBCSink

15

从ApacheFlink1.13开始，通过使用事务提交数据，JDBCSink可以对支持XA事务

的数据库提供力恰好一次」的一致性支持。这一特性要求目标数据库必须有（或链接

到）一个XA事务处理器。

PyFlinkTableAPI在Group窗口上支持用户自定义的聚合函数

PyFlinkTableAPI现在对Group窗口同时支持基于Python的用户自定义聚合函数

（User-definedAggregateFunctions,UDAFs）以及PandasUDAFs。这些函数对许

多数据分析或机器学习训练的程序非常重要。

在Flink1.13之前,这些函数仅能在无限的Group-by聚合场景下使用。Flink1.13优

化了这一限制。

Batch执行模式下Sort-mergeShuffle优化

ApacheFlink1.13优化了针对批处理程序的Sort-mergeBlockingShuffle的性能和

内存占用情况。这一Shuffle模式是在Flink1.12的FLIP-148[17]中引入的。

这一优化避免了大规模作业下不断出现OutOfMemoryError:DirectMemory的问题，

并且通过I/。调度和broadcast优化提高了性能（尤其是在机械硬盘上）。

HBase连接器支持异步维表查询和查询缓存

HBaseLookupTableSource现在可以支持异步查询模式和查询缓存。这极大的提高

了使用这一Source的Table/SQL维表Join的性能，并且在一些典型情况下可以减

少对HBase的I/O请求数量。

在之前的版本中，HBaseLookupSource仅支持同步通信，从而导致作业吞吐以及资

源利用率降低。

16

升级Flink1.13需要注意的改动：

•FLINK-21709口8]-老的Table&SQLAPI计划器已经被标记为废弃，并且将在Flink

1.14中被删除。Blink计划器在若干版本之前已经被设置为默认计划器，并且将成

为未来版本中的唯一计划器。这意味着BatchTableEnvironmentWDataSetAPI互

操作后续也将不再支持。用户需要切换到统一的TableEnvironment来编写流或者

批的作业。

•FUNK-22352口9]-Flink社区决定废弃对Apachemesos的支持，未来有可能会进

一步删除这部分功能。用户最好能够切换到其它的资源管理系统上。

•FLINK-21935[20]-state.backend.async这一配置已经被禁用了，因为现在Flink

总是会异步的来保存快照(即之前的配置默认值)，并且现在没有实现可以支持

同步的快照保存操作。

•FLINK-17012[21]-Task的RUNNING状态被细分为两步：INITIALIZING和

RUNNINGOTask的INITIALIZING阶段包括加载state和在启用unaligned

checkpoint时恢复In-flight数据的过程。通过显式区分这两种状态，监控系统可

以更好的区分任务是否已经在实际工作。

•FUNK-21698[22]-NUMERIC和TIMESTAMP类型之间的直接转换存在问题，

现在已经被禁用，例如CAST(numericASTIMESTAMP(3))。用户应该使用T0_

TIMESTAMP(FROM_UNIXTIME(numeric))来代替。

•FUNK-22133[23]-新的Source接口有一个小的不兼容的修改，即

SplitEnumerator.snapshotStateQ方法现在多接受一个checkpointid参数来表示正

在进行的snapshot操作所属的checkpoint的id。

•FLINK-19463[24]-由于老的Statebackend接口承载了过多的语义并且容易引起困

惑，这一接口被标记为废弃。这是一个纯API层的改动，而并不会影响应用运行时。

对于如何升级现有作业，请参考作业迁移指引[25]。

五、其它资源

17

二进制和代码可以从ApacheFlink官网的下载页面[26]获得，最新的PyFUnk发布可

以从PyPI[27]获得。

如果想要升级到ApacheFlink1.13,请参考发布说明[28]。这一版本与之前1.x的版

本在标记为@Public的接口上是兼容的。

用户也可以查看新版本修改列表[29]与更新后的文档[30]来获得修改和新功能的详

细列表。

■原文链接:/news/2021/05/03/release-l.lB.O.html

参考链接：

[1]/downloads.html

[2]https://flink.apache.Org/community.html#mailing-lists

[3]/jira/projects/FLINK/summary

[4]/projects/flink/flink-docs-release-l.13/docs/concepts/flink-

architecture/#flink-application-execution

[5]/projects/flink/flink-docs-release-L13/docs/deployment/resource-

providers/standalone/overview/#application-mode

[6]/projects/flink/flink-docs-release-l.13/docs/deployment/resource-

providers/standalone/kubernetes/#deploy-application-cluster

[7]https://ci.apache.Org/projects/flink/flink-docs-release-l.13/docs/deployment/elastic_

scaling/#reactive-mode

[8]/projects/flink/flink-docs-release-l.13/docs/ops/debugging/flame_graphs

[9]https://ci.apache.0rg/pr0jects/flink/flink-d0cs-release-l.l3/docs/ops/state/state_

backends/#the-embeddedrocksdbstatebackend

[10]/blog/the-impact-of-disks-on-rocksdb-state-backend-in-flink-a-

case-study

[11]/projects/flink/flink-docs-release-1J3/docs/deployment/resource-

providers/native_kubernetes/

18

[12]https://ci.apache.org/projects/flink/flink-docs-release-l-l3/docs/ops/state/

checkpoints/#unaligned-checkpoints

[13]h注ps:〃/apache/flink-ml

[14]/alibaba/Alink

[15]/alibaba/flink-ai-extended

[16]/1883/

[17]/confluence/display/FLINK/FLIP-148%3A+lntroduce+Sort-Merge+Based

+Blocking+Shuffle+to+Flink

[18]/jira/browse/FLINK-21709

[19]h注ps:〃issues.apacheQ「g/ji「a/b「owse/FLINK-22352

[20]/jira/browse/FLINK-21935

[21]/jira/browse/FLINK-17012

[22]/jira/browse/FLINK-21698

[23]/jira/browse/FLINK-22133

[24]/jira/browse/FLINK-19463

[25]https://ci.apach/projects/flink/flink-docs-release-1.l3/docs/ops/state/state_

backends/#migrating-from-legacy-backends

[26]/downloads.html

[27]/project/apache-flink/

[28]/projects/flink/flink-docs-release-l.13/release-notes/flink-l.13

[29]/jira/secure/ReleaseNote.jspa?projectld=12315522&version=12349287

[30]/projects/flink/flink-docs-release-l.13/

深入解读FlinkSQL1.1319

深入解读RinkSQL1.13

来源IFlink中文社区

作者|徐榜江@阿里

摘要：ApacheFlink社区在5月份发布了1.13版本，带来了很多新的变化。本文整

理自徐榜江（雪尽）5月22日在北京的FlinkMeetup分享的《深入解读FlinkSQL

1.13»,内容包括：

❶FlinkSQL1.13概览

❷核心feature解读

o重要改进解读

QFlinkSQL1.14未来规划J

❷总结

20

一、FlinkSQL1.13概览

Flink1.13

Flink1.13issues

Others,92,9%

>解决Issues:1000+

>贡献者：200+

>Table/SQL:400+

BTable/SQL■Runtime«API-Connectors

L■Checkpoint■StatebackendBOthers

FlinkSQL1.13是一个社区大版本，解决的issue在1000个以上，通过上图我们可以

看到，解决的问题大部分是关于Table/SQL模块，一共400多个issue占了总体的

37%左右。这些issue主要围绕了5个FLIP展开，在本文中我们也会根据这5个方

面进行介绍，它们分别是：

21

下面我们对这些FLIP进行详细解读。

二、核心feature解读

1.FLIP-145:支持WindowTVF

社区的小伙伴应该了解，在腾讯、阿里巴巴、字节跳动等公司的内部分支已经开发了

这个功能的基础版本。这次Flink社区也在Flink1.13推出了TVF的相关支持和优化。

下面将从WindowTVF语法、近实时累计计算场景、Window性能优化、多维数据分析,

来解读这个新功能。

FLIP-145:支持WindowTVF

WindowTVF语法

■1.1WindowTVF语法

在1.13版本前，window的实现是通过一个特殊的SqlGroupedWindowFunction:

SELECT

TUMBLE_S1ART(bidtimeJNTERVAL'10'MINUTE),

TUMBLE_END(bidtimeJNTERVAL'10*MINUTE),

TUMBLE_ROWTIME(bidtimeJNTERVAL'10'MINUTE),

SUM(price)

FROMMyTable

GROUPBYTUMBLE(bidtimeJNTERVAL'10'MINUTE)

22

在1.13版本中，我们对它进行了Table-ValuedFunction的语法标准化：

SELECTWINDOW_start,WINDOW_end,WINDOW_time,SUM(price)

FROMTable(TUMBLE(TablemyTable,DESCRIPTOR(biztime),INTERVAL'10'MINUTE))

GROUPBYWINDOW_start,WINDOW_end

通过对比两种语法，我们可以发现：TVF语法更加灵活，不需要必须跟在GROUPBY

关键字后面，同时WindowTVF基于关系代数，使得其更加标准。在只需要划分窗口

场景时,可以只用TVF,无需用GROUPBY做聚合，这使得TVF扩展性和表达能力更强,

支持自定义TVF(例如实现TOP-N的TVF)。

FUnkSQL>

$EL£CT-FROMTABLE(

fnmiElTABlfNyTable.DESCRIFTOR(bldtlM),IMTfRVAL'IfNINUTf5>)；

Ibidtlac|price|Iten|window.start|window_end|

WindowTVF语法

＞完整的关系代数表达

＞输入是一个关系，输出也是一个关系

＞每个关系对应成一个数据集

上图中的示例就是利用TVF做的滚动窗口的划分，只需要把数据划分到窗口，无需

聚合；如果后续需要聚合，再进行GROPBY即可。同时，对于熟悉批SQL的用户

来说，这种操作是非常自然的，我们不再需要像113版本之前那样必须要用特殊的

SQLGroupedWindowFunction将窗口划分和聚合绑定在一起。

目前WindowTVF支持tumblewindow,hopwindow,新增了cumulatewindow;

sessionwindow预计在1.14版本也会支持。

23

■1.2CumulateWindow

CumulateWindow

Cumulatewindow就是累计窗口，简单来说，以上图里面时间轴上的一个区间为窗

口步长。

•第一个window统计的是一个区间的数据；

•第二个window统计的是第一区间和第二个区间的数据；

•第三个window统计的是第一区间，第二个区间和第三个区间的数据。

累积计算在业务场景中非常常见，如累积UV场景。在UV大盘曲线中：我们每隔10

分钟统计一次当天累积用户UV。

24

近实时累计计算

INSERTINTOcuauUtIve.uv

SELECTdate.str,MAX(tine_$tr),COUWT(DISTINCTuser.id)uv

FROM(

UV大盘曲线SELECT

DATE_FO«MAT(ts,•yyyy-**-<W')asdate.str,

SUBSTR(DATE_FORMAT(tS,||astWe.Str,

user-id

＞结果更精确FROMuser_behavlor)

GROUPBYdate_str;

＞追数据时曲线不会跳变

INSERTDfTOCUWlatlVC.Uv

SELECTwindow.end,COIMT(D1ST1NCTuser.ld)asuv

FMMTABLE(

CUMULATE(TABLEuser_t>en»vlor,OESCRIPT(M(tS>.INTERVAL’16'MINUTES,INTERVAL

GMUPBYwlndow.start,wlndow-efld;

在1.13版本之前，当需要做这种计算时，我们一般的SQL写法如下:

INSERTINTOcumulative_UV

SELECTdate_stcMAX(time_str)zCOUNT(DISTINCTuserjd)asUV

FROM(

SELECT

DAFEJORMAr^yyyy-MM-dd')asdate_strz

SUBSTR(DATE_FORMAF(ts；HH:mm)L4)||'0,astime_stc

userjd

FROMuser_behavior

)

GROUPBYdate_str

先将每条记录所属的时间窗口字段拼接好，然后再对所有记录按照拼接好的时间窗口

字段，通过GROUPBY做聚合，从而达到近似累积计算的效果。

・1.13版本前的写法有很多缺点，首先这个聚合操作是每条记录都会计算一次。其次,

在追逆数据的时候，消费堆积的数据时，UV大盘的曲线就会跳变。

25

•在1.13版本支持了TVF写法，基于cumulatewindow,我们可以修改为下面的写法,

将每条数据按照EventTime精确地分到每个Window里面,每个窗口的计算通过

watermark触发，即使在追数据场景中也不会跳变。

INSERTINTOcumulative_UV

SELECTWINDOW_end,COUNT(DISTINCTuserjd)asUV

FROMTable(

CUMULATE(Tableuser_behaviocDESCRIPTOR(ts),INTERVAL'1O'MINUTES,INTERVAL'1'DAY))

)

GROUPBYWINDOW_start,WlNDOW_end

UV大盘曲线效果如下图所示：

近实时累计计算

UV大盘曲线

每10分钟统计一次当天的累

计用户UV

■1.3Window性能优化

ApacheFlink1.13社区开发者们对WindowTVF进行了一系列的性能优化，包括:

•内存优化：通过内存预分配，缓存window的数据，通过windowwatermark触

发计算，通过申请一些内存buffer避免高频的访问state;

26

•切片优化：将window切片，尽可能复用已计算结果，inhopwindow,cumulate

window。计算过的分片数据无需再次计算，只需对切片的计算结果进行复用；

•算子优化：window算子支持local-global优化；同时支持count(distinct)自动解

热点优化；

•迟到数据：支持将迟到数据计算到后续分片，保证数据准确性。

Window性能优化

内存优化切片优化算子优化迟到数据

将window切片，

通过内存预分配，支持

尽可能复用已计算window

缓存window的数local-global优化,支持迟到数据计算

结果，如hop

据，通过window同时支持到后续分片，保证

window,

watermark触发count(distinct)自数据准确性

cumulate

计算