企业官网建设流程全解析

• 大数据
• 流处理
• 批处理
• 数据工程

Flink 的 Table API 与 SQL 是一套流批统一的声明式 API，在有限的批式输入和无限的流式输入下具备相同的语义。由于关系代数与 SQL 最初是为批处理设计的，流式场景下的状态管理成为理解与调优的关键。本文以docs/content.zh/docs/dev/table/concepts/overview.md为核心，系统讲解流式表程序的状态使用方式、空闲状态维持时间（State TTL）的三种配置途径、从 Flink v1.18 起支持的算子级状态 TTL（含 CompiledPlan 完整实操），并延伸状态化更新与演化等进阶话题，帮助你掌握状态维度下的流式 SQL 生产实践。

流批统一：状态是流式表程序的灵魂

Flink 的 Table API 与 SQL 在批式与流式输入下共享同一套语义。区别在于：批式查询天然拥有有限的输入集，可以一次性完成计算；而流式查询面对的是无限的数据流，必须以**连续查询（Continuous Query）**的形式持续运行，因此必须依赖状态来保存跨时间维度的中间结果。

一个流模式下运行的表程序（Table program）可以完整利用 Flink 作为有状态流处理器的能力：

• 配置不同的 state backend（如 RocksDB、Heap），以适配不同规模的状态存储需求；
• 配置多种 checkpoint 选项，以满足不同的容错与恢复需求；
• 对正在运行的 Table API & SQL 管道生成 savepoint，并在之后用其恢复应用状态。

状态使用：声明式管道中的隐式状态

由于 Table API & SQL 程序是声明式的，状态会在哪里、如何被使用并不直接可见。**Planner（优化器）**负责判断是否需要状态来得到正确的计算结果，并尽可能把管道优化成使用更少状态的形式。

从概念上讲，源表从来不会在状态中被完全保存——实现者处理的是逻辑表，即动态表（Dynamic Table），各算子的状态完全取决于具体用到的操作。

显式状态算子：Join、聚合与去重

包含连接（Join）、聚合（Aggregation）或去重（Deduplication）等操作的语句，需要在 Flink 抽象的容错存储内保存中间结果，这类算子被称为状态算子。

例如对两个表执行普通 Join，基于正确的 SQL 语义，运行时假设两表会在任意时间点进行匹配，因此算子需要保存两个表的全部输入。为了控制状态规模，Flink 提供了优化窗口 Join 和时段 Join，利用 watermarks 概念（即时间属性）让过期的数据不再参与匹配，从而显著缩小状态。

另一个经典例子是词频统计：

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;

这里word是分组的键，连续查询为每个观察到的word维护一个中间状态来保存当前词频。由于输入word的值随时间变化，且查询持续运行，Flink 会为每个word维护一个中间状态，总状态量会随着新word的出现不断增长——这正是流式聚合状态下最需要警惕的内存与存储风险点。

隐式状态算子：SELECT 也可能引入状态

形如SELECT ... FROM ... WHERE这种只包含字段映射或过滤器的查询通常是无状态的。但在某些情况下，根据输入数据的特征或配置，状态算子会被隐式地推导出来：

• 输入表是不带UPDATE_BEFORE的更新流（详见表到流的转换）；
• 或配置了table-exec-source-cdc-events-duplicate。

下面的例子展示了对 upsert-kafka 源表执行最简单的SELECT *：

CREATE TABLE upsert_kakfa ( id INT PRIMARY KEY NOT ENFORCED, message STRING ) WITH ( 'connector' = 'upsert-kafka', ... ); SELECT * FROM upsert_kakfa;

upsert-kafka 源表的消息类型只包含INSERT、UPDATE_AFTER和DELETE，而下游可能要求完整的 changelog（包含UPDATE_BEFORE）。因此，虽然查询本身不包含任何状态计算，优化器依然会隐式地推导出一个 ChangelogNormalize 状态算子来生成完整的 changelog。

空闲状态维持时间：table.exec.state.ttl

空闲状态维持时间参数table.exec.state.ttl定义了状态的键在被更新后要保持多长时间才被移除。在上述词频例子中，某个word的计数会在配置的时间内未更新时被立刻移除。

该配置项的源码定义位于flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java：

• 键名为table.exec.state.ttl，类型为 Duration；
• 默认值为0 ms，含义是永不清理状态；
• 清理状态会引入额外的簿记开销（bookkeeping），因此默认关闭。

移除状态的键之后，连续查询会完全忘记它曾经见过这个键；如果一条记录带有一个曾被移除状态的键，该记录会被当作对应键的第一条记录处理。在词频例子中，这意味着cnt会再次从0开始计数——这是配置 TTL 时最容易忽略的语义影响。

指定状态生命周期的三种方式

从 Flink v1.18 开始，Table API & SQL 支持多种粒度的状态 TTL 配置方式，下表（源自原文档）总结了它们的适用面与优先级：

STATE_TTL 查询提示

STATE_TTL提示以 SQL 注释形式作用于具体算子。从源码flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/StateTtlHint.java可以看出其实现要点：

• 对于双输入算子（如 Join），支持形如STATE_TTL('T1' = '1d', 'T2' = '2d')的键值对写法，分别指定左右输入的 TTL（源码中LEFT_INPUT映射为输入侧 0，其余映射为输入侧 1）；
• 对于单输入算子（如分组聚合），支持STATE_TTL('T1' = '2d')形式；
• TTL 值支持d（天）、h（小时）等 Flink 时间单位，通过TimeUtils.parseDuration解析为毫秒。

对应的测试flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/StateTtlHintTest.java覆盖了大量边界场景，例如：

-- 为 Join 的左右输入分别指定 TTL select /*+ STATE_TTL('T2' = '2d', 'T1' = '1d') */* from T1 join T2 on T1.a1 = T2.a2 -- 为分组聚合指定 TTL select /*+ STATE_TTL('T1' = '2d') */ count(*) from T1 group by a1

测试同时验证了非法用法会被拒绝，例如提示选项与输入表名不匹配（报错The options of following hints cannot match the name of input tables or views）、STATE_TTL()不带任何键值选项（报错Invalid STATE_TTL hint, expecting at least one key-value options specified.）等情况。

配置算子粒度的状态 TTL（高级特性）

注意：这是一个需要小心使用的高级特性。它仅适用于作业中使用了多个状态、且每个状态需要不同 TTL 的场景。无状态作业无需关注；若作业仅使用一个状态，仅需设置作业级 TTL 参数table.exec.state.ttl即可。

从 Flink v1.18 开始，Table API & SQL 支持以每个状态算子的入边数为粒度配置细粒度状态 TTL：

• OneInputStreamOperator（单输入）可配置一个状态的 TTL；
• TwoInputStreamOperator（如双流 Join）可分别为左状态和右状态配置 TTL；
• 更一般地，具有 K 个输入的MultipleInputStreamOperator可以配置 K 个状态 TTL。

典型使用场景：

1. 为双流 Join的左右流配置不同 TTL：双流 Join 会生成拥有两条输入边的TwoInputStreamOperator状态算子，分别用两个状态保存来自左流和右流的更新；
2. 在同一作业中为不同的状态计算设置不同 TTL：例如一个 ETL 作业先用ROW_NUMBER进行去重，再用GROUP BY进行聚合，会生成两个拥有单条输入边的OneInputStreamOperator状态算子，可为它们分别设置不同的 TTL。

需要说明的是，基于窗口的操作（如窗口连接、窗口聚合、窗口 Top-N 等）和 Interval Join 不依赖table.exec.state.ttl控制状态保留，因此它们的状态无法在算子级别配置。

第一步：生成 Compiled Plan

配置过程首先使用COMPILE PLAN语句生成一个 JSON 文件，它表示序列化后的执行计划。注意COMPILE PLAN不支持查询语句SELECT ... FROM ...，只支持INSERT类语句或语句集合。

Java 方式：

TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); tableEnv.executeSql( "CREATE TABLE orders (order_id BIGINT, order_line_id BIGINT, buyer_id BIGINT, ...)"); tableEnv.executeSql( "CREATE TABLE line_orders (order_line_id BIGINT, order_status TINYINT, ...)"); tableEnv.executeSql( "CREATE TABLE enriched_orders (order_id BIGINT, order_line_id BIGINT, order_status TINYINT, ...)"); // CompilePlan#writeToFile only supports a local file path, if you need to write to remote filesystem, // please use tableEnv.executeSql("COMPILE PLAN 'hdfs://path/to/plan.json' FOR ...") CompiledPlan compiledPlan = tableEnv.compilePlanSql( "INSERT INTO enriched_orders \n" + "SELECT a.order_id, a.order_line_id, b.order_status, ... \n" + "FROM orders a JOIN line_orders b ON a.order_line_id = b.order_line_id"); compiledPlan.writeToFile("/path/to/plan.json");

Scala 方式：

val tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()) tableEnv.executeSql( "CREATE TABLE orders (order_id BIGINT, order_line_id BIGINT, buyer_id BIGINT, ...)") tableEnv.executeSql( "CREATE TABLE line_orders (order_line_id BIGINT, order_status TINYINT, ...)") tableEnv.executeSql( "CREATE TABLE enriched_orders (order_id BIGINT, order_line_id BIGINT, order_status TINYINT, ...)") val compiledPlan = tableEnv.compilePlanSql( """ |INSERT INTO enriched_orders |SELECT a.order_id, a.order_line_id, b.order_status, ... |FROM orders a JOIN line_orders b ON a.line_order_id = b.order_line_id |""".stripMargin) // CompilePlan#writeToFile only supports a local file path, if you need to write to remote filesystem, // please use tableEnv.executeSql("COMPILE PLAN 'hdfs://path/to/plan.json' FOR ...") compiledPlan.writeToFile("/path/to/plan.json")

SQL CLI 方式：

Flink SQL> CREATE TABLE orders (order_id BIGINT, order_line_id BIGINT, buyer_id BIGINT, ...); [INFO] Execute statement succeeded. Flink SQL> CREATE TABLE line_orders (order_line_id BIGINT, order_status TINYINT, ...); [INFO] Execute statement succeeded. Flink SQL> CREATE TABLE enriched_orders (order_id BIGINT, order_line_id BIGINT, order_status TINYINT, ...); [INFO] Execute statement succeeded. Flink SQL> COMPILE PLAN 'file:///path/to/plan.json' FOR INSERT INTO enriched_orders > SELECT a.order_id, a.order_line_id, b.order_status, ... > FROM orders a JOIN line_orders b ON a.order_line_id = b.order_line_id; [INFO] Execute statement succeeded.

COMPILE PLAN的 SQL 语法如下：

COMPILE PLAN [IF NOT EXISTS] <plan_file_path> FOR <insert_statement>|<statement_set>; statement_set: EXECUTE STATEMENT SET BEGIN insert_statement; ... insert_statement; END; insert_statement: <insert_from_select>|<insert_from_values>

该语句会在指定位置生成一个 JSON 文件。除本地路径外，COMPILE PLAN还支持写入hdfs://、s3://等 Flink 支持的文件系统，请确保为目标写入路径设置了写入权限。

第二步：修改 Compiled Plan 中的状态 TTL

每个状态算子会在 JSON 计划中显式生成一个名为state的数组，结构如下。理论上一个拥有 k 路输入的状态算子拥有 k 个状态：

"state": [ { "index": 0, "ttl": "0 ms", "name": "${1st input state name}" }, { "index": 1, "ttl": "0 ms", "name": "${2nd input state name}" }, ... ]

找到需要修改的状态算子，将 TTL 设置为带毫秒单位的正整数。例如将第一个状态算子的 TTL 设置为 1 小时：

{ "index": 0, "ttl": "3600000 ms", "name": "${1st input state name}" }

这一 JSON 结构的字段定义可在源码flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/StateMetadata.java中找到：index（该状态属于算子的第几路输入，从 0 开始计数）、ttl（该路输入状态的保留时间，单位毫秒）、name（状态描述，如deduplicate-state、join-left-state等）。此外，该源码还实现了向后兼容逻辑：若状态元数据列表为空，则回退为从表配置table.exec.state.ttl读取统一 TTL。

一个需要留意的经验法则：下游状态算子的 TTL 不应小于上游状态算子的 TTL。

第三步：执行 Compiled Plan

EXECUTE PLAN语句会反序列化上述 JSON 文件，进一步生成 JobGraph 并提交作业。通过EXECUTE PLAN提交的作业，其状态算子的 TTL 值从文件中读取，配置项table.exec.state.ttl的值会被忽略。

Java 方式：

TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()); tableEnv.executeSql( "CREATE TABLE orders (order_id BIGINT, order_line_id BIGINT, buyer_id BIGINT, ...)"); tableEnv.executeSql( "CREATE TABLE line_orders (order_line_id BIGINT, order_status TINYINT, ...)"); tableEnv.executeSql( "CREATE TABLE enriched_orders (order_id BIGINT, order_line_id BIGINT, order_status TINYINT, ...)"); // PlanReference#fromFile only supports a local file path, if you need to read from remote filesystem, // please use tableEnv.executeSql("EXECUTE PLAN 'hdfs://path/to/plan.json'").await(); tableEnv.loadPlan(PlanReference.fromFile("/path/to/plan.json")).execute().await();

Scala 方式：

val tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode()) tableEnv.executeSql( "CREATE TABLE orders (order_id BIGINT, order_line_id BIGINT, buyer_id BIGINT, ...)") tableEnv.executeSql( "CREATE TABLE line_orders (order_line_id BIGINT, order_status TINYINT, ...)") tableEnv.executeSql( "CREATE TABLE enriched_orders (order_id BIGINT, order_line_id BIGINT, order_status TINYINT, ...)") // PlanReference#fromFile only supports a local file path, if you need to read from remote filesystem, // please use tableEnv.executeSql("EXECUTE PLAN 'hdfs://path/to/plan.json'").await() tableEnv.loadPlan(PlanReference.fromFile("/path/to/plan.json")).execute().await()

SQL CLI 方式：

Flink SQL> CREATE TABLE orders (order_id BIGINT, order_line_id BIGINT, buyer_id BIGINT, ...); [INFO] Execute statement succeeded. Flink SQL> CREATE TABLE line_orders (order_line_id BIGINT, order_status TINYINT, ...); [INFO] Execute statement succeeded. Flink SQL> CREATE TABLE enriched_orders (order_id BIGINT, order_line_id BIGINT, order_status TINYINT, ...); [INFO] Execute statement succeeded. Flink SQL> EXECUTE PLAN 'file:///path/to/plan.json'; [INFO] Submitting SQL update statement to the cluster... [INFO] SQL update statement has been successfully submitted to the cluster: Job ID: 79fbe3fa497e4689165dd81b1d225ea8

EXECUTE PLAN的 SQL 语法：

EXECUTE PLAN [IF EXISTS] <plan_file_path>;

完整示例：为双流 Join 的左右状态配置不同 TTL

下面通过一个计算订单明细的双流 Join 作业，演示完整的算子级 TTL 配置流程。

① 生成 compiled plan

-- left source table CREATE TABLE Orders ( `order_id` INT, `line_order_id` INT ) WITH ( 'connector'='...' ); -- right source table CREATE TABLE LineOrders ( `line_order_id` INT, `ship_mode` STRING ) WITH ( 'connector'='...' ); -- sink table CREATE TABLE OrdersShipInfo ( `order_id` INT, `line_order_id` INT, `ship_mode` STRING ) WITH ( 'connector' = '...' ); COMPILE PLAN '/path/to/plan.json' FOR INSERT INTO OrdersShipInfo SELECT a.order_id, a.line_order_id, b.ship_mode FROM Orders a JOIN LineOrders b ON a.line_order_id = b.line_order_id;

生成的 JSON 文件内容如下（节选关键部分）：

{ "flinkVersion" : "1.18", "nodes" : [ { "id" : 1, "type" : "stream-exec-table-source-scan_1", "scanTableSource" : { "table" : { "identifier" : "`default_catalog`.`default_database`.`Orders`", "resolvedTable" : { ... } } }, "outputType" : "ROW<`order_id` INT, `line_order_id` INT>", "description" : "TableSourceScan(table=[[default_catalog, default_database, Orders]], fields=[order_id, line_order_id])", "inputProperties" : [ ] }, { "id" : 2, "type" : "stream-exec-exchange_1", "inputProperties" : [ ... ], "outputType" : "ROW<`order_id` INT, `line_order_id` INT>", "description" : "Exchange(distribution=[hash[line_order_id]])" }, { "id" : 3, "type" : "stream-exec-table-source-scan_1", "scanTableSource" : { "table" : { "identifier" : "`default_catalog`.`default_database`.`LineOrders`", "resolvedTable" : {...} } }, "outputType" : "ROW<`line_order_id` INT, `ship_mode` VARCHAR(2147483647)>", "description" : "TableSourceScan(table=[[default_catalog, default_database, LineOrders]], fields=[line_order_id, ship_mode])", "inputProperties" : [ ] }, { "id" : 4, "type" : "stream-exec-exchange_1", "inputProperties" : [ ... ], "outputType" : "ROW<`line_order_id` INT, `ship_mode` VARCHAR(2147483647)>", "description" : "Exchange(distribution=[hash[line_order_id]])" }, { "id" : 5, "type" : "stream-exec-join_1", "joinSpec" : { ... }, "state" : [ { "index" : 0, "ttl" : "0 ms", "name" : "leftState" }, { "index" : 1, "ttl" : "0 ms", "name" : "rightState" } ], "inputProperties" : [ ... ], "outputType" : "ROW<`order_id` INT, `line_order_id` INT, `line_order_id0` INT, `ship_mode` VARCHAR(2147483647)>", "description" : "Join(joinType=[InnerJoin], where=[(line_order_id = line_order_id0)], select=[order_id, line_order_id, line_order_id0, ship_mode], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])" }, { "id" : 6, "type" : "stream-exec-calc_1", "projection" : [ ... ], "condition" : null, "inputProperties" : [ ... ], "outputType" : "ROW<`order_id` INT, `line_order_id` INT, `ship_mode` VARCHAR(2147483647)>", "description" : "Calc(select=[order_id, line_order_id, ship_mode])" }, { "id" : 7, "type" : "stream-exec-sink_1", "configuration" : { ... }, "dynamicTableSink" : { "table" : { "identifier" : "`default_catalog`.`default_database`.`OrdersShipInfo`", "resolvedTable" : { ... } } }, "inputChangelogMode" : [ "INSERT" ], "inputProperties" : [ ... ], "outputType" : "ROW<`order_id` INT, `line_order_id` INT, `ship_mode` VARCHAR(2147483647)>", "description" : "Sink(table=[default_catalog.default_database.OrdersShipInfo], fields=[order_id, line_order_id, ship_mode])" } ], "edges" : [ ... ] }

② 修改状态 TTL

上述 JSON 中，id=5的 Join 算子的状态信息如下。index代表状态属于算子的第几路输入（从 0 开始），当前左右流的 TTL 均为"0 ms"，表示 TTL 尚未开启：

"state": [ { "index": 0, "ttl": "0 ms", "name": "leftState" }, { "index": 1, "ttl": "0 ms", "name": "rightState" } ]

现在将左流 TTL 设置为"3000 ms"，右流设置为"9000 ms"：

"state": [ { "index": 0, "ttl": "3000 ms", "name": "leftState" }, { "index": 1, "ttl": "9000 ms", "name": "rightState" } ]

③ 执行 compiled plan

保存修改后，使用EXECUTE PLAN语句提交作业，此时提交的作业中 Join 的左右流便使用了上述不同的 TTL：

EXECUTE PLAN '/path/to/plan.json'

状态化更新与演化

表程序在流模式下执行时被视为标准查询：它们被定义一次后，将一直作为静态的端到端（end-to-end）管道运行。对于这种状态化管道，查询语句的改动和 Flink Planner 的改动都有可能产生完全不同的执行计划，这使表程序的状态化升级与演化具有挑战性。

例如，为了添加一个过滤谓词，优化器可能决定重排 Join 或改变内部算子的 schema，这会阻碍从 savepoint 的恢复——因为改变后的拓扑和算子状态的列布局与旧计划存在差异。

因此：

• 查询实现者需要确保改动在优化计划前后是兼容的。可以在 SQL 中使用EXPLAIN，或在 Table API 中使用table.explain()获取详情（参见解释一个表）；
• 由于新的优化器规则不断被添加，算子变得更加高效和专用，升级到更新的 Flink 版本也可能造成不兼容的计划。

警告：当前框架无法保证状态可以从 savepoint 映射到新的算子拓扑上。换言之：savepoint 只在查询语句和 Flink 版本保持恒定的情况下才被支持。

由于社区拒绝在版本补丁（如1.13.1至1.13.2）上对优化计划和算子拓扑进行修改的贡献，将 Table API & SQL 管道升级到新的 bug fix 发行版应当是安全的；然而主次（major-minor）版本的更新（如1.12至1.13）不被支持。

鉴于这两个限制（修改查询语句、修改 Flink 版本），建议在升级后、切换到实时数据之前，先用历史数据对升级后的表程序做"暖机"（即初始化），验证其能否正常启动与恢复。Flink 社区正致力于通过混合源（Hybrid Source）让这一切换尽可能方便。

延伸阅读

围绕流式表程序，以下文档与本文形成完整知识体系：

• 动态表：动态表的核心概念，是理解流式 SQL 语义的基础；
• 时间属性：时间属性及其在 Table API & SQL 中的使用方式；
• 时态（Temporal）表：时态表的概念与应用；
• 流上的 Join：流式场景下支持的几种 Join；
• 流上的确定性：流计算确定性的解释；
• 查询配置：Table API & SQL 特有的全部配置项。

相关源码佐证可进一步阅读：flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java（table.exec.state.ttl的默认值与语义定义）、flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/StateTtlHint.java（STATE_TTL提示的解析实现）、flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/StateMetadata.java（CompiledPlan 中状态元数据的 JSON 结构与向后兼容逻辑），以及测试flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/StateTtlHintTest.java。

• 大数据
• 流处理
• 批处理
• 数据工程