Flink SQL实时数仓案例详解：从Kafka链路到窗口Join的完整实践

我第一次系统学 Flink SQL 的时候，卡住我的不是语法，而是手里没有一套能从头到尾跑通的完整例子。文档里的 SQL 片段单独看都对，真要把它们串成一条从 Kafka 到结果表的实时数仓链路，全是坑：连接器版本对不上、Kafka 里根本没有数据、窗口结果迟迟不输出、时区差了八个小时。这个仓库最开始只是我自己调试用的脚本集合，后来整理成了一个适合实时数仓学习和面试的 Flink SQL 案例仓库，每个案例都配了数据生成器、完整 SQL 脚本和结果核对说明，clone 下来就能直接跑。

这篇文章我会把这个仓库的设计思路、两个核心案例的完整拆解、面试怎么用、以及我在调试过程中反复踩过的坑全部写出来。如果你正在学 Flink SQL，或者准备面实时数仓岗位，这篇文章应该能帮你少走很多弯路。

1. 从"看懂SQL"到"跑通链路"：案例仓库的定位

1.1 你大概率也掉进过的学习断层

先说一个我在社区里经常看到的问题：很多人能看懂 Flink SQL 的语法，官方文档里的连接器参数、窗口函数、Join 语法都读过，但让他自己从零搭一条实时数据处理链路，就不知道从哪下手了。

仔细想想，这个断层主要来自三方面。第一，文档是碎片化的。Flink 官网的 Kafka 连接器页面只讲 Kafka 连接器，窗口聚合页面只讲窗口聚合，语法层面每页都对，但没人告诉你这两段 SQL 怎么拼到一起、中间的 watermark 怎么设计、结果表用 print 还是 upsert-kafka。第二，网上能找到的案例普遍版本偏老，拿 Flink 1.12 时代的写法套到 1.18 上，作业能提交但行为完全不同，折腾半天以为是自己的问题。第三，也是最重要的一点，官方 QuickStart 太玩具了，datagen 生成数据、print 输出结果，跑通了也不知道这个结果对不对，更不知道真实业务里数据从哪里来、要写到哪里去。

所以我在整理这个仓库时定的第一个原则就是：每个案例必须是一个完整的数据链路闭环——有输入数据的生成逻辑、有处理 SQL、有可验证的输出结果，而不是孤立的语法片段。

1.2 仓库的定位：帮你在"数据流动"里建立手感

这个仓库面向三类人。第一类是刚看完 Flink 基础语法、想动手跑通一个真实场景的自学者；第二类是准备实时数仓岗位面试、需要一个能讲清楚的项目案例的候选人；第三类是团队里要做技术预研、想快速验证某个方案可行性的开发者。

仓库里所有案例共用一套本地环境，Docker Compose 一键启动，里面有 Flink、Kafka、MySQL 这几个核心组件。数据生成器用 Python 写，模拟电商订单、支付流水、用户日志这类常见业务数据，往 Kafka 里灌。SQL 脚本直接提交到 Flink SQL Client 里执行，结果落到 Kafka 或者 MySQL，然后用脚本去核对。

我的建议是：不要把这个仓库当成"答案"，而是当成"参照物"。每个案例的 README 里都写了"这个案例在真实业务中对应什么问题""为什么这样设计"——你先自己写一遍，再对照看差别，收获会大得多。

2. 仓库结构：按实时数仓处理脉络组织的三层案例

2.1 目录设计

仓库的目录结构很简单，你 clone 下来之后看到的是这样：

rtdw-demo/ ├── docker/ │ ├── docker-compose.yml │ └── mysql/ │ └── init.sql ├──>version: '3' services: mysql: image: mysql:8.0 ports: - "3306:3306" environment: MYSQL_ROOT_PASSWORD: "123456" MYSQL_DATABASE: "rt_dw" volumes: - ./mysql/init.sql:/docker-entrypoint-initdb.d/init.sql:ro kafka: image: bitnami/kafka:3.6 ports: - "9092:9092" environment: - KAFKA_CFG_NODE_ID=0 - KAFKA_CFG_PROCESS_ROLES=controller,broker - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093 - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - ALLOW_PLAINTEXT_LISTENER=yes flink-jobmanager: image: flink:1.18.1 ports: - "8081:8081" command: jobmanager environment: - FLINK_PROPERTIES=jobmanager.rpc.address: flink-jobmanager volumes: - ./flink-lib:/opt/flink/lib flink-taskmanager: image: flink:1.18.1 command: taskmanager environment: - FLINK_PROPERTIES=jobmanager.rpc.address: flink-jobmanager volumes: - ./flink-lib:/opt/flink/lib

启动命令很简单，在 docker 目录下执行 docker compose up -d，等两三分钟让 Kafka 和 Flink 都进入健康状态。这里有个我之前吃过亏的地方：Flink 连接 Kafka 需要的 connector jar 包，Flink 发行版里是不自带的。你要先去下载 flink-sql-connector-kafka 对应版本的 jar，放到 docker/flink-lib 目录里，否则 SQL 作业能提交，但 source 端永远读不到数据。这个坑我后面还会专门讲。

3.2 生成订单数据：模拟真实业务的长尾分布

环境起来之后，我们来看数据生成器。gen_order.py 的核心逻辑很简单，就是往 Kafka 的 order_topic 里持续写 JSON 格式的订单事件。

import json import random import time from datetime import datetime from kafka import KafkaProducer users = list(range(1, 10001)) # 模拟真实场景：大量小额订单，少量大额订单 amounts = [round(random.lognormvariate(3, 0.8), 2) for _ in range(10000)] producer = KafkaProducer( bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode() ) order_seq = 0 while True: order_seq += 1 event = { 'order_id': 'ORD%08d' % order_seq, 'user_id': random.choice(users), 'amount': round(random.choice(amounts), 2), 'order_time': datetime.now().isoformat(timespec='milliseconds') } producer.send('order_topic', event) time.sleep(0.1)

这段代码有两个设计点值得注意。一是金额用对数正态分布生成，这样大部分订单金额落在几十到几百之间，偶尔出现几千的大单，接近真实电商场景。二是通过 time.sleep(0.1) 控制发送速率，相当于每秒写 10 条订单，窗口聚合的结果肉眼可见地滚动。

3.3 建表 SQL 与分窗聚合

接下来是核心的 SQL。第一步，在 Flink SQL 里创建订单源表，声明 Kafka 连接信息和 JSON 格式，同时给 order_time 字段定义 watermark，允许最多 5 秒的乱序数据。

CREATE TABLE order_topic ( order_id STRING, user_id BIGINT, amount DECIMAL(10, 2), order_time TIMESTAMP(3), WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'order_topic', 'properties.bootstrap.servers' = 'kafka:9092', 'properties.group.id' = 'order_group', 'format' = 'json', 'scan.startup.mode' = 'earliest-offset' );

第二步，创建结果表。这里我用 upsert-kafka 连接器，按窗口的起止时间作为主键，把聚合结果写回 Kafka。用 upsert-kafka 而不是普通 kafka，是因为窗口结果在迟到数据到来时会更新，upsert 语义能保证下游拿到的是最新值。

CREATE TABLE order_minute_sink ( window_start TIMESTAMP(3), window_end TIMESTAMP(3), order_count BIGINT, total_amount DECIMAL(12, 2), PRIMARY KEY (window_start, window_end) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'order_minute_result', 'properties.bootstrap.servers' = 'kafka:9092', 'key.format' = 'json', 'value.format' = 'json' );

第三步，执行窗口聚合。GROUP BY TUMBLE(order_time, INTERVAL '1' MINUTE) 表示按 1 分钟的滚动窗口分组，TUMBLE_START 和 TUMBLE_END 取出窗口的起止边界。

INSERT INTO order_minute_sink SELECT TUMBLE_START(order_time, INTERVAL '1' MINUTE) AS window_start, TUMBLE_END(order_time, INTERVAL '1' MINUTE) AS window_end, COUNT(order_id) AS order_count, SUM(amount) AS total_amount FROM order_topic GROUP BY TUMBLE(order_time, INTERVAL '1' MINUTE);

3.4 提交作业和验证结果

启动生成器，然后进入 Flink SQL Client 提交脚本：

docker exec -it flink-jobmanager ./bin/sql-client.sh -f /opt/flink/sql/order_minute_gmv.sql

作业提交后，你会在 Flink Web UI 的 8081 端口看到作业在 Running 状态。验证结果有两种方式。第一种是直接消费结果 topic：

docker exec -it kafka kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic order_minute_result \ --from-beginning

你会看到类似这样的输出：

{"window_start":"2024-11-20 15:01:00","window_end":"2024-11-20 15:02:00","order_count":612,"total_amount":48723.50} {"window_start":"2024-11-20 15:01:00","window_end":"2024-11-20 15:02:00","order_count":613,"total_amount":48901.25}

注意同一窗口出现了两条数据，这就是 upsert 的效果：第 612 条订单后又有迟到的订单进入窗口，结果被更新了。如果当时用普通 kafka sink，下游就会看到两个独立事件，语义完全不同。

3.5 如何"确认结果是正确的"

这一小节我觉得是最重要的。实时任务和离线任务最大的区别是：离线任务跑完你拿结果和 Hive 表对一下就行，实时任务没有"最终"一说，窗口结果会不断更新，你到底信哪个数？

我的核对思路是"三层核对法"。第一层核对源头，直接消费 Kafka 原数据 topic，统计一个时间段内写入多少条、金额总和是多少；第二层核对中间逻辑，打开 Flink Web UI 看算子接收和输出的 record 数，确认没有数据在某个算子被莫名丢弃；第三层核对结果，消费结果 topic，在窗口结束后等一个超过 watermark 的时间，再取最终值，和离线预计算的同一窗口总量做对比。

三层都能对上，基本可以认为这条链路是可信的。如果对不上，就逐层排查，看断层出在哪一层。这个方法后来也成了我面试时经常讲的点——面试官很在意候选人有没有"结果可信"的意识。

4. 实战拆解二：订单流与支付流 Join 出明细宽表

4.1 先想清楚为什么需要双流 Join

第一个案例是 DWS 层的分钟级指标，第二个案例要下沉到 DWD 层，做明细宽表。业务场景是这样的：订单和支付是两条独立的 Kafka topic，订单流里只有订单金额，支付流里才有支付流水号和支付时间。我们要把这两条流按订单 ID 关联在一起，再补齐用户维表信息，输出一条包含订单、支付、用户信息的完整明细。

直接在流上做等值 join 是不行的，因为两条流里同一笔订单的出现时间不一样，订单先产生，支付要过一会儿才到，两边永远不在同一时刻出现在窗口里。Flink SQL 里有三种流上关联方式，适用场景各有不同。

订单和支付这条场景最适合用 Interval Join，因为订单产生时间和支付时间天然有一个前后间隔，我们可以用时间条件来兜底，只关联 10 分钟内能配上对的订单和支付。

4.2 核心 SQL：订单流+支付流的 Interval Join

先创建支付流源表，同样定义 watermark。支付流的 watermark 基于 pay_time 字段，和订单流的 order_time 是两条独立的时间线。

CREATE TABLE payment_topic ( payment_id STRING, order_id STRING, pay_amount DECIMAL(10, 2), pay_time TIMESTAMP(3), WATERMARK FOR pay_time AS pay_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'payment_topic', 'properties.bootstrap.servers' = 'kafka:9092', 'format' = 'json', 'scan.startup.mode' = 'earliest-offset' );

然后建一个视图，把订单流和支付流做 Interval Join。注意这里的 ON 条件有两个部分：等值条件 order_id，以及时间条件 pay_time BETWEEN order_time - 10 分钟 AND order_time + 10 分钟。Flink 要求 Interval Join 的时间边界必须写成这种对称的 BETWEEN 形式，目的是让两条流在各自的事件时间里都能找到匹配区间。

CREATE VIEW order_payment_detail AS SELECT o.order_id, o.user_id, o.amount, p.payment_id, p.pay_time FROM order_topic AS o JOIN payment_topic AS p ON o.order_id = p.order_id AND p.pay_time BETWEEN o.order_time - INTERVAL '10' MINUTE AND o.order_time + INTERVAL '10' MINUTE;

这里有一个很多人第一次接触会疑惑的点：Flink 是怎么实现这个 join 的？实际上 Flink 会在状态里同时缓存订单流和支付流的数据，每来一条支付数据，就去状态里找对应的订单数据；每来一条订单数据，也去状态里找对应的支付数据。时间条件的作用是给状态里的过期数据定一个清理边界，避免状态无限增长。

4.3 再接一次维表，拼出完整用户画像

订单和支付关联完之后，我们还需要用户维表信息。在真实的实时数仓里，用户维表存在 MySQL 或者 HBase 里，订单明细里只有 user_id，没有用户名和城市，需要通过 Lookup Join 动态查询补齐。

在 Flink SQL 里创建一张 JDBC 维表：

CREATE TABLE dim_user ( user_id BIGINT PRIMARY KEY NOT ENFORCED, user_name STRING, city STRING ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://mysql:3306/rt_dw', 'table-name' = 'dim_user', 'username' = 'root', 'password' = '123456', 'lookup.cache.max-rows' = '10000', 'lookup.cache.ttl' = '10min' );

然后和前面的 join 结果做 Lookup Join。FOR SYSTEM_TIME AS OF 是 Lookup Join 的固定语法，语义是"在这一条明细数据的业务时间里，去维表里查它当时对应的维表快照"。

SELECT d.order_id, d.user_id, d.amount, d.pay_time, u.user_name, u.city FROM order_payment_detail AS d JOIN dim_user FOR SYSTEM_TIME AS OF d.pay_time AS u ON d.user_id = u.user_id;

Lookup Join 看起来简单，但性能上有一个大坑：如果不加缓存，Flink 会对每一条明细数据实时发起一次 JDBC 查询，数据量一上来，MySQL 的连接池直接被击穿。所以我在建表语句里加了 lookup.cache.max-rows 和 lookup.cache.ttl，把最热的 1 万个用户缓存 10 分钟，查不到的才穿透到 MySQL。代价是维表更新最多有 10 分钟延迟，业务上能接受就无所谓——这个取舍在面试里也是一个很自然的讨论点。

4.4 宽表结果写到哪里

这个案例的结果表我建议直接写 MySQL，用 upsert 语义。既然明细宽表的主键是 order_id，就可以把 MySQL 结果表的主键定义为 order_id，Flink 的 JDBC sink 支持更新插入：

CREATE TABLE order_wide_sink ( order_id STRING PRIMARY KEY NOT ENFORCED, user_id BIGINT, user_name STRING, city STRING, amount DECIMAL(10, 2), payment_id STRING, pay_time TIMESTAMP(3) ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://mysql:3306/rt_dw', 'table-name' = 'order_wide_sink', 'username' = 'root', 'password' = '123456' );

写完宽表之后，下游的报表、推荐、风控都直接查 MySQL 这张表，这就是典型的 DWD 层明细宽表建模思路。你在面试里能把这个链路完整讲出来，比背一百个概念都有说服力。

5. 面试时怎么把仓库案例讲成自己的方法论

5.1 面试官问"请讲一个实时计算项目"时的应答结构

很多人面试讲到实时数仓项目，容易陷入流水账：我们用了 Flink SQL、接了 Kafka、做了窗口统计、写到了 MySQL。讲完面试官根本记不住。我自己的经验是：项目介绍必须有一个"从问题到决策"的主线，不能只罗列技术名词。

我通常用四步结构。第一步讲业务背景和数据规模，比如"每分钟大约有几千笔订单，业务方需要实时看到 GMV 和订单量"；第二步讲技术选型，重点是"为什么用 Flink SQL 而不是 DataStream API"，我会说团队以 SQL 开发为主、状态管理和时间语义需要 Flink、SQL 更容易维护；第三步讲链路设计，从 Kafka 到源表、窗口计算、结果表，这时候就把仓库里的案例细节拿出来；第四步讲你踩过的坑和解决思路，比如时区问题、维表缓存问题、数据核对问题。

这个结构的好处是：面试官从头到尾听到的都是"为什么"，而不是"用什么"。

5.2 高频追问和可用案例素材

项目介绍完之后，面试官通常会顺着追问。我整理了四个最常见的追问方向，以及仓库里对应的案例素材。

第一个追问是"窗口怎么处理迟到数据"。你直接讲第一个案例里的 watermark 设计和 upsert-kafka 结果表，说明迟到数据会更新窗口结果，结果表的 upsert 语义保证下游读到最新值。

第二个追问是"维表数据怎么和流关联"。你讲第二个案例里的 Lookup Join，重点说明缓存参数怎么设置、缓存带来的延迟和性能权衡。

第三个追问是"数据倾斜怎么办"。你可以说在数据生成器里模拟过热 key 分布，然后讲两阶段聚合的思路，即先在本地进行一次预聚合，再在全局做第二次聚合。

第四个追问是"如果流量翻五倍，链路哪里先扛不住"。这个问题最考验功底，你要从 Kafka 分区数、Flink 并行度、MySQL 连接数、维表缓存命中率几个角度逐一分析，而不是给一个笼统的"加机器"。

5.3 最容易暴露"没做过真实项目"的细节

我面过一些候选人，简历上写着精通 Flink，但追问到细节就露馅了。最典型的问题有三个。

第一，说不清连接器 jar 包版本。真做过项目的人一定被 jar 包折磨过，Flink 版本和 connector 版本强绑定，你能讲出"flink-sql-connector-kafka 的版本必须和 Flink 内核版本保持一致，否则会有不可预期的行为"，面试官就知道你不是只看过文档。

第二，说不清结果表为什么用 upsert-kafka。如果只是背概念，你讲不出迟到数据驱动结果更新的语义；如果你真的在仓库里跑过这个案例，看过同一个窗口输出两条数据，你讲起来是完全不一样的。

第三，没有核对结果的意识。面试官问一句"你怎么保证算出来的数据是对的"，很多候选人就卡住了。你在仓库里实践过的三层核对法，这时候就是最好的答案。

6. 端到端调试中反复出现的五个坑

6.1 连接器 jar 包缺失：作业能提交，但 source 端永远没数据

这是仓库刚整理时我踩的第一个坑，也是最容易让人懵的坑。Flink 镜像本身只包含核心组件，连接 Kafka、MySQL 的 connector 都需要自己额外下载 jar 包放到 lib 目录，否则启动 SQL Client 创建 Kafka 源表时不会直接报错，但作业提交后 source 端一直在等数据，Kafka 里明明有数据就是读不进来。

排查方法很简单：看 Flink Web UI 里 source 算子的 records received 指标。如果一直是 0，先确认是不是 jar 包没放进去，而不是去怀疑 Kafka 集群的问题。

6.2 时区差八个小时：window_start 显示不对，但数据没算错

实时数仓里时区问题几乎是必踩的。我在跑第一个案例时，明明北京时间下午 15 点产生的数据，窗口聚合结果打印出来却是早上 07 点。当时我一度以为是数据生成器的时间格式有问题，排查了一圈才发现是 Flink 默认使用 UTC 时区，TIMESTAMP 转成字符串显示时差了 8 小时。

解决办法是在 SQL Client 里执行：

SET 'table.local-time-zone' = 'Asia/Shanghai';

或者直接改 flink-conf.yaml 里的配置。这个坑给我最大的教训是：要先区分"数据本身的时间"和"展示出来的时间"。watermark 是单调递增的 epoch 时间，它不随时区变化；变了的是 TIMESTAMP 的字符串表示。理解了这个，你就不会被各种时区问题绕晕。

6.3 维表查询不加缓存：MySQL 被打满，作业吞吐骤降

我在调试 Lookup Join 时做过一次对比实验。不加 lookup.cache 配置的时候，每秒 100 条订单数据，每条都要实时查一次 MySQL，作业吞吐从几千掉到几十，MySQL 端 Full table scan 的连接数直接爆掉。加上 cache 配置之后，吞吐恢复，MySQL 的压力也下来了。

这里要提醒一句：缓存不是加的越大越好。lookup.cache.ttl 设得越长，维表更新延迟越高。如果业务对用户信息的时效性要求高，比如新注册用户要立刻看到完整画像，缓存策略就要重新设计。

6.4 Upsert-Kafka 结果表必须声明主键

我第一次用 upsert-kafka 的时候，建表语句没写主键，结果作业提交直接被校验拒绝。Flink 要求 upsert-kafka 的源表和结果表都必须声明主键，因为它要靠主键区分哪两条数据是更新关系，没有主键就无从谈起 upsert 语义。

这个报错信息挺明显的，但很多人会忽略，特别是有 DataStream 开发经验的人，容易下意识认为 Kafka 就是 append-only 的，没必要主键。其实 upsert-kafka 在底层还是 append-only，只是通过 Key 和 Value 的形式把更新逻辑编码进去，主键就是 Key 的字段定义。

6.5 对账对不上：先查总量，再查明细

最后这个坑是我们实际核对结果时遇到的。跑完窗口聚合，拿结果和离线预计算总量对比，怎么都对不上，少了那么几十条。后来发现是数据生成器在持续写入，你很难画一条干净的"从几点到几点"的边界，因为 Kafka 里的数据一直在流动，离线这边统计的区间和实时这边窗口的边界差了半分钟。

我的建议是：核对总量之前，先设定一个严格的测试窗口段。比如停掉生成器，手动往 Kafka 里精确写入 100 条数据，时间戳都落在同 1 分钟窗口内，然后等结果表输出，再对比总数是否等于 100。做这种受限实验，比在持续生产的数据流上核对要高效得多。测试通过后，再放开生成器做压力验证。这套方法后来被我写进了仓库的 docs/数据核对方法.md，成了所有案例的通用验收标准。

我在整理这套案例时还有一个体会：实时数仓的学习曲线陡，但绝大多数卡点不是原理太难，而是环境、版本、细节这些"脏活"消耗了太多精力。这个仓库把脏活都提前处理干净了，剩下的就是专心理解数据链路本身。如果你照着案例跑一遍，把每个"为什么"都弄明白，再去面试实时数仓岗位，心里会踏实很多。