Flink SQL 中的流式概念：状态算子

传统的关系模型和 SQL 最开始都是为了批式处理而设计的，当把一个关系型查询应用到流式处理上时，在实现和转换的过程中，会有很多和批处理场景非常不同的地方，典型的例子就是：为了实现 SQL 的某些语义，Flink 必须在流上维持状态，典型的代表就是：连接、聚合 、去重 这些操作，它们都是“状态算子”，本质原因还是因为：流处理的表是无界的，流式查询是持续不停的，所以在流上维持状态是必须的。

此外，我们应意识到：由于 Table API & SQL 程序是声明式的，管道会哪里维持状态以及状态如何被使用都是不明确的，就是说不能从 SQL 直接简单地推断出来，另外，Flink 还会对查询进行优化，尽可能地减少“状态”的使用。

下面是官方文档给出的一个状态算子的示例：

CREATE TABLE doc (
    word STRING
) WITH (
    'connector' = '...'
);
CREATE TABLE word_cnt (
    word STRING PRIMARY KEY NOT ENFORCED,
    cnt  BIGINT
) WITH (
    'connector' = '...'
);

INSERT INTO word_cnt
SELECT word, COUNT(1) AS cnt
FROM doc
GROUP BY word;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

这里的聚合函数 count 就需要状态维持，同时又由于分组（group by）的存在，要维持的状态数据就一下变多了，每一个单词都要独立维护一个对应的状态。下图是针对上面的查询语句“编译”（转换）出的流式程序的图解：

在这张详细的图解中，我们应该注意这些重点：

1. count函数是一个状态算子，它的要维持状态数据，也就是每个单词的词频，这些状态数据又同时是下游的输入数据
2. 状态数据需要实时地推送到下游，状态数据的变更也是以 changelog 形式传导的，所以才会有 +U('hello', 2)，-U('hello', 1)这样的消息产生

除了 连接、聚合 、去重 这些显式的状态算子，还有一些“隐式”的状态算子，按官方文档的介绍是说：由优化器隐式推导出来的。这里面的实现机理暂时还不清楚，但是例子是非常典型的！我们在《Flink 实时数仓关键技术解读：Upsert Kafka 和 动态表（Dynamic Table）》这篇文章中曾经详细地解读过 upsert-kafka 作为 sink 时写入到 kafka 中的数据，当再次以这些数据作为 source 进行流式读取时，upsert-kafka 是能够完整推导出 changelog 数据的，利用的就是这里所谓的“隐式推导”能力，具体地说就是一个叫 ChangelogNormalize 的状态算子。

在持续运行的流上维持状态可能是一个成分非常大的操作，因为流是不会停止的，随着时间的推移和大量数据的涌入，状态数据可能会越积越多，导致内存挤爆。所以 Flink 提供了状态的 TTL 机制，当状态在一定时间内没有被更新后就会被自动移除，这个参数就是：table.exec.state.ttl

定义了状态的键在被更新后要保持多长时间才被移除。 在之前的查询例子中，word 的数目会在配置的时间内未更新时立刻被移除。

通过移除状态的键，连续查询会完全忘记它曾经见过这个键。如果一个状态带有曾被移除状态的键被处理了，这条记录将被认为是对应键的第一条记录。上述例子中意味着 cnt 会再次从 0 开始计数。

---

补充介绍：

管道 （Pipeline）：Flink 文档中会反复出现这个名词，在 Flink 中，它指的是一个流式查询从 Source 到 Sink 的完整 DAG，中间是各种算子，简单地说就是：一个查询被“翻译”成一个流后的所有的处理环节。