嗨，大家好，我是老阳。今天给大家带来一篇关于Flink SQL流计算核心思想设计的文章。

在小伙伴看下面之前，先看看这篇文章的整体思路。如果你按照博主的思路走会更清楚：

首先，分析一下将SQL应用于流处理的思想。SQL在批处理方面的应用已经非常成熟。通过比较流批处理在输入、数据处理、输出方面的异同，分析将SQL应用于流处理时需要解决的核心问题。分析如何使用SQL动态输入表技术将输入数据流映射到SQL中的输入表。分析如何使用SQL连续查询技术将SQL中的计算逻辑映射到操作语义。使用SQ L动态表连续查询技术的两种技术方案进行流SQL实际应用到两个常见案例来分析两种类型的SQL连续查询：更新查询追加查询分析如何在SQL Blogger中使用SQL动态输出表技术将输出数据流映射到输出表认为看完这一节应该掌握：

SQL动态输入表和SQL动态输出表SQL连续查询的两种类型分别对应查询场景和SQL语义

SQL 动态表连续查询

。在流式SQL诞生之前，所有基于SQL的数据查询都是基于批量数据的，并不存在将SQL应用于流式数据处理的说法。

所以如果要将SQL应用于流处理，就必须站在巨人的肩膀上(批量数据处理的过程)，那么具体的分析思路如下：

第一步：首先比较批处理和流处理的异同：如果有相同的部分，可以直接重用；不同的是我们需要克服和注意的。第二步：挑出1中提到的差异，如果要满足这种差异，分析一下目前有哪些技术是相似的。第三步：进一步开发这些类似的技术，以满足将SQL应用于流式任务的要求。博主将按照以上三个步骤一步步介绍动态表诞生的背景以及这个概念是如何诞生的。

00-1010首先比较常见的批处理和流处理中数据源(输入表)、处理逻辑和数据宿(结果表)的异同。

-输入表处理逻辑结果表批处理静态表：输入数据有限，是一个有界集批处理计算：每次执行一个查询，可以访问完整的输入数据，然后计算完整的结果数据。静态表：数据限流处理动态表：输入数据不限，数据实时增加，连续流计算：执行时无法访问完整的输入数据，每次计算的结果都是中间结果动态表：数据不限。

对比上面的流批处理，我们得到了将SQL应用于流任务需要解决的三个核心点：

SQL输入表：分析如何将实时、连续的输入流数据表示为SQL中的输入表。SQL处理计算：分析什么样的底层处理技术可以将SQL查询逻辑翻译成实时处理流式输入数据，进而产生流式输出数据。SQL输出表：分析如何在SQL中将SQL查询输出的连续流数据表示为输出表。综合以上三点，本节介绍动态表和连续查询两种技术方案：

动态表：连续的输入输出数据映射到动态表；连续查询：输入数据和输出数据的实时处理技术

1.SQL 应用于流处理的思路

动态表。这里的动态实际上是与批处理的静态(有界)相比较的。

静态表：应用于批量数据，静态表可以理解为不随时间实时变化。通常，一个新分区以一天一小时的间隔生成。动态表：动态表随时间实时变化。将SQL系统中的表的概念应用到Flink中是核心点。看一个具体案例。下图显示了click事件流(左)如何转换为动态表(右)。当数据源生成更多的点击事件记录时，映射的动态表也会增长。这是动态表的概念：

动态表

00-1010连续查询。

一些高级关系数据库系统提供了一个称为物化视图的特性。

物化视图实际上是一个SQL查询，就像传统的虚拟视图一样。然而，与虚拟视图不同，物化视图缓存查询的结果，因此在请求访问视图时不需要重新查询。

计算，可以直接获取物化视图的结果，小伙伴萌可以认为物化视图其实就是把结果缓存了下来。

举个例子：批处理中，如果以 Hive 天级别的物化视图来说，其实就是每天等数据源 ready 之后，调度物化视图的 SQL 执行然后产生新的结果提供服务。那么就可以认为一条表示了输入、处理、输出的 SQL 就是一个构建物化视图的过程。

映射到我们的流任务中，输入、处理逻辑、输出这一套流程也是一个物化视图的概念。相比批处理来说，流处理中，我们的数据源表的数据是源源不断的。那么从输入、处理、输出的整个物化视图的维护流程也必须是实时的。

因此我们就需要引入一种实时视图维护（Eager View Maintenance）的技术去做到：一旦更新了物化视图的数据源表就立即更新视图的结果，从而保证输出的结果也是最新的。

这种实时视图维护（Eager View Maintenance）的技术就叫做连续查询。

注意：

⭐ 连续查询（Continuous Query）不断的消费动态输入表的的数据，不断的更新动态结果表的数据。

⭐ 连续查询（Continuous Query）的产出的结果 = 批处理模式在输入表的上执行的相同查询的结果。相同的 SQL，对应于同一个输入数据，虽然执行方式不同，但是流处理和批处理的结果是永远都会相同的。

5.SQL 流处理实际应用：动态表 & 连续查询技术的两个实战案例

总结前两节，动态表 & 连续查询两项技术在一条流 SQL 中的执行流程总共包含了三个步骤，如下图及总结所示：

Query

⭐ 第一步：将数据输入流转换为 SQL 中的动态输入表。这里的转化其实就是指将输入流映射（绑定）为一个动态输入表。上图虽然分开画了，但是可以理解为一个东西。⭐ 第二步：在动态输入表上执行一个连续查询，然后生成一个新的动态结果表。⭐ 第三步：生成的动态结果表被转换回数据输出流。

我们实际介绍一个案例来看看其运行方式，以上文介绍到的点击事件流为例，点击事件流数据的字段如下：

[ user: VARCHAR, // 用户名 cTime: TIMESTAMP, // 访问 URL 的时间 url: VARCHAR // 用户访问的 URL]

⭐ 第一步，将输入数据流映射为一个动态输入表。以下图为例，我们将点击事件流（图左）转换为动态表 (图右)。当点击数据源源不断的来到时，动态表的数据也会不断的增加。

Dynamic Table

⭐ 第二步，在点击事件流映射的动态输入表上执行一个连续查询（Continuous Query），并生成一个新的动态输出表。

下面介绍两个查询的案例：

第一个查询：一个简单的 GROUP-BY COUNT 聚合查询，写过 SQL 的都不会陌生吧，这种应该都是最基础，最常用的对数据按照类别分组的方法。

如下图所示 group by 聚合的常用案例。

time

那么本案例中呢，是基于 clicks 表中 user 字段对 clicks 表（点击事件流）进行分组，来统计每一个 user 的访问的 URL 的数量。下面的图展示了当 clicks 输入表来了新数据（即表更新时），连续查询（Continuous Query）的计算逻辑。

group agg

当查询开始，clicks 表(左侧)是空的。

⭐ 当第一行数据被插入到 clicks 表时，连续查询（Continuous Query）开始计算结果数据。数据源表第一行数据 [Mary,./home] 输入后，会计算结果 [Mary, 1] 插入（insert）结果表。⭐ 当第二行 [Bob, ./cart] 插入到 clicks 表时，连续查询（Continuous Query）会计算结果 [Bob, 1]，并插入（insert）到结果表。⭐ 第三行 [Mary, ./prod?id=1] 输出时，会计算出[Mary, 2]（user 为 Mary 的数据总共来过两条，所以为 2），并更新（update）结果表，[Mary, 1] 更新成 [Mary, 2]。⭐ 最后，当第四行数据加入 clicks 表时，查询将第三行 [Liz, 1] 插入（insert）结果表中。

注意上述特殊标记出来的字体，可以看到连续查询对于结果的数据输出方式有两种：

⭐ 插入（insert）结果表⭐ 更新（update）结果表

大家对于插入（insert）结果表这件事都比较好理解，因为离线数据都只有插入这个概念。

但是更新（update）结果表就是离线处理中没有概念了。这就是连续查询中中比较重要一个概念。后文会介绍。

接下来介绍第二条查询语句。

第二条查询与第一条类似，但是 group by 中除了 user 字段之外，还 group by 了 tumble，其代表开了个滚动窗口（后面会详细说明滚动窗口的作用），然后计算 url 数量。

group by user，是按照类别（横向）给数据分组，group by tumble 滚动窗口是按时间粒度（纵向）给数据进行分组。如下图所示。

time

图形化一解释就很好理解了，两种都是对数据进行分组，一个是按照类别分组，另一种是按照时间分组。

与前面一样，左边显示了输入表 clicks。查询每小时持续计算结果并更新结果表。clicks 表有三列，user，cTime，url。其中 cTime 代表数据的时间戳，用于给数据按照时间粒度分组。

tumble window

我们的滚动窗口的步长为 1 小时，即时间粒度上面的分组为 1 小时。其中时间戳在 12:00:00 - 12:59:59 之间有四条数据。13:00:00 - 13:59:59 有三条数据。14:00:00 - 14:59:59 之间有四条数据。

⭐ 当 12:00:00 - 12:59:59 数据输入之后，1 小时的窗口，连续查询（Continuous Query）计算的结果如右图所示，将 [Mary, 3]，[Bob, 1] 插入（insert）结果表。⭐ 当 13:00:00 - 13:59:59 数据输入之后，1 小时的窗口，连续查询（Continuous Query）计算的结果如右图所示，将 [Bob, 1]，[Liz, 2] 插入（insert）结果表。⭐ 当 14:00:00 - 14:59:59 数据输入之后，1 小时的窗口，连续查询（Continuous Query）计算的结果如右图所示，将 [Mary, 1]，[Bob, 2]，[Liz, 1] 插入（insert）结果表。

而这个查询只有插入（insert）结果表这个行为。

6.SQL 连续查询的两种类型：更新（Update）查询 & 追加（Append）查询

虽然前一节的两个查询看起来非常相似（都计算分组进行计数聚合），但它们在一个重要方面不同：

⭐ 第一个查询（group by user），即（Update）查询：会更新先前输出的结果，即结果表流数据中包含 INSERT 和 UPDATE 数据。小伙伴萌可以理解为 group by user 这条语句当中，输入源的数据是一直有的，源源不断的，同一个 user 的数据之后可能还是会有的，因此可以认为此 SQL 的每次的输出结果都是一个中间结果，当同一个 user 下一条数据到来的时候，就要用新结果把上一次的产出中间结果（旧结果）给 UPDATE 了。所以这就是 UPDATE 查询的由来（其中 INSERT 就是第一条数据到来的时候，没有之前的中间结果，所以是 INSERT）。⭐ 第二个查询（group by user, tumble(xxx)），即（Append）查询：只追加到结果表，即结果表流数据中只包含 INSERT 的数据。小伙伴萌可以理解为虽然 group by user, tumble(xxx) 上游也是一个源源不断的数据，但是这个查询本质上是对时间上的划分，而时间都是越变越大的，当前这个滚动窗口结束之后，后面来的数据的时间都会比这个滚动窗口的结束时间大，都归属于之后的窗口了，当前这个滚动窗口的结果数据就不会再改变了，因此这条查询只有 INSERT 数据，即一个 Append 查询。

上面是 Flink SQL 连续查询处理机制上面的两类查询方式。我们可以发现连续查询的处理机制不一样，产出到结果表中的结果数据也是不一样的。针对上面两种结果表的更新方式，Flink SQL 提出了 changelog 表的概念来进行兼容。

changelog 表这个概念其实就和 MySQL binlog 是一样的。会包含 INSERT、UPDATE、DELETE 三种数据，通过这三种数据的处理来描述实时处理技术对于动态表的变更：

⭐ changelog 表：即第一个查询的输出表，输出结果数据不但会追加，还会发生更新⭐ changelog insert-only 表：即第二个查询的输出表，输出结果数据只会追加，不会发生更新

7.SQL 流处理的输出：动态输出表转化为输出数据

可以看到我们的标题都是随着一个 SQL 的生命周期的。从输入流映射为 SQL 动态输入表、实时处理底层技术 - SQL 连续查询到本小节的 SQL 动态输出表转化为输出数据。都是有逻辑关系的。

我们上面介绍到了连续查询（Continuous Query）的输出结果表是一个 changelog。其可以像普通数据库表一样通过 INSERT、UPDATE 和 DELETE 来不断修改。

它可能是一个只有一行、不断更新 changelog 表，也可能是一个 insert-only 的 changelog 表，没有 UPDATE 和 DELETE 修改，或者介于两者之间的其他表。

在将动态表转换为流或将其写入外部系统时，需要对这些不同状态的数据进行编码。Flink 的 Table API 和 SQL API 支持三种方式来编码一个动态表的变化:

⭐ Append-only 流：输出的结果只有 INSERT 操作的数据。⭐ Retract 流：⭐ Retract 流包含两种类型的 message：add messages 和 retract messages 。其将 INSERT 操作编码为 add message、将 DELETE 操作编码为 retract message、将 UPDATE 操作编码为更新先前行的 retract message 和更新（新）行的 add message，从而将动态表转换为 retract 流。⭐ Retract 流写入到输出结果表的数据如下图所示，有 -，+ 两种，分别 - 代表撤回旧数据，+ 代表输出最新的数据。这两种数据最终都会写入到输出的数据引擎中。⭐ 如果下游还有任务去消费这条流的话，要注意需要正确处理 -，+ 两种数据，防止数据计算重复或者错误。

retract

⭐ Upsert 流：⭐ Upsert 流包含两种类型的 message：upsert messages 和 delete messages。转换为 upsert 流的动态表需要唯一键（唯一键可以由多个字段组合而成）。其会将 INSERT和 UPDATE 操作编码为 upsert message，将 DELETE 操作编码为 delete message。⭐ Upsert 流写入到输出结果表的数据如下图所示，每次输出的结果都是当前每一个 user 的最新结果数据，不会有 Retract 中的 - 回撤数据。⭐ 如果下游还有一个任务去消费这条流的话，消费流的算子需要知道唯一键（即 user），以便正确地根据唯一键（user）去拿到每一个 user 当前最新的状态。其与 retract 流的主要区别在于 UPDATE 操作是用单个 message 编码的，因此效率更高。下图显示了将动态表转换为 upsert 流的过程。