做实时计算和实时数仓的同学这两年多多少少都听过这句话能上 Flink SQL 就尽量别自己写 DataStream。而聊到 Flink SQL 社区版本1.13 绝对是一个绕不开的里程碑。很多团队从 1.10、1.11 一夜之间升到 1.13是因为这个版本把“动态表”的心智模型、时态表和维表 Join 的语法补完整了也让窗口聚合从老式 GROUP BY 语法平滑切换到了新的函数式语法。今天这篇就从一位实际用 1.13 跑了近一年生产任务的博主视角把 Flink SQL 1.13 的定位、核心特性、实操路径和踩坑记录一次性讲透。这篇文章适合谁刚接手 Flink SQL 或者准备升级 1.13 的开发、负责实时数仓平台运维的工程师以及想在 SQL 层做流批一体的架构师。你要什么基础会一点 SQL知道 DataStream 是什么大概就能跟上。我会把很多概念都拆开讲尽量不空谈原理直接谈“我在什么场景下怎么选”。1. 1.13 在 Flink 生态中的定位与整体设计思路1.1 从 1.12 到 1.13流批一体走向实用的关键一步Flink 在 1.11 到 1.12 之间慢慢把 DataSet 边缘化把批处理统一到 DataStream 执行引擎里这是很伤筋动骨的事。到了 1.13流批一体的落地感才真正出来。你可以用同一张源表、同一段 SQL在流模式下做实时计算在批模式下直接做全量回溯。这不只是 API 层面统一Table API/SQL 层的 planner 也统一了处理逻辑和优化规则对两种模式基本一致。从我自己的感受看1.13 最大的变化不是新增了一堆函数而是把很多“网上搜不到标准写法”的用法正式化了。比如算子级 TTL、列式查询优化、窗口表值函数、时态表函数。以前这些要么靠外围工具绕要么靠 DataStream 算子自己实现现在只要在 SQL 里配置参数就能完成。生产环境真正需要的能力被慢慢揉进了 SQL 引擎这才是 1.13 值得升级的根本原因。1.2 为何 1.13 是许多团队升级的“及格线”很多团队还在用 1.11 甚至 1.10最大原因是“跑得好好的一动就出事”。但 1.13 在功能完整性和执行稳定性上跨了一截社区里大部分“Flink SQL 实战经验帖”也都默认基于 1.13。主要原因有三条。第一1.13 的 Planner 错误信息比之前友好很多很多 SQL 校验错误能直接定位到是哪段表达式、哪个函数签名出了问题排查效率高了不少。第二窗口 TVF 语法彻底落地对流式聚合、累计窗口、会话窗口的写法形成了统一范式你不太需要再记两种不同的 Group By 语义。第三Flink 社区对资源的配置模型在 1.13 做了收敛新参数的命名更规范无论你提交到 Standalone、YARN 还是 K8s语义差异都比旧版本小。也就是说对生产团队来说1.13 算是一个“重新出发”的版本老版本的坑你要避开新特性你能直接用上下游生态也基本适配到位。2. 核心特性拆解动态表、时态表与连接器体系2.1 动态表SQL 处理无界流的心智模型理解 Flink SQL 1.13第一个绕不开的词叫动态表。传统数据库的表是一组静态行但流式场景里数据无限到达我们需要一种能持续变化、又能在 SQL 语法里被描述的关系。Flink 的处理方式是把流映射到一张“随时间不断追加或更新”的虚拟表每一次查询都会按照当前已到达的数据集计算一次结果结果也随时间变化。举个例子你要从 Kafka 读取用户行为日志用CREATE TABLE定义它的字段、数据类型和Watermark再对这张表执行SELECT user_id, COUNT(*) FROM user_actions GROUP BY user_id。在传统 SQL 里这是一个普通查询在 Flink SQL 中它变成了一个“连续查询”集群会一直运行它每当 Kafka 有新数据聚合结果就更新一次。这种连续查询的底层执行方式有两种Append 流和更新流。前者只追加适合日志类输出后者可以更新之前发出的结果适合写数据库或构建实时指标。实际操作时必须注意动态表结果并不总能映射成 Append 流。带GROUP BY的聚合、带JOIN且右表会更新的场景都是更新流。定义 Sink 连接器时要明确你的下游能不能接受 retract 消息否则会出现写入重复或者旧数据不删除的问题。比如写 Kafka如果只是 Append 模式但你的 SQL 结果是更新流运行时会报错写 MySQL 则要等primary key和 Upsert 语义都对齐1.13 的 upsert-kafka 和 JDBC 连接器对此处理得已经很成熟。2.2 时态表与时态 Join补齐实时维表查询的一块拼图Flink SQL 1.13 里非常实用的一块是时态表Temporal Table。时态表可以简单理解为一张“随时间变化且每个时间点都能查询当时快照”的表。它特别适合两类场景拉链表版本关联以及实时维表关联。时态表分两种事件时间时态表和处理时间时态表。事件时间时态表的数据包含有效时间区间你用FOR SYSTEM_TIME AS OF去关联某条数据发生时对应的版本。处理时间时态表则对应“当前最新状态”这是实时维表非常常见的做法。比如你有一个用户维度表存放在 MySQL 里每分钟同步一次更新流里的订单要补上最新的用户信息就可以把 MySQL 表定义为一张处理时间时态表用 SQL 做 Lookup JoinFlink 会按照每条记录的处理时间去维表里查一次。但这里有个很容易掉进去的坑如果直接用同步查询维表每条数据都触发一次数据库请求QPS 一高数据库就容易被拖垮。1.13 对 Lookup Join 的支持已经比较完善可以在WITH参数里配置缓存。这里说一个我常用的配置组合CREATE TABLE dim_user ( user_id BIGINT, user_name STRING, level STRING, PRIMARY KEY (user_id) NOT ENFORCED ) WITH ( connector jdbc, url jdbc:mysql://xxx:3306/dim, table-name dim_user, lookup.cache.max-rows 10000, lookup.cache.ttl 30 min, lookup.max-retries 3 );lookup.cache.ttl是缓存过期时间lookup.cache.max-rows是缓存条目上限。这两个参数要按数据量级和实时性需求折中TTL 太短数据库压力大TTL 太长指标里会出现延迟更新的维度信息。我一般先把 TTL 设为 10 到 30 分钟维表变更不频繁放长一些变更频繁就缩短。2.3 1.13 的连接器要点从源到维表再到数据湖Flink SQL 1.13 对连接器的覆盖已经相当全面。官方连接器里有 Kafka、JDBC、Elasticsearch、HBase、Hive、FileSystem加上社区常用的 Pulsar、Doris、ClickHouse、Iceberg、Hudi 等基本能在 SQL 层完成全链路开发。这里有个经验我建议把所有外部系统都先用EXPLAIN或小数据量试跑验一遍连接器语义尤其是 Kafka 连接器的scan.startup.mode和format。1.13 的 Kafka SQL 连接器支持earliest-offset、latest-offset、group-offsets、specific-offsets和timestamp五种启动模式。线上最稳妥的配置通常是group-offsets配合独立的 Consumer Group 管理位点但要注意并发度变化会导致 group rebalance造成短暂重复消费。如果对数据准确性敏感建议用specific-offsets或者timestamp让作业从指定位置开始。另外1.13 的upsert-kafka连接器非常值得一用。它把 Kafka 当成 Key-Value Store利用消息 key 做去重和更新。你可以在 Kafka 里既写更新流也保留 changelog 语义下游再用 JDBC sink 落库。对于实时数仓的分层加工这比直接用普通 JSON 格式更灵活。3. 实操过程从零跑通一个 Flink SQL 任务3.1 环境与依赖准备先把环境搭起来。Flink 1.13 支持 Java 8 和 Java 11我用的是 Java 8 配合 Maven。如果你是裸跑本地模式只需要官方发布的二进制包解压后直接能跑。生产上多半是通过 Flink CLI 或者平台把 SQL 提交到集群我会先在本机把 SQL 调试稳定。依赖上最省心的是把需要的 connector 放到lib/目录。需要注意一点有些团队把 flink-sql-connector-kafka 的 jar 都塞进lib然后多个作业共用这会导致依赖冲突。更干净的做法是用-C参数或者设置flink.download-provider等方式隔离依赖。但在 1.13 的默认 CLI 下最简单的还是把 connector 打包到作业 jar 里或者统一放在lib但确保版本一致。还有一处容易漏如果你要用 JSON 格式和 Debezium/Canal 格式需要分别引入flink-json和flink-connector-kafka。1.13 老版本的某些连接器 jar 已经改名一定要认准 flink-sql-connector-而不是老的 flink-connector-。3.2 建表语法与连接器配置直接给一段典型的源表定义。假设我们要处理 Kafka 里的订单消息消息是 JSON 格式包含订单ID、用户ID、金额、创建时间。CREATE TABLE orders ( order_id BIGINT, user_id BIGINT, amount DECIMAL(10, 2), create_time TIMESTAMP(3), WATERMARK FOR create_time AS create_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic ods_orders, properties.bootstrap.servers kafka:9092, properties.group.id flink_orders_group, scan.startup.mode group-offsets, format json );这段 SQL 里有两个创建表时必须注意的细节。第一WATERMARK FOR create_time AS create_time - INTERVAL 5 SECOND。这个 watermark 声明告诉 Flink数据里的事件时间最多延迟 5 秒超过 5 秒的记录会被视为迟到数据默认会被丢弃或不参与计算。真实业务中如果上游存在长时间故障或乱序严重5 秒可能不够需要结合 Kafka 消息时间戳和延迟指标一起判断。宁可多留一点空闲期也不要因为水位推进太慢导致窗口不触发。第二字段类型。DECIMAL(10, 2)在 Flink SQL 里对应 BigDecimal反序列化时如果 JSON 里是数字字符串可能会报错。1.13 在格式校验上比旧版严格建议上游统一输出为数字类型或者在 SQL 里用CAST做显式转换。来看 Sink。如果要把聚合结果同步到 MySQL可以这样配置CREATE TABLE order_stats ( user_id BIGINT, total_amount DECIMAL(10, 2), order_cnt BIGINT, PRIMARY KEY (user_id) NOT ENFORCED ) WITH ( connector jdbc, url jdbc:mysql://mysql:3306/test, table-name order_stats, username root, password ****** );PRIMARY KEY ... NOT ENFORCED表示 Flink 不会自己校验唯一性只把它作为该表的逻辑主键信息。JDBC sink 会根据主键自动生成INSERT ... ON DUPLICATE KEY UPDATE语句从而实现 Upsert。3.3 执行语义与参数调优Flink SQL 1.13 作业的合理调优我认为要先明确执行模式再谈并行度和状态。你可以通过SET execution.checkpointing.interval 1 min设置检查点间隔如果要做精确一次语义还需要配置下游连接器的语义支持。Kafka Sink 在 1.13 默认是至少一次需要额外加上sink.semantic exactly-once才能实现两阶段提交。并行度设置上一个常见错误的做法是盲目把parallelism.default调到很大。实际上并行度要按数据量和算子类型设计。比如 source 并行度通常建议与 kafka topic 分区数保持一致这样每个分区由同一个 task 读取维护 consumer 线程方便窗口算子并行度受 key 分布影响如果某个 key 数据巨大会出现单个 task 数据倾斜需要去考虑加盐或按二级 key 拆分。另外1.13 里table.exec.state.ttl和table.exec.mini-batch.enabled是 SQL 调优使用频率最高的参数。Mini-batch 能在一定延迟范围内大幅减少聚合的更新次数适合高吞吐的 COUNT/SUM。我通常这样用SET table.exec.mini-batch.enabled true; SET table.exec.mini-batch.allow-latency 5 s; SET table.exec.mini-batch.size 10000;三个参数合在一起就是每 5 秒或积攒 10000 条数据后触发一次微型批处理。这个机制可以极大减少下游写入频率但要注意它会让结果输出延迟增大不适合对实时性要求特别高的图表场景。3.4 提交部署时的常见配置1.13 支持 Session 模式和 Per-Job 模式提交。我建议生产环境优先用 Per-Job 模式每个任务拥有独立 JobManager 和资源配置避免多个任务争抢导致全链路雪崩。提交时最关键的是设置好 JobManager 内存和 TaskManager 内存。Flink 1.13 的内存模型里taskmanager.memory.process.size和taskmanager.memory.managed.size很容易搞混。process.size是进程总内存managed.size是给 Flink 内部数据结构比如排序、缓存用的堆内/堆外内存。一个经验1.13 的 Python 和 SQL 用户如果发现任务频繁 GC 或反压不要第一反应是加并发。先看 Managed Memory 够不够。窗内排序和 Join 会把中间数据放在 Managed Memory默认比例是 40%。对于大窗口大状态任务可以把taskmanager.memory.managed.fraction调高到 0.6 甚至更高同时适当减小 heap 比例。提交命令通常是flink run-application -t yarn-application \ -Djobmanager.memory.process.size2048m \ -Dtaskmanager.memory.process.size4096m \ -Dtaskmanager.numberOfTaskSlots2 \ -c org.apache.flink.table.client.SqlRunner \ /path/to/flink-sql-runner.jar /path/to/your_sql_file.sql如果你用平台那逻辑也一样把 SQL 文件或者文本脚本交给平台执行。4. 高频问题与排查技巧实录4.1 状态大小与控制参数1.13 的状态是大家最关心的点。状态太大就会导致 checkpoint 慢、恢复慢、反压。常见解决路径是配置 TTL把不需要保留的 key 删除掉。在 SQL 层面聚合查询产生的状态默认不会永久保留但需要你显式设置过期时间。table.exec.state.ttl可以设置状态存活时间比如SET table.exec.state.ttl 7 d。这能限制每个 key 的聚合状态存在 7 天。另一个是连接器或函数内部的自定义状态可能就不会受这个参数控制需要用 DataStream API 的 TTL 配置单独处理。我踩过的一个坑是用COUNT(DISTINCT user_id)这种高基数指标状态膨胀极快。原因是 Flink 需要在状态里维护一个完整的 set 集合每个独立 key 都占用内存。1.13 并没有内置近似去重函数替代方案是用 HyperLogLog 这类外部实现或者先按天、按小时做分桶去重再在更上层汇总。有几个社区 UDF 可以直接用但务必自己压测。4.2 时间字段与 Watermark 的坑时间类型是 Flink SQL 新手最容易卡住的地方。Kafka 消息里的时间字段可能是字符串也可能是长整型 epoch millis。建表时你要先声明成BIGINT再用TO_TIMESTAMP_LTZ(field, 3)转换或者直接声明为TIMESTAMP(3)。1.13 推荐用TIMESTAMP_LTZ表示带时区的绝对时间尤其在TIMESTAMP切换夏令时的时候旧写法很可能出现错位。还有一个高频错误多个 source 表时间类型不一致。比如一张订单表的create_time是TIMESTAMP(3)另一张流的op_time是TIMESTAMP_LTZ(3)。两者直接进行 Temporal Join 时配套时间属性经常对不上。我的建议是统一用TIMESTAMP_LTZ(3)并在连接器端就完成转换。Watermark 的另一个坑是空闲分区。假设 Kafka topic 有 8 个分区其中 3 个分区没有数据Flink 默认 Watermark 要等所有输入分区都推进到某个时间才会推进到该时间。于是那 3 个空分区会卡住整个窗口触发。1.13 支持设置空闲源超时table.exec.source.idle-timeout把它设为 1 分钟或更短空分区超过这段时间后就标记为空闲不再阻挡 watermark 推进。4.3 序列化、数据类型与空值问题SQL 任务跑起来不报错但结果总少几条这个问题我觉得必须单独提。很大原因是格式解析时的null处理。比如 JSON 数据里有个字段没有值Kafka 连接器默认把缺失字段解析成null。如果你的查询里有amount * 0.1那这一行结果就是null再配合聚合或者写入很容易被忽略或导致数据不准确。接 JSON 数据时要么在建表时给字段配置默认值要么在 SQL 里用COALESCE(amount, 0)提前清洗。类型映射的问题也一样。MySQL 的TINYINT映射到 Flink 可能是TINYINT而有些 BI 读到的是无符号类型1.13 在连接器层有严格的字节映射CAST(balance AS BIGINT)有时候能救急但最保险的还是修改上游表结构把字段类型统一到更大范围。另外如果你看到org.apache.flink.table.api.ValidationException: Type is not supported多半是自定义 UDF 使用了不受支持的 Java 类型。1.13 里 UDF 的参数类型要尽量用基础类型比如int用Integer数组用int[]而不是ListInteger避免类型推断失败。4.4 1.13 升级后的兼容性注意很多团队是从 1.11 直接跳上来的这里有几个不兼容点要提前检查。第一flink-table依赖的包名和之前不一样老代码如果直接按org.apache.flink.table.api.scala引用需要改成org.apache.flink.table.api.bridge.scala。第二一些旧的 planner 参数在 1.13 改名了比如planner相关的配置执行时会提示已弃用最好提前清理。第三如果自定义了连接器或 format需要从代码层面适配新的 SPI 接口不能直接拿老 jar 放在 lib 下用我遇到过因为缺少工厂接口导致表找不到连接器的异常。升级顺序我建议是先在一个独立集群把 SQL 语法试跑再跑数据对账。重点核对三类任务含JOIN的维表任务、含窗口的聚合任务、写 Kafka 的 Upsert 任务。这三类在 1.13 里语义变化最明显最容易在升级后续变形。4.5 算子反压与资源瓶颈排查反压在 Flink Web UI 上可以直接看到。当一个算子长时间处于反压状态先别急着调并行度要看数据源端是不是已经有积压。从 Kafka 到 SQL 作业如果 source 端并行度不够就会表现为部分分区堆积部分分区很空闲。1.13 支持在 SQL 的WITH里设置scan.internel.spill之类的底层参数但这类参数改起来风险高我基本不动优先把 source 并行度增加到 topic 分区数的整数倍比如 8 个分区并行度设为 4 或 8。还有一个容易忽略的瓶颈是 JDBC sink。如果你下游 MySQL 单表写入吞吐有限不管你怎么加 Flink 并行度都会在 sink 端堆积。1.13 的 JDBC sink 支持攒批参数比如sink.buffer-flush.max-rows 1000, sink.buffer-flush.interval 2s这两个参数能让写入分批执行减少反复建连。但对千亿级指标写入最好直接切到批量导入工具或数据湖方案JDBC 只做单条低延迟场景。5. 我把 1.13 用于实时数仓后的一些实际体会先把结论摆出来Flink SQL 1.13 是我眼中“可大规模生产化”的 Flink SQL 起点别再纠结要不要从老版本升级。如果你正在做一个实时数仓架构大概率会是这样Kafka 放 ODS 层Flink SQL 做 DWD/ADS 加工结果写到消息队列、MySQL、Olap 数据库或数据湖。1.13 在这个链条里的角色就是最灵活的加工引擎它让你用表结构描述整个数据流把流计算和批处理尽量藏在 SQL 后面。我个人的使用经验中最值钱的一条是“把业务规则尽量下沉到 SQL 建表语句里”。比如主键约束、时间属性、Watermark、连接器参数都先在建表阶段定义清楚后续查询代码就可以干净很多。相反如果建表阶段偷懒把字段全部声明成 STRING后面到处CAST一旦数据量上来解析开销会明显放大还没有任何类型收益。我用 1.13 跑过一天几十亿条数据的场景绝大多数性能问题都是出在类型定义混乱而不是引擎本身。第二点感受是1.13 适合你逐步把老任务用 SQL 重写。不要一次性迁移所有作业。可以挑一个最简单的值监控指标从 DataStream 改成 Flink SQL跑通后再慢慢增加 JOIN、窗口、维表关联。好处是你能熟悉新语法同时降低风险。我在团队内部也是这个策略先把实时大屏指标切过去稳定两周后再动核心数仓链路。如果你还想继续深挖可以沿着这几个方向扩展一是结合 Flink CDC 把 MySQL 8.0 的 changelog 直接接入配合 upsert-kafka 实现实时离线一体数仓二是用 Iceberg 或 Hudi 做批流读写让 Flink SQL 的数据成果都有统一的存储底座三是把常用的 SQL 模板抽象出来做成平台上的“实时计算任务包”后面新人直接填空式开发。篇幅所限今天先聊到这儿。最后再说一个实在的建议任何新 SQL 任务上线前一定要准备一份数据对账脚本用批模式跑同一份历史数据对比两次聚合结果。这个流程救过我太多次了。