Flink SQL 在美团实时数仓中的增强与实践

2023-04-23 17:58:39

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 字段不一致也会导致乱序问题。