2018 年 12 月 Blink 宣布开源，经历了约一年的时间 Flink 1.9 于 2019 年 8 月 22 发布。在 Flink 1.9 发布之前字节跳动内部基于 master 分支进行内部的 SQL 平台构建。经历了 2~3 个月的时间字节内部在 19 年 10 月份发布了基于 Flink 1.9 的 Blink planner 构建的 Streaming SQL 平台，并进行内部推广。在这个过程中发现了一些比较有意思的需求场景，以及一些较为奇怪的 BUG。

虽然最新的 Flink 版本已经支持 SQL 的 DDL，但 Flink 1.9 并不支持。字节内部基于 Flink 1.9 进行了 DDL 的扩展支持以下语法：

同时 Flink 1.9 版本不支持的 watermark 定义在 DDL 扩展后也支持了。

我们在推荐大家尽量的去用 SQL 表达作业时收到很多“SQL 无法表达复杂的业务逻辑”的反馈。时间久了发现其实很多用户所谓的复杂业务逻辑有的是做一些外部的 RPC 调用，字节内部针对这个场景做了一个 RPC 的维表和 sink，让用户可以去读写 RPC 服务，极大的扩展了 SQL 的使用场景，包括 FaaS 其实跟 RPC 也是类似的。在字节内部添加了 Redis/Abase/Bytable/ByteSQL/RPC/FaaS 等维表的支持。

同时还实现了多个内部使用的 connectors：

并且为 connector 开发了配套的 format：PB/Binlog/Bytes。

除了对 Flink 本身功能的扩展，字节内部也上线了一个 SQL 平台，支持以下功能：

除了对功能的扩展，针对 Flink 1.9 SQL 的不足之处也做了一些优化。

1）支持了 window Mini-Batch

Mini-Batch 是 Blink planner 的一个比较有特色的功能，其主要思想是积攒一批数据，再进行一次状态访问，达到减少访问状态的次数降低序列化反序列化的开销。这个优化主要是在 RocksDB 的场景。如果是 Heap 状态 Mini-Batch 并没什么优化。在一些典型的业务场景中，得到的反馈是能减少 20~30% 左右的 CPU 开销。

2）扩展 window 类型

目前 SQL 中的三种内置 window，滚动窗口、滑动窗口、session 窗口，这三种语意的窗口无法满足一些用户场景的需求。比如在直播的场景，分析师想统计一个主播在开播之后，每一个小时的 UV(Unique Visitor)、GMV(Gross Merchandise Volume) 等指标。自然的滚动窗口的划分方式并不能够满足用户的需求，字节内部就做了一些定制的窗口来满足用户的一些共性需求。

3）window offset

这是一个较为通用的功能，在 Datastream API 层是支持的，但 SQL 中并没有。这里有个比较有意思的场景，用户想要开一周的窗口，一周的窗口变成了从周四开始的非自然周。因为谁也不会想到 1970 年 1 月 1 号那天居然是周四。在加入了 offset 的支持后就可以支持正确的自然周窗口。

1）延迟 Join

维表 Join 的场景下因为维表经常发生变化尤其是新增维度，而 Join 操作发生在维度新增之前，经常导致关联不上。 

所以用户希望如果 Join 不到，则暂时将数据缓存起来之后再进行尝试，并且可以控制尝试次数，能够自定义延迟 Join 的规则。这个需求场景不单单在字节内部，社区的很多同学也有类似的需求。

基于上面的场景实现了延迟 Join 功能，添加了一个可以支持延迟 Join 维表的算子。当 Join 没有命中，local cache 不会缓存空的结果，同时将数据暂时保存在一个状态中，之后根据设置定时器以及它的重试次数进行重试。

2）维表 Keyby 功能

通过拓扑我们发现 Cacl 算子和 lookUpJoin 算子是 chain 在一起的。因为它没有一个 key 的语义。 

当作业并行度比较大，每一个维表 Join 的 subtask，访问的是所有的缓存空间，这样对缓存来说有很大的压力。

但观察 Join 的 SQL，等值 Join 是天然具有 Hash 属性的。直接开放了配置，运行用户直接把维表 Join 的 key 作为 Hash 的条件，将数据进行分区。这样就能保证下游每一个算子的 subtask 之间的访问空间是独立的，这样可以大大的提升开始的缓存命中率。

除了以上的优化，还有两点目前正在开发的维表优化。

目前 Flink 支持的三种 Join 方式；分别是 Interval Join、Regular Join、Temporal Table Function。

前两种语义是一样的流和流 Join。而 Temporal Table 是流和表的的 Join，右边的流会以主键的形式形成一张表，左边的流去 Join 这张表，这样一次 Join 只能有一条数据参与并且只返回一个结果。而不是有多少条都能 Join 到。

它们之间的区别列了几点：

可以看到三种 Join 方式都有它本身的一些缺陷。

对于以上的不足之处字节内部都做了对应的修改。

对于 SQL 作业来说一旦发生条件变化都很难从 checkpoint 中恢复。

SQL 作业确实从 checkpoint 恢复的能力比较弱，因为有时候做一些看起来不太影响 checkpoint 的修改，它仍然无法恢复。无法恢复主要有两点；

例子1：并行度发生修改导致无法恢复。

source 是一个最常见的有状态的算子，source 如果和之后的算子的 operator chain 逻辑发生了改变，是完全无法恢复的。 

下图左上是正常的社区版的作业会产生的一个逻辑， source 和后面的并行度一样的算子会被 chain 在一起，用户是无法去改变的。但算子并行度是常会会发生修改，比如说 source 由原来的 100 修改为 50，cacl 的并发是 100。此时 chain 的逻辑就会发生变化。

针对这种情况，字节内部做了修改，允许用户去配置，即使 source 的并行度跟后面整体的作业的并行度是一样的，也让其不与之后的算子 chain 在一起。

例子2：DAG 改变导致无法恢复。

这是一种比较特殊的情况，有一条 SQL (上图)，可以看到 source 没有发生变化，之后的三个聚合互相之间没有关系，状态竟然也是无法恢复。

作业之所以无法恢复，是因为 operator ID 生成规则导致的。目前 SQL 中 operator ID 的生成的规则与上游、本身配置以及下游可以 chain 在一起的算子的数量都有关系。因为新增指标，会导致新增一个 Calc 的下游节点，进而导致 operator ID 发生变化。 

为了处理这种情况，支持了一种特殊的配置模式，允许用户配置生成 operator ID 的时候可以忽略下游 chain 在一起算子数量的条件。

例子3：新增聚合指标导致无法恢复。

这块是用户诉求最大的，也是最复杂的部分。用户期望新增一些聚合指标后，原来的指标要能从 checkpoint 中恢复。

可以看到图中左部分是 SQL 生成的算子逻辑。count，sum，sum，count，distinct 会以一个 BaseRow 的结构存储在 ValueState 中。distinct 比较特殊一些，还会单独存储在一个 MapState 中。 

这导致了如新增或者减少指标，都会使原先的状态没办法从 ValueState 中正常恢复，因为 VauleState 中存储的状态 “schema” 和新的（修改指标后）的 “schema”不匹配，无法正常反序列化。

在讨论解决方案之前，我们先回顾一下正常的恢复流。先从 checkpoint 中恢复出状态的 serializer，再通过 serializer 把状态恢复。接下来 operator 去注册新的状态定义，新的状态定义会和原先的状态定义进行一个兼容性对比，如果是兼容则状态恢复成功，如果不兼容则抛出异常任务失败。 

不兼容的另一种处理情况是允许返回一个 migration（实现两个不匹配类型的状态恢复）那么也可以恢复成功。

针对上面的流程做出对应的修改：

通过以上的修改基本就可以做到正常的，新增的聚合指标从拆开的方案恢复。

字节跳动内部对流批一体和业务推广之前，技术团队提前做了大量技术方面的探索。整体判断是 SQL 这一层是可以做到流批一体的语义，但实践中却又发现不少不同。

比如说流计算的 session window，或是基于处理时间的 window，在批计算中无法做到。同时 SQL 在批计算中一些复杂的 over window，在流计算中也没有对应的实现。

但这些特别的场景可能只占 10% 甚至更少，所以用 SQL 去落实流批一体是可行的。

这张图是比较常见的和大多数公司里的架构都类似。这种架构有什么缺陷呢？

鉴于上面的问题，提出了基于 Flink 的流批一体架构来解决。

基于 Flink 实现的流批一体架构：

• 统一的 SQL：通过一套 SQL 来表达流和批计算两种场景，减少开发维护工作。

• 复用 UDF：流式和批式计算可以共用一套 UDF。这对业务来说是有积极意义的。

• 引擎统一：对于业务的学习成本和架构的维护成本都会降低很多。

• 优化统一：大部分的优化都是可以同时作用在流式和批式计算上，比如对 planner、operator 的优化流和批可以共享。

什么是 retract 放大？

上图有 4 张表，第一张表进行去重操作 (Dedup)，之后分别和另外三张表做  Join。逻辑比较简单，表 A 输入(A1)，最后产出 (A1,B1,C1,D1) 的结果。

当表 A 输入一个 A2，因为 Dedup 算子，导致数据需要去重，则向下游发送一个撤回 A1 的操作 -(A1) 和一个新增 A2 的操作 +(A2)。第一个 Join 算子收到 -(A1)  后会将 -(A1) 变成 -(A1,B1) 和 +(null,B1)(为了保持它认为的正确语义) 发送到下游。之后又收到了 +(A2) ，则又向下游发送 -(null,B1) 和 +(A2,B1) 这样操作就放大了两倍。再经由下游的算子操作会一直被放大，到最终的 sink 输出可能会被放大  1000 倍之多。

如何解决？ 

将原先 retract 的两条数据变成一条 changelog 的格式数据，在算子之间传递。算子接收到 changelog 后处理变更，然后仅仅向下游发送一个变更 changelog 即可。

1.功能优化

2.业务扩展