企业官网建设流程全解析

Flink SQL 1.13 这个版本，我在生产环境里用了一年多，从最初抱着“试试看”的心态，到后来把团队里一大半的实时任务都迁到这套语法上，中间踩过的坑和收获的经验都不少。如果你正在纠结要不要从 DataStream API 切到 Flink SQL，或者已经在用但总被各种细节卡住，这篇东西应该能给你一些实用参考。我会直接按自己做项目时的思考路径来讲，重点放在 1.13 里真正影响日常开发的变化上，比如时区语义、动态表、维表关联和实际部署的坑。

1. 为什么先说清楚 1.13 的定位

1.1 流批一体在 SQL 层的落地节点

Flink SQL 1.13 在流批一体的路线上属于一个关键节点。之前几个版本的 SQL 能力虽然一直在补，但很多是“能用”和“好用”之间的距离。比如老版本里流处理和批处理在 SQL 语义上有不少隐性差异，同一套查询切到流模式可能结果就不一样，这对研发来说特别难受。1.13 开始，社区把精力重点放在统一语义上，尤其是时区、时间类型、类型系统这几个底层问题，这些看着不起眼，却决定了上层 SQL 能不能在流批两种模式下保持一致。

我个人的感受是，1.13 之后写 Flink SQL 更像写传统数据库 SQL 了，很多心智负担被抹平。以前需要记住“这个函数在流模式不支持”“这个类型要 cast 一下才能用”之类的琐碎约束，在 1.13 里少了很多。特别是对从 Hive、传统数仓转过来的同事，上手成本明显降低。当然，它还不是完美的，但作为生产使用的起点，1.13 是一个比较稳的选择。

1.2 关键变化一览

我先用一张表列出 1.13 里和日常开发最相关的核心变化，后面再一个个拆开讲，方便你先有个整体印象。

表格里这些点，任何一个都够写一篇长文，但实际使用中最容易出问题的，还是时区和动态表这两块。它们一个是“看起来简单但坑最多”，一个是“理解了就一通百通”，所以我下面会重点展开。

2. 时区处理：最容易被忽视的隐形炸弹

2.1 老版本时区混乱的根源

如果你从 1.11 或更早的版本迁移过来，一定遇到过这种诡异场景：同一份 SQL，在测试集群跑结果正常，上生产就所有窗口数据都偏移了。查了半天，最后发现是两个集群的jobmanager.timezone配置不一样。老版本 Flink SQL 的时间处理本质上没有把“时间点”和“墙上时间”区分清楚，TIMESTAMP类型直接映射到本地时区的LocalDateTime，当集群时区不确定时，结果就跟着飘。

这里有个基础概念值得先说透：TIMESTAMP是无时区语义的，它只表示一个日历日期和时间，比如“2024-05-20 10:00:00”，在不同时区下它还是这个字面值。而TIMESTAMP_LTZ（Long Time Zone）表示的是绝对时间点，带时区信息，比如某个 epoch 毫秒值，在不同时区下显示出来的墙上时间会不同，但它代表的是同一个瞬间。

流处理里的时间本质上是绝对时间点，因为事件发生的那一刻是客观的。如果你把事件时间定义成TIMESTAMP，Flink 会把它当成本地时间字符串来处理，一旦集群时区和数据源头时区不一致，窗口计算就会偏移。这就是老版本时区问题最根本的原因。

2.2 1.13 里如何正确配置时区

1.13 的解决方案是让用户在 SQL 级别显式声明时区，而不是依赖集群的默认值。最核心的配置就是table.local-time-zone。你可以在作业提交时这样指定：

./bin/flink run \ -D table.local-time-zone=Asia/Shanghai \ -c your.MainClass \ your-job.jar

或者在代码里通过TableConfig设置：

Configuration configuration = new Configuration(); configuration.setString("table.local-time-zone", "Asia/Shanghai"); EnvironmentSettings settings = EnvironmentSettings.newInstance() .inStreamingMode() .withConfiguration(configuration) .build(); TableEnvironment tEnv = TableEnvironment.create(settings);

这里有一个要注意的点：table.local-time-zone指定的是 SQL 引擎内处理时间函数的时区，它影响LOCALTIME、CURRENT_TIMESTAMP、NOW()以及PROCTIME这些函数，也影响将TIMESTAMP_LTZ转换为字符串时所用的时区，但它不会自动帮你把上游数据里的字符串转换掉。

我见过不少同事在这里栽跟头：以为设置了时区，上游 Kafka 里存的"2024-05-20 10:00:00"字符串就会自动被当成东八区处理。实际上不是的，你需要自己用CAST或TO_TIMESTAMP函数把它转成TIMESTAMP_LTZ，并且明确加上时区偏移。举例如下：

-- 假设上游字符串是 "2024-05-20T10:00:00+08:00" -- 方法1：直接带偏移解析 SELECT TO_TIMESTAMP_LTZ(ts_str, 8) FROM source_table; -- 方法2：解析成无时区时间再指定目标时区 SELECT CAST(ts_str AS TIMESTAMP(3)) AT LOCAL TIME ZONE FROM source_table;

AT LOCAL TIME ZONE这个语法是 1.13 里很好用的工具，它的作用是把无时区的TIMESTAMP按会话时区解释成一个绝对时间点，得到TIMESTAMP_LTZ。简单说，如果你确定上游存的字符串是东八区的墙上时间，你就可以写CAST(ts_str AS TIMESTAMP(3)) AT LOCAL TIME ZONE，在时区配置为东八区时，它就正确转换成了绝对时间。

2.3 三种时间字段在生产里怎么选

Flink SQL 里有三种时间字段：事件时间（event_time）、处理时间（proc_time）、摄入时间（ingestion_time）。1.13 里它们的定义方式没有本质变化，但时区规则更清晰了。

事件时间是业务真正关心的数据发生时刻，我用得最多。定义方式是通过WATERMARK子句：

CREATE TABLE user_log ( user_id STRING, action STRING, ts_str STRING, -- 假设 ts_str 是带时区偏移的字符串 ts_ltz AS TO_TIMESTAMP_LTZ(ts_str, 8), WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'user_log', 'properties.bootstrap.servers' = '...', 'properties.group.id' = 'log-group', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' );

这里我建议，尽量在源头就把字符串解析成TIMESTAMP_LTZ作为事件时间字段。TIMESTAMP_LTZ在窗口、聚合、Join 中表现最稳定，不会因为集群或作业配置的时区改变而改变语义。

处理时间相对简单，就是算子所在机器当前的系统时间，使用时直接PROCTIME()函数就行。它适合对准确性要求不高的场景，比如计算“从数据进入系统到处理完的耗时”，或者某些实时大屏的简易监控。

摄入时间是数据进入 Flink 源头算子的时间，它介于事件时间和处理时间之间，由系统自动生成。1.13 里仍然支持ts AS METADATA FROM 'timestamp'这种语法来声明摄入时间，但在生产里我用得反而少了。因为摄入时间对“数据迟到”这种问题没有帮助，它只是“到了 Flink 才发现数据晚了”的时间点，如果你的 App 端埋点延迟很高，摄入时间仍然会让窗口计算结果不理想。

实践心得：时间字段的选型优先级，我个人排法是事件时间 > 摄入时间 > 处理时间。除非你的场景真的不在乎数据顺序和延迟，否则不要用处理时间做业务统计，因为它和机器负载强相关，结果不可复现。

3. 动态表：理解流批一体的钥匙

3.1 动态表到底是什么

动态表这个概念在 Flink SQL 里不是 1.13 才有的，但 1.13 对它的语义做了大量完善，让“流上的表”和“批里的表”能更好统一。我在给团队内部分享时常用一个生活类比：动态表就像一个持续更新的 Excel 表格，每当有新数据进来，表格里某些行就被插入、更新或删除，而你的 SQL 查询就像是在这个不断变化的表格上反复执行，每次执行的结果也在不断变化。

这个类比帮助很大。你写一条SELECT查询，在批处理里它跑完就出结果；在流处理里，它会一直跑下去，每来一条数据都可能触发一次结果更新。理解了这个，你就明白为什么流式 SQL 里会有“结果表”和“撤回”这些概念——因为数据更新了，之前发出去的结果需要修正。

1.13 里，动态表在流上的更新方式主要有三种：追加（Append）、更新（Upsert）和撤回（Retract）。你可以在 DDL 里通过PRIMARY KEY NOT ENFORCED来声明主键，这样下游就能用 Upsert 的方式看到更新：

CREATE TABLE dwd_order ( order_id BIGINT PRIMARY KEY NOT ENFORCED, user_id BIGINT, amount DECIMAL(10, 2), status STRING ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'dwd_order', 'properties.bootstrap.servers' = '...', 'key.format' = 'json', 'value.format' = 'json' );

加了主键之后，Flink SQL 才能正确表达“同一主键的更新”是更新而不是新增。如果你做聚合后要写回 Kafka 或者 MySQL，这条规则尤其重要。我遇到过团队同学没有声明主键，结果下游表里同一个订单 ID 出现多行，全是聚合产生的重复记录，后来排查了很久才意识到是缺了主键声明。

3.2 连续查询与结果更新

连续查询（Continuous Query）是动态表上的查询操作。这句话听着抽象，实际就是：普通的批查询执行一次就结束了，连续查询因为输入是不断变化的动态表，所以输出也在不断变化。

在 1.13 里，连续查询的结果更新机制已经比较成熟。比如下面这条分组聚合：

SELECT user_id, COUNT(*) AS cnt, SUM(amount) AS total FROM orders GROUP BY user_id;

在流模式下，每来一条订单，对应 user_id 的计数和总额就会更新一次。Flink 会向下游发送两条消息：一条是撤回旧的(cnt, total)值，一条是追加新的值。这就是典型的 Retract 模式。如果你把这些结果通过upsert-kafka连接器写出，Kafka 里看到的就是主键对应的最新值。

理解这一点，对排查“为什么我下游出现两条一模一样的数据，只是数值不同”这类问题非常有帮助。那些不是重复数据，是更新流的两条消息。想要下游只看最新值，就得用支持 Upsert 的存储，或者写 SQL 时把结果再聚合一次。

3.3 窗口聚合在 1.13 里的细节

窗口聚合是流处理用得最多的功能，1.13 对窗口的边界定义更明确，我实际使用中感受最深的有两点。

第一点，滚动窗口和滑动窗口的start、end字段语义更清晰。比如滚动窗口，窗口区间是左闭右开，即[start, end)，事件时间为end时刻的数据会落到下一个窗口。这一点在对比两个窗口的明细数据去重时尤其重要，因为“窗口边界到底算哪个窗口”总是让新同学困惑。

SELECT TUMBLE_START(ts_ltz, INTERVAL '1' HOUR) AS window_start, TUMBLE_END(ts_ltz, INTERVAL '1' HOUR) AS window_end, user_id, COUNT(*) FROM orders GROUP BY TUMBLE(ts_ltz, INTERVAL '1' HOUR), user_id;

TUMBLE_START和TUMBLE_END是专门用来输出窗口起止时间的函数。如果你用老写法直接GROUP BY窗口函数拿不到这两个值，1.13 里这两个辅助函数已经是标配。

第二点，迟到数据的处理。流处理里数据晚到是常态，1.13 里 Watermark 的推进机制和窗口允许的延迟时长配合得比较好。窗口会在 Watermark 超过窗口结束时间时触发计算，但结果不会立刻关闭，在allowedLateness指定的时间范围内，迟到的数据到来后还能触发一次更新。

-- 允许迟到10秒的数据触发窗口更新 SELECT TUMBLE_START(ts_ltz, INTERVAL '1' MINUTE) AS window_start, user_id, COUNT(*) FROM orders GROUP BY TUMBLE(ts_ltz, INTERVAL '1' MINUTE), user_id;

这里有个容易误解的点，需要配合 Watermark 策略：

WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND

Watermark 延迟 5 秒，窗口允许迟到 10 秒，那实际上窗口会在事件时间推进到窗口结束 + 5 秒时首次触发，然后继续等迟到数据到窗口结束 + 10 秒左右，超过之后彻底关闭，后续数据只能进侧输出流或者被丢弃。这个参数组合要根据你的业务容忍度来调，不能盲目照搬。

4. 从 SQL 到可部署的任务

4.1 用 SQL 文件组织任务的可维护性

很多团队第一次接触 Flink SQL，都是用 SQL Client 敲一条命令测一测，能跑通了就往代码里塞。其实到 1.13，用 SQL 文件组织任务已经非常成熟，维护起来比写一堆 Java 代码清爽得多。

我在项目里通常使用这样的方式：把每个作业的 SQL 放在独立文件里，用-f参数执行。如果需要初始化上下文，比如注册 Catalog、设置时区、加载依赖的 DDL，就放在-i指定的初始化脚本里。

./bin/sql-client.sh \ -i scripts/init.sql \ -f scripts/order_stats.sql

init.sql里放这类内容：

SET table.local-time-zone = Asia/Shanghai; SET execution.checkpointing.interval = 30s; SET execution.checkpointing.externalized-checkpoint-retention = RETAIN_ON_CANCELLATION; SET pipeline.auto-watermark-interval = 200ms; CREATE CATALOG my_hive WITH ( 'type' = 'hive', 'default-database' = 'default' ); USE CATALOG my_hive;

order_stats.sql里就是业务逻辑，从建源表、建维表、写结果到最终INSERT INTO一条龙。这种组织方式有几个好处：一是代码评审时直接看 SQL 就行，业务逻辑一目了然；二是环境切换时只要改init.sql里的 Catalog 或时区，不用动业务 SQL；三是新同事上手只要懂 SQL 就能改需求。

4.2 连接器与 Catalog 的配置要点

1.13 的连接器生态已经比较完整，Kafka、JDBC、Elasticsearch、HBase、Hive 这些主流存储都能通过 DDL 里的WITH参数快速对接。但连接器配置里有些细节，我踩过坑之后才发现文档里没写透。

拿最常见的 Kafka 连接器举例。生产环境里我强烈建议把properties.group.id显式配置上，并且不要用默认值，否则同一个作业并行度改变后会重新消费，导致重复计算。另外，scan.startup.mode这个参数要根据场景选择：

• earliest-offset：从头消费，适合离线追数或者首次上线
• latest-offset：只消费新数据，适合对历史数据不敏感的场景
• timestamp：指定时间戳之后的数据，适合补偿一段时间的数据
• specific-offsets：指定分区具体 offset，适合精确修复

我在做数据补偿时经常用到timestamp，比如上线后发现昨天有一个小时的统计有问题，想重算那一个小时，就设置'scan.startup.mode' = 'timestamp'和'scan.startup.timestamp-millis'为对应毫秒值，作业跑完再关掉。这个能力比重新跑整个历史数据省钱省力得多。

再说 Catalog。1.13 的 Catalog 机制让我可以把表结构统一管理起来，最常用的就是HiveCatalog。启用方式如下：

CREATE CATALOG hive_catalog WITH ( 'type' = 'hive', 'hive-conf-dir' = '/etc/hive/conf', 'default-database' = 'default' );

用了 HiveCatalog 之后，最大的便捷是表结构和分区信息都在 Hive Metastore 里，Flink 作业直接拿现成的表用，不用每个作业都重复写一长串 DDL。还有一个隐藏好处：如果你同时跑离线 Hive 任务和实时 Flink 任务，表结构能保持完全一致，下游分析时对字段语义的理解也一致。

4.3 流批切换的注意事项

1.13 里同一个 SQL 在流和批模式下跑，绝大多数情况下结果语义是保持一致的，这是社区花大力气做的“流批一体”。但实际切换时仍有一些细节需要注意。

第一，写法上要留意STATEMENT SET的用法。如果一个作业里有多条INSERT INTO，要写在一个STATEMENT SET里作为一条完整作业提交，否则每条INSERT会被解析成独立作业，资源各算各的。

BEGIN STATEMENT SET; INSERT INTO dwd_order_stat SELECT ... FROM orders GROUP BY ...; INSERT INTO dwd_user_stat SELECT ... FROM users GROUP BY ...; END;

第二，流模式下SELECT * FROM table会无限输出，如果你只是临时查一下数据，建议加LIMIT。但在 1.13 的流模式下LIMIT不生效，你得用TUMBLE窗口或者改成批模式跑。这是新同学经常困惑的地方：“为什么我LIMIT 10没用？”

第三，批和流的WATERMARK处理不同。批模式下就算你定义了 Watermark，它也不会像流模式那样影响窗口触发时机，因为批处理本身就是有界输入，Watermark 主要起一个“记录最大事件时间”的作用。我在做批量数据回填时，会单独准备一套没有 Watermark 的 SQL，避免语义混淆。

5. 常见问题排查实录

5.1 时区导致窗口数据错位

这是我在生产环境里遇到最多的问题，没有之一。现象是：Kafka 里有"2024-05-20 10:00:00"这样的字符串，我解析成事件时间后做 1 小时滚动窗口，发现结果里每小时的数据量和直接看原始时间戳统计的数据量对不上，总是差 8 小时。

排查步骤是这样的：

1. 先确认上游字符串的时区语义，是 UTC 还是东八区还是无时区
2. 确认table.local-time-zone配置
3. 用 SQL Client 跑一个简单的SELECT ts_ltz, CAST(ts_ltz AS STRING) FROM source_table看看解析后的值
4. 确认 Watermark 和窗口使用的是同一个字段

如果发现解析后的时间比预期少 8 小时或多了 8 小时，基本就是时区转换的问题。解决办法就是我在前面说的：要么用TO_TIMESTAMP_LTZ直接指定偏移，要么用AT LOCAL TIME ZONE转换。切记不要依赖隐式转换，字符串解析成TIMESTAMP不会带时区信息。

5.2 状态膨胀与 Checkpoint

流作业跑久了，状态越来越大，最终导致 Checkpoint 超时或者恢复时间过长。1.13 里给状态管理提供了不少工具，排查思路一般从两个方向入手。

第一个方向是检查 Keyed State 是否无限制增长。如果是窗口聚合，注意窗口结果是否长时间不被清理。比如天级窗口，如果 Watermark 不推进，窗口状态永远不会清理。这时候先看数据源是否还有新的 Watermark 发出，可以用Flink Web UI的 Watermark 指标确认。

第二个方向是合理设置 TTL（Time-to-Live）。1.13 里建议在启动 SQL 前先设置：

SET table.exec.state.ttl = 1h;

这条配置对 Flink SQL 作业中由 SQL 生成的有状态算子统一设置状态过期时间。需要注意，状态 TTL 设置得太短会导致 Join 丢数据，因为维表数据可能过期被清理，所以要根据业务容忍度调整。我一般先设置成业务窗口的 2 到 3 倍，等稳定后再逐步下调。

5.3 数据延迟与 Watermark 疑难

有一次我排查一个实时大屏数据不准的问题，指标是“最近 5 分钟订单量”。现象是数值偶尔会跳变，有时从 100 突然变成 80。后来发现是因为 Watermark 延迟设得太短（1 秒），而部分数据源网络抖动导致数据晚到 2 秒，这些晚到的数据触发了窗口的二次更新“撤回+重发”，下游大屏的累计值就产生了抖动。

这种问题在 1.13 里的处理方式比较清晰：

1. 明确业务允许的结果修正时间窗口
2. 配置合适的 Watermark 延迟，比如INTERVAL '5' SECOND
3. 合理设置窗口允许的延迟，让数据在一定时间范围内还可以修正
4. 下游展示端做好“更新”的逻辑，不要简单累加

如果你发现结果频繁跳动，先看侧输出流有多少被丢弃的迟到数据，再做窗口时长的配置调整。调参要有依据，我是通过收集一段时间的迟到数据量来估计延迟分布的，而不是一拍脑袋决定。

5.4 多表 Join 的常见坑

Join 在 Flink SQL 里非常常用，但也最容易出问题。1.13 里最稳的实时 Join 模式是使用INTERVAL时间范围让左右表在时间窗口内关联，比如：

SELECT o.order_id, o.user_id, u.user_name, o.amount FROM orders o JOIN users u ON o.user_id = u.user_id AND o.ts_ltz BETWEEN u.ts_ltz - INTERVAL '5' MINUTE AND u.ts_ltz + INTERVAL '5' MINUTE

为什么要加时间范围？因为实时流上的 Join 需要缓存数据，无界 Join 会把所有历史数据都放在状态里，既大又慢。加了时间范围之后，Flink 可以只保留这个范围内的状态，作业资源消耗大幅下降。

另一个坑是主键的声明。两张表 Join 的关联字段如果没有明确 PRIMARY KEY，一些情况下会退化成普通双流 Join，状态里为每条数据都保存副本，状态增长非常快。我在使用上游 Kafka 表时，至少会给主键表加上PRIMARY KEY NOT ENFORCED，保证 Flink 能识别 upsert 语义，状态相对可控。

6. 写在最后的几个小建议

6.1 SQL 里放什么，代码里放什么

1.13 虽然让 SQL 变强了，但不代表所有逻辑都该用 SQL 写。我现在的分工习惯是：数据源定义、清洗、聚合、Join、写出，全部用 SQL；自定义 UDF、复杂状态逻辑、精确的窗口触发控制，用 Java 写 UDF 或者单独 DataStream 算子。从维护性角度讲，这不只是“偷懒”，而是让每一种手段都在自己擅长的地方发挥作用。

举一个例子，我们有个场景需要从一条明细里解析出多个维度标签，并且不同标签的解析规则完全不一样。这种逻辑写成一大段 SQL CASE WHEN 会非常难读，但分别写成几个小的 UDF 就清楚得多。1.13 的 UDF 接口已经很稳定，ScalarFunction和TableFunction完全够用。

6.2 先在 SQL Client 里跑通再提交作业

这个习惯帮我省了无数时间。任何一条要上生产的 SQL，我都会先在 SQL Client 里用一小段测试数据跑通，确认字段类型、窗口语义、结果值都符合预期，再打包成作业提交。别小看这一步，SQL Client 的本地调试能力虽然和线上环境有差距，但语法校验、函数报错、类型不匹配这类问题都能提前暴露。

如果你用的 IDE 支持，也可以直接在测试环境用TableEnvironment跑，配合collect方法看结果。

6.3 把 Flink Web UI 当成第一排查工具

作业出问题时，我第一件事不是看日志，而是打开 Flink Web UI。JobGraph里能看到整个 SQL 被翻译成的算子链路，每个算子的recordsIn、recordsOut、Watermark都能看到实时值。如果某个算子recordsIn一直不涨，大概率是上游 Kafka 没数据或者 Watermark 不推进。如果某个算子的Watermark长时间停滞，那就要回头检查源表的 Watermark 定义。

1.13 的 Web UI 在展示 SQL 算子链路上已经相当清楚，每个算子都标注了对应的 SQL 片段。根据我的经验，80% 的线上问题都能通过 Web UI 上的指标初步定位，然后再针对性地看日志细节。

6.4 保持 SQL 版本升级的谨慎

如果你已经在使用 1.12 或更早版本，并且线上任务很多，别急着一次性全量升级到 1.13。我的做法是：先用部分非核心任务做小范围验证，重点对比两类任务：

• 有窗口聚合和状态清理逻辑的任务
• 使用了TIMESTAMP和字符串时间转换的任务

因为这两类最容易受到时区语义变化的影响。等到小范围验证通过、数据比对一致，再进行大范围迁移。兼容性方面，1.13 对 1.12 的大部分 SQL 是兼容的，但TIMESTAMP的语义严格化之后，一些之前靠“隐式本地时区”跑的逻辑可能会变，这是我在实际升级中唯一遇到过需要改动的地方。

7. 从 1.13 往后看的选型思路

现在回看 Flink SQL 1.13，它解决的问题不是“把 SQL 变得多炫酷”，而是把流批一体这个理念真正落到了语法和语义层面。我后来也简单看过后续版本的功能演进，时区、动态表、Join 这些基础能力在 1.13 打下的底子一直沿用下来，这也说明当时的选型思路没有跑偏。

对于刚开始接触 Flink SQL 的团队，我仍然建议从 1.13 这条线切入，先把时区、动态表、窗口、Join 这几个核心概念吃透。这些概念在各个版本里是相通的，把它们搞明白了，后续无论升级到哪个版本，你都能比较平滑地过渡。别急着追新特性，先把最核心的思维方式建立起来，写出来的作业质量会完全不一样。