关于后端:Flink-SQL-在美团实时数仓中的增强与实践

摘要：本文整顿自美团数据系统研发工程师董剑辉 & 美团数据系统研发工程师张彬，在 Flink Forward Asia 2022 平台建设专场的分享。本篇内容次要分为五个局部：

1. Flink SQL 在美团
2. SQL 作业细粒度配置
3. SQL 作业变更反对从状态复原
4. SQL 正确性问题排查能力建设
5. 将来瞻望

点击查看直播回放和演讲 PPT

一、Flink SQL 在美团

目前 Flink SQL 在美团已有 100+ 业务方接入应用，SQL 作业数也已达到了 5000+，在整个 Flink 作业中占比 35%，同比增速达到了 115%。

SQL 作业的快速增长给咱们带来了许多新的问题和挑战，次要包含以下几点：

• SQL 作业无奈细粒度批改 StateTTL、并发等配置导致资源节约。
• SQL 批改逻辑无奈从原先状态复原。
• SQL 作业呈现数据正确性问题难以排查。

上面将一一介绍这些问题以及如何解决。

二、SQL 作业细粒度配置

目前 Flink 不反对细粒度设置 TTL、算子间分区关系以及并发等配置。尤其是 TTL，在 DataStream 作业中，用户能够依据需要自定义决定状态保留的 TTL 时长，而 Flink SQL 作业目前 TTL 的设置只反对作业粒度，这会造成肯定水平的资源节约，上面咱们来看两个具体的业务示例。

第一个场景，不同算子对状态的保留时长不同。比方该作业的逻辑是去重后进行关联聚合，去重算子只须要设置 1h 的 TTL，而聚合算子要求 1 终日的数据，目前的解决形式只能是全副设置为 1 天，造成资源节约。

第二个场景，Join 算子左右流的业务周期不统一，以及一些非专用的维表须要应用 Regular Join 感知维表的回撤。针对这类场景，目前只能将数据做冷热拆散，别离和实时流以及维表做关联，并做数据去重，大大提高开发运维老本。

这个问题的难点在于 SQL 作业应用的状态对于用户都是黑盒，因而咱们的指标是让用户低门槛地感知并批改 TTL。

咱们最终采取的解决方案是提供一套外置服务 graph-service，也叫可编辑执行打算。在上线前，咱们利用这个服务动态剖析用户作业的拓扑图，并采集展现 TTL，将其凋谢给用户编辑。当用户批改某算子或某流 TTL，再将新的 TTL 配置作为引擎参数传递给 Flink 引擎，加强执行打算。这外面波及的两个外围流程是采集 TTL 和加强执行打算。

首先来看采集 TTL，须要思考两个问题。

• 第一个问题是在哪个阶段采集 TTL 信息。因为 Flink 的 TTL 信息与状态绑定，只有在创立具体的状态描述符时能力通晓，而 Transformation 层无奈得悉作业的状态状况，因而咱们最终决定在 ExecNode 到 Transformation 的转换过程中采集 TTL 信息。
• 第二个问题是怎么标识 TTL。咱们给 ExecNode 减少了如 Transformation ID 一样的标识，并引入了一个工作栈，存储正在被翻译的 ExecNode。每次 ExecNode 在调用 translateToPlanInternal 办法前，咱们获取自增的 ExecNode ID，并将其插入到工作栈中。当 ExecNode 的翻译完结后，从栈顶移除，再建设 Transformation 到 ExecNode，再到 TTL 的映射关系。

其次咱们来看如何加强执行打算，下面是一个模仿的作业拓扑图，通过剖析咱们能采集到 Join 算子和聚合算子的 TTL 信息，以及 Transformation ID 和 ExecNode ID，而后咱们凋谢给用户编辑，假如用户将 Join 算子的 TTL 从 1 天设置为 1 小时。

之后咱们将新的 TTL 配置传入到引擎的 TableConfig 中。当 Join 算子调用 TranslateToPlanInternal 时，它在创立状态读取 TTL 配置的过程中读取工作栈栈顶获取以后正在被翻译的 ExecNode，从而失去 ExecNode ID，再从 TableConfig 中读取对应的 TTL 配置，笼罩默认配置。

上图是第一个场景的作业测试成果。

首先不应用咱们的能力，默认将作业整体的 TTL 设置为 1 天。能够发现在高峰期均匀每小时的 Container CPU 使用率回升到了 107%，即便在低峰期也有 67.3%，且过程中算子呈现了几次反压。

之后咱们应用精细化设置 TTL 能力，将去重算子设置为 1 小时，其余有状态算子 TTL 仍保留 1 天，在高峰期均匀每小时的 Container CPU 使用率仅有 14.8%，而在低峰期则为 7.54%，且过程中无任何反压。除此之外，Checkpoint 大小也从 8.54G 升高到了 1.8G。

接下来是可编辑执行打算提供的其余能力优化。首先是分区关系，作业内上下游算子连接数过多，会占用较大的 Network buffer 内存，从而影响作业的失常启停，基于可编辑执行打算能力，咱们能够手动将 Rebalance 边批改为 Rescale。

比方上图的示例，右边上游算子有 2000 个并发，而上游的 Sink 算子只有 1000 个并发。在这种场景下，Flink SQL 会默认生成 Rebalance 的连贯形式，共需 2000*1000，共 200 万个逻辑连贯。

通过可编辑执行打算能力，咱们手动将 Rebalance 设置为 Rescale 后，它只须要 2000 个连贯，大大降低了 Network buffer 的内存需要。

除此之外，咱们基于可编辑执行打算还提供了以下三种能力：

• 反对独自批改算子并发并从状态复原。
• 反对独自批改算子的 slotSharingGroup。
• 反对批改 ChainStrategy 并从状态复原。

三、SQL 作业变更反对从状态复原

目前 Flink SQL 的状态复原机制较为严苛，在很多场景下，作业变更无奈从原先状态复原，造成了大量的资源节约和运维老本。针对这个问题，咱们对状态迁徙这个问题域做了具体的场景剖析。

Flink SQL 作业变更能够分为两类场景，版本升级（Upgrade）和同版本内的作业降级（Migration）。进一步地，咱们聚焦实时数仓下的 Migration 场景，大抵能够分为三种，Graph Migration、Operator Migration、SavepointMigration。

• Graph Migration：作业的变更仅产生在 Pipleline 的拓扑构造层面，即节点与边的属性发生变化。这类场景咱们能够通过分享的第一个工作可编辑执行打算来解决。
• Operator Migration：作业变更仅产生在算子状态层级，DAG 不发生变化。这类场景包含新增了一些聚合指标、关联新增属性等等。
• SavepointMigration：作业在 DAG 层面和算子状态同时产生变更。对应的场景是用离线数据为新工作初始化状态。

咱们能够看到 SavepointMigration 次要是 Graph Migration 和 Operator Migration 的复合场景。因而咱们本次分享次要聚焦 Operator Migration 场景。通过咱们针对 Graph Migration 和 Operator Migration 实现的能力的组合，咱们打算在后续的工作中再去欠缺 SavepointMigration。

联合美团目前的业务需要，以后最为迫切的场景是用户在 DWD 层对于宽表加工场景对事实表关联须要加属性，以及 DWS 层须要减少聚合指标，即前文所探讨的 Operator Migration 场景。

首先，咱们定义了一个名为 KeyedStateMetadata 的数据结构用来标识每个 KeyedState 的元数据，在创立算子时咱们将状态元数据信息（KeyedStateMetatda）注入到动态的上下文中，并在解析用户作业执行图时获取。在之后的迁徙状态过程中，咱们联合状态元数据和 State-Process-API 来创立一个新的 Savepoint，实现状态迁徙。

上图是一个迁徙聚合算子的示例。首先咱们收集聚合算子的 KeyedStateMetadata，而后读取老的 Savepoint，利用 State-Process-API 将当中的状态进行转换。最初咱们将新的状态 dump 到新的 Savepoint 中，并让这个作业从新的 Savepoint 中复原。

上面咱们来介绍一下 KeyedStateMetadata 的构造。首先一个算子可能有一个或多个 Keyed State，而每个 Keyed State 都会对应一个 KeyedStateMetadata，针对每个 KeyedStateMetadata 咱们会存储的状态元数据信息包含状态名、状态的数据类型、TTL 配置、自定义的 StateContext 接口。其中 StateContext 是用来检测两个状态 Schema 是否兼容的一个接口设计。

上图右侧是一个具体的示例。首先看 AppendOnlyTopNFunction，咱们会采集到它的状态名 data-state-with-append、它的两个数据类型以及它的 TTL 是 1 天，最终咱们会为它建设一个叫做 RankStateContext 的上下文构造，用来在状态兼容性校验中检测它的状态 Schema 是否与其余状态兼容。

通过采取状态元数据信息和 State-Process-API，咱们解决了状态迁徙的技术难点，但为了定义清晰的状态迁徙能力边界以及防止用户在未反对的场景下应用状态迁徙能力失败，造成的资源节约、运维老本，咱们提供了事先剖析的能力。

事先剖析能力能够分为以下三层校验：

• SQL 层应用 AST 做业务逻辑兼容性校验。
• 基于可编辑执行打算做拓扑逻辑兼容性校验。
• 状态 Schema 兼容性校验。

这里补充一下为什么咱们须要业务逻辑兼容性校验，因为状态 Schema 的兼容校验更多是基于底层的技术能力的视角，它自身不具备可辨认业务语义的特色，比方 sum 和 max 对应到状态上的数据类型是一样的，但在业务语义上这两者是齐全无奈兼容的，因而咱们在这里减少了对业务逻辑进行兼容性校验的补充，确保用户会用、用对状态迁徙能力。

那么基于后面的三层校验，咱们一共会有四种剖析后果，别离对应的技术语义和业务语义如下：

• COMPATIBLE_AS_IS，指作业能够间接从老状态复原，对应的含意是新老作业是没有产生任何变动。
• COMPATIBLE_AFTER_RENAME，指作业通过调整 Operator ID 后可从老状态复原。它对应的业务场景是批改算子并发或者调整作业的 chain 逻辑等。
• COMPATIBLE_AFTER_MIGRATION，指作业不可间接从老状态复原，必须通过状态迁徙制作新状态后，可从新状态复原。对应的场景是新增聚合、去重以及 Join 等算子指标或者字段，也是咱们本次分享重点所解决的场景。
• INCOMPATIBLE，指作业的新老状态齐全不兼容，且无奈通过迁徙制作任何新状态。对应的场景是其余 SQL 逻辑改变，如替换指标程序，增减算子以及一些咱们可能仍未反对状态迁徙的场景。这个也是咱们后续对状态迁徙须要欠缺的工作方向之一。

上面咱们来具体介绍一下事先剖析的测验流程。

首先新老作业的 SQL，咱们将它解析失去 AST。而后须要确保新作业的指标业务语义是向后兼容老作业的，比方指标程序调换，这个就是在这一层进行校验。如果发现不兼容，会间接返回 INCOMPATIBLE 的校验后果。

之后利用 graph-service 将它翻译成 MTJsonPlan，而后校验算子个数是否不统一，以及作业拓扑图是否产生了变动。如果这两个有任意条件没有通过，都会返回 INCOMPATIBLE 的测验后果。如果这两个后果都通过，咱们会计算失去新老算子的映射 Map，并对这个映射 Map 中的每一对新老算子查看是否都有状态或是否都有 TTL，以及是否能够通过状态迁徙的能力进行复原。如果当中任意条件不满足，都会返回 INCOMPATIBLE 的测验后果。

当以上条件都满足后，咱们会测验它的新老算子状态是否须要迁徙。如果须要迁徙，咱们会返回 COMPATIBLE_AFTER_MIGRATION 的后果。而如果新老算子的状态不须要迁徙就能够复原的话，咱们会再进一步校验它的 Operate ID 是否产生了变动。如果产生了变动，咱们会返回 COMPATIBLE_AFTER_RENAME 的校验后果。如果没有发生变化，咱们会认为这个作业的剖析后果是 COMPATIBLE_AS_IS，即这个作业和老作业没有任何变动。

上图是咱们的产品示例，在抉择制作新的 Savepoint 迁徙时会进行事先剖析进行校验，以上是校验后果不统一的状况，因为我对新老作业做了替换指标程序的批改。

最初总结一下针对状态迁徙该问题域咱们所做的工作。首先咱们遇到了什么问题呢？

咱们遇到的问题是 Flink SQL 原生提供的状态恢复能力较弱，无奈反对作业变更。在美团实时数仓场景下，SQL 作业须要减少聚合指标或去重关联字段时无奈从原先状态复原，给用户的作业迭代造成了许多艰难。

针对这个问题，首先咱们对状态迁徙的问题域进行了具体的剖析，细分了场景，并联合美团现状聚焦于 Migration 场景，反对了聚合减少指标、事实表关联以及去重、排序等场景下减少字段的状态迁徙能力。

在此基础上，咱们针对生产环境提供了事先剖析能力，确保用户会用且用对状态迁徙能力，防止无意义的资源节约和运维老本。

四、SQL 正确性问题排查能力建设

美团在大力推广 Flink 作业 SQL 化，咱们在运维业务 Flink SQL 作业时遇到了三类问题，别离为丢数问题、乱序问题及 FlinkSQL 使用不当导致的正确性问题。因短少辅助工具，无奈疾速定位出问题，影响线上业务无奈失常数据生产，妨碍 Flink 作业 SQL 化过程。

对于业务的同学来讲是如何验证 Flink SQL 作业正确性的问题呢？有以下三种路径：

• Flink SQL 作业与已有的自研零碎后果比照。
• 通过实时作业与离线的作业后果比照。
• Flink SQL 作业主备链路双跑后果比照。

通过以上三种形式，当业务发现 Flink SQL 作业有正确性的问题时，又面临了以下三大痛点问题。

• 排查门槛很高，对于业务同学来讲不理解 Flink SQL 底层原理，对于平台同学来讲不理解用户业务。呈现正确性的问题后，无从下手。
• 排查定位周期长，因为没有可借助的工具，所以须要破费几天甚至更长时间定位问题。
• 重大影响线上业务数据失常产出，用户不得不将 SQL 作业从新迁回到原来的作业上，这大大妨碍了 Flink 作业 SQL 化的过程。

为了解决用户的痛点问题，咱们须要一套辅助零碎。因为 Flink SQL 的作业只有最终后果，短少两头记录过程。所以如果能记录下每一条数据在 Flink SQL 各个算子中的流转过程，对于排查定位问题是十分有帮忙的，就像监控设施一样，能够通过回放监控录像来排查定位问题，基于这样的思路，咱们就开发了一套辅助零碎。

在开发这套零碎之前，咱们对业界相干的产品做了简略的调研。发现目前在分布式场景下故障排查有成熟的 Trace 零碎，与咱们要定位排查 Flink SQL 正确性问题十分相似。

接下来咱们先来简略理解下 Trace 零碎的原理。如左上图所示，通过一次残缺的 rpc 调用，通过 A、B、C、D、E 五个微服务中的拜访，将服务依赖调用的 16 条记录都保留下来。Trace 零碎会给每条残缺的链路一个惟一 Trace ID 作为标记。通过 Trace ID 这个标记就能够关联起一条残缺的调用链路。

上图中左下角局部示意是一个残缺的 Trace 零碎所须要的三个重要环节，别离是数据埋点上报，数据收集剖析，数据后果展现，这也是 Flink SQL 排查工具所须要的三个环节。

简略理解了 Trace 零碎之后，接下来咱们将 Trace 零碎与 Flink SQL 排查工具所须要的能力做下比照。

• Trace 零碎的一次 rpc 调用具备全局关联性，而对于 Flink SQL 来说只能做到同一个 Task 间部分关联性。
• Trace 零碎中数据上报须要业务在要害办法中手动埋点，然而对于 Flink SQL 来说手动埋点代价极高，咱们冀望与 Flink 引擎解耦，不便当前 Flink 版本升级保护。
• Trace 零碎中数据量大，容许局部数据失落，而 Flink SQL 排查工具是不容许数据失落的，心愿能反对打印局部算子的输入输出。
• Trace 零碎中的数据有全局关联性能够做到主动归因，在 Flink 中数据没有全局关联性，只能手动剖析不能做到主动归因。

通过比照发现 Trace 零碎不适用于 Flink SQL 正确性问题排查，须要基于以上内容定制开发。

在解说 Flink SQL 排查零碎之前，咱们简略回顾下 Flink 相干的知识点。

首先是 Flink SQL 算子摸底，Flink SQL 波及到的算子有 30+ 个，篇幅的起因我这里只列出了局部算子，算子十分多而且有些算子是通过 codegen 代码生成技术实现的。很显然咱们要在 Flink SQL 算子埋点，开发成本很高。

咱们留神到这些算子有一个独特的特点就是都继承与 AbstractStreamOperator，而该类中有对 record 及 watermark 解决的要害办法，比方 setKeyContextElement1/2 与 processWatermark1/2。

这部分是 Task 启动后数据是如何流转到 Operator 的过程。通过 MailboxProcessor 循环调用获取数据并将数据最终传递 OperatorChain，OperatorChain 将数据交由第一个 Operator 解决，也就是由 mainoperator 解决，这里就开始调用下面介绍 setKeyContextElement 及 processWatermark1 那几个办法，接下来咱们看下这几个办法是如何在 OperatorChain 调用的。

通过下面这张流程图咱们发现，数据被解决之前都要通过 setKeyContextElement1/2 办法，数据流转到下一个 OperatorChain 时要调用 pushToRecordWriter 办法，对于 watermark 解决也相似。所以对这几个要害的办法进行监听，就能管制算子的输入输出了。

有了下面的几个要害办法还不够，还须要解决数据解析的问题。这里能够看到 SQL 转换成 StreamGraph 后，StreamGraph 中的每个 StreamNode 都记录该算子的输出及输入类型信息。对于 Flink SQL 来说，数据传输类型都是序列化后的 rowdata。有了数据的类型信息，算子间传输的数据能失常解析进去吗？答案是必定的，我这里先留一个疑难，前面在整体架构中会重点讲这部分的内容。

通过以上的技术剖析，咱们更加动摇应用字节码加强技术将数据解析过程与 Flink 引擎分来到，以不便前面 Flink 版本的保护降级，接下来具体介绍下整体架构及实现细节。

接下来看下整体架构，有五个局部组成。

• 平台入口，用户在平台开启 Flink SQL Debug 性能调试作业，抉择要输入数据的算子 ID 后，而后提交作业。
• TM 启动时，数据监听程序监听 Flink SQL 要害办法，解析算子的输入输出数据，图中的小齿轮代表着解析数据的 javaagent 程序。
• 将解析后的数据同步发送到 Kafka 中。
• 通过工具将 Kafka 中的数据同步到 OLAP 引擎中。
• 最初通过查问剖析 OLAP 引擎 中的数据，排查定位问题。

首先是平台入口。从上图左侧能够看到，须要关上输出算子粒度的明细开关。另外，须要抉择要在哪些算子上打印它的输入输出数据，有了这部分内容之后提交作业。

作业提交到 Yarn 之后，怎么监听算子数据呢？咱们采纳是 Byte Buddy 字节码加强框架实现对 Flink 算子数据的解析，通过监听以上要害的办法达到与 Flink 引擎解耦的目标，左边的图是数据解析程序输入的内容。上面对 Value 及 input_order 具体介绍一下。

在后面介绍中咱们留有一个疑难，streamnode 中保留了输入输出类型信息，如何解析出数据呢？对于 Flink SQL 来说算子间传输的是序列化后的 Rowdata，能够通过固定办法通过传递类型及字段索引参数，调用 getField 办法就能够解析出数据了。

这里只有解析后的数据，只有值没有字段信息，为什么要与字段信息管理起来呢？因为解析后的数据中有可能有多个雷同的值，为了精准检索后果，须要将字段与数值对应起来。怎么获取字段信息并将数据与字段信息关联起来呢？在 SQL 转换成 Transformation 过程中，ExecNode 类中有个重要的 TranslateToPlan 办法，须要在该办法上加强，将算子的输入输出字段解析后保留在 StreamConfig 中，在解析程序中就能够将字段与数据关联起来了。

这里有一个难点，对于一般 Flink SQL 是 OK 的，但对于数据湖场景的 Flink SQL 算子之间的数据传递不仅仅有序列化后的 RowData，还有 Kryo 类型的数据，如 HoodieRecord，HoodieRecord 的解析须要依赖 Hudi Schema 信息。这里在解析程序中是无奈获取到的，有一个简略奇妙的方法，就是调用 HoodieRecord 的 toString 办法，在 toString 办法中让 Hudi 本身去解析，就能够高效灵便的解决数据湖场景下的数据解析的问题了。

介绍完数据解析，接下来介绍的是数据在 Task 中关联性的问题，咱们在字段中通过 input_order 记录了某个 sub_task 粒度的 record 输出程序编号，用于标记数据在一个 Subtask 中的关联性。

为什么要设计这个字段，是因为 chain 在一起的算子的字段可能不一样，比方 chain 在一起的有五个 Operator 前两个 Operator 都有 ID 字段，后三个 Operator 没有 ID 字段，如果依据 ID 查问，后三个 Operator 就没有数据，为了展现一条数据在 Subtask 中残缺链路，所以指定同一个 Subtask 同一个 input_order 就能够说筛选出残缺的数据链。前面 Case 剖析也有介绍 input_order。

对于数据关联性有以下三种状况：

• 对于同步算子之前数据传输，当数据通过 Subtask 所有的 Operator 解决后能力解决下一条数据。当数据进入 Subtask 的第一个 Operator 时 input_order +1，前面的 Operator 应用第一个 Operator 的 input_order，这些 Operator 的 input_order 要么一样或者要么局部为空。
• 对于开启了 mini-batch 性能后的算子，算子会攒批后处理，这一批数据也有关联性。
• 对于 LookupJoin，在同步状况下跟第一种状况是一样的，异步状况下算子间有雷同的字段，能够通过该字段来关联算子关系。

以上的内容是如何解析数据，接下来介绍如何输入数据到 Kafka 中。为了防止冗余数据输入，默认状况下只打印除 Source 以外的所有算子的输出数据，以及除了 Sink 以外的所有 Task 中 tailOperator 的输入数据（当上游算子呈现乱序时，能够追溯到上游算子数据状况），用户也能够只抉择输入局部算子的数据。

有了算子的输入输出标准之后，上面介绍数据是如何收集到 Kafka 中的。第一种计划是，把数据输入到日志文件中，通过日志收集的形式输入到 Kafka，通过测试报告发现，写入速度顶峰时能够达到 600-800Mb/s，目前 TM 日志采纳 rolling 形式保留，写入速度远远高于收集速度，有数据失落的危险。所以该收集日志的形式不符合要求。

所以咱们采纳了第二种计划，将算子的数据间接通过 socket 形式同步输入到 Kafka 中，如果写入的数据很快，收集的慢就会对上游算子产生反压，限度上游算子的解决发送数据的速率，既能保障业务的 SQL 能够顺利执行又能保障算子输入的数据不失落保障残缺的数据收集到 Kafka 中。

上面介绍下咱们应用 Flink SQL 排查工具帮忙业务解决问题的三类 Case。

Case1：Flink SQL 本身 bug 导致的正确性问题。如上图所示，用户的 SQL 很简略就是个 Deduplicate 去重的 SQL 作业。

景象：存在丢数景象，丢数的 ID 不固定且无规律，然而用户能够提供丢数的 ID。

论断：通过排查发现，其实是 Flink bug 导致的，就是 localtimestamp 函数存在精度上的 bug，在应用 to_date(cast(localtimestamp as varchar)，‘yyyy-MM-hh HH:mm:ss.SSS’) 时，当工夫为整秒（2022-05-01 12:10:10.000）时 to_date 函数解析失败，不满足条件，导致数据失落，Flink SQL 作业却失常运行。

通过 Flink SQL 排查工具，指定 ID，输出上面的查问 SQL，发现 Calc 只有输出数据，没有输入数据，断定在 Calc 算子中将该 ID 的值过滤掉了。在对照 calc codegen 代码逻辑及 Flink 代码发现在整点时存在 bug，导致数据失落。解决这个 bug 就很简略了，只须要在 Flink 中实现对整点的精度解决的逻辑就行了。

下面的查问中 Calc 有 ID 字段，当不存在 ID 字段时，这个时候就须要用 input_order 来关联一条数据在整个 Subtask 中的数据链了。如上图左下角的 SQL，指定 subtask id operator id 及 input_order 来查问这条数据的残缺数据链。

Case2：Flink SQL 设计缺点导致的正确性问题（乱序）。与 Flink SQL 本身 bug 的区别是，Flink SQL 本身 bug 是指，Flink SQL 能解决某类问题，然而有 bug，Flink SQL 设计缺点是指 Flink SQL 解决不了某类问题。

景象：用户 SQL 作业后果乱序导致后果不对。

论断：Flink SQL Join 左右流 一对多关系，右流应用的是 NoUniqueKey，NoUniqueKey 应用的是 MapState，而 MapState 无奈保证数据程序，所以查问这类后果会有乱序的状况。除了此类问题，Flink SQL 中如果存在屡次 Keyby 并且 Key 字段不统一也会导致乱序问题。

Case3：Flink SQL 使用不当导致的正确性问题。这类 Case 也十分常见。

景象：用户 SQL 作业丢数导致后果不对。

论断：通过工具排查，发现用户设置 State TTL 是 2 个小时，理论有超过了 2 小时的数据过去，状态过期，数据关联不上，丢数导致后果不对。除了 State TTL 设置不对的状况，还有业务本身逻辑，SQL 表白等应用问题。

通过应用这个工具之后，有时候也能够证实 Flink SQL 作业没有问题，而是比照作业有问题。有了该工具后排查问题时长从天级别升高到了小时，甚至分钟级别，大大缩短了排查故障时长，失去了用户的认可与信赖，为 Flink 作业 SQL 化过程保驾护航。

五、将来瞻望

将来瞻望次要分为以下三局部：

• Flink SQL 细粒度配置

  • 在细粒度资源管理上，目前细粒度资源管理只反对 API 设置，所以也须要在 SQL 场景通过 Flink SQL 灵便配置的性能反对细粒度的资源管理。
  • Flink SQL 灵便配置联合 Flink autopilot 机制搭配应用，使得 SQL 作业能主动调整到比拟现实状态。

• Flink SQL State

  • 心愿 Flink SQL State 具备可查问的能力。
  • 摸索 SQL 扭转后反对以懒迁徙形式从状态复原。

• Flink SQL 排查工具

  • 心愿依据积攒的教训，对 Flink SQL 反对上线前危险提醒。
  • 解决发现的已知乱序及性能问题。

点击查看直播回放和演讲 PPT

更多内容

流动举荐

阿里云基于 Apache Flink 构建的企业级产品 - 实时计算 Flink 版现开启流动：
99 元试用 实时计算 Flink 版（包年包月、10CU）即有机会取得 Flink 独家定制卫衣；另包 3 个月及以上还有 85 折优惠！
理解流动详情：https://www.aliyun.com/product/bigdata/sc