企业官网建设流程全解析

1. 项目概述：为什么要在 Zeppelin 里用 Flink SQL 写 Kafka？

你有没有遇到过这样的场景：业务团队凌晨三点发来消息，说“实时大屏上订单量突然归零了”，运维查了一圈发现是某个 Flink 作业挂了，重启后又得手动补数据；或者数据分析师想快速验证一个新指标逻辑，得先找开发写个 Java/Scala 任务、打包、提交、等日志、再查 Kafka 主题——一整套流程走完，灵感早凉了。而这个标题“Flink SQL on Zeppelin（4）——聚合结果写入Kafka”，说的正是把实时计算的“最后一公里”彻底拉平：让 SQL 写完，结果就稳稳落进 Kafka，中间不卡壳、不跳转、不依赖额外脚本。

这不是炫技，而是真实产线里反复被验证过的提效路径。我参与过的某电商实时风控项目，最初用 Flink DataStream 做用户行为聚合，结果写入 Kafka 的逻辑散落在 3 个不同类里，参数硬编码在producer.properties文件中，每次调整分区策略或序列化方式，都得重新编译发布。后来换成 Flink SQL + Zeppelin，整个链路变成：SQL 编辑框里敲完INSERT INTO kafka_sink SELECT user_id, COUNT(*) FROM events GROUP BY user_id, TUMBLING(INTERVAL '5' MINUTES)，点一下运行，5 秒内就能在 Kafka 消费端看到结构化 JSON 数据流。关键在于，Zeppelin 不只是个“SQL 控制台”，它天然支持多段依赖执行、变量注入、结果可视化回写，还能把整个 notebook 导出为可复用的模板——这意味着，一个 BI 工程师能基于已有 notebook 快速克隆出“GMV 分钟级聚合”版本，只需改两行 SQL 和主题名，不用碰任何配置文件。

这里的核心关键词是Flink SQL、Zeppelin、Kafka Sink，三者组合解决的是“实时计算结果如何低门槛、高可靠、可追溯地进入下游消费生态”这个根本问题。它不替代 Flink 的底层能力，而是把 Flink 最强大的流式 SQL 引擎，装进一个带状态管理、协作编辑、结果反馈的交互式沙盒里。适合谁？首先是数据工程师——你不用再为每个临时分析任务单独起一个 Flink 作业；其次是算法同学——训练特征流需要稳定、低延迟的聚合数据源；还有平台建设者——这是构建统一实时数据服务网关的最小可行单元。接下来我会从设计思路、核心细节、实操步骤到排障经验，一层层拆开这个看似简单的“SQL 写 Kafka”，告诉你哪些地方表面平静，底下暗流汹涌。

2. 整体架构设计与方案选型逻辑

2.1 为什么不是直接用 Flink CLI 或 REST API？

有人会问：Flink 自带 SQL Client，也能连 Kafka，为啥非得绕一圈到 Zeppelin？答案藏在三个维度里：状态管理、协作成本、可观测性。

Flink SQL Client 是单机命令行工具，启动时加载一次 Catalog 配置，但无法保存会话状态。比如你定义了一个 Kafka 表，执行SELECT * FROM kafka_source LIMIT 10查看数据没问题，可一旦退出重进，所有表定义全丢，下次还得CREATE TABLE一遍。更麻烦的是，它不支持变量替换——你想测试不同时间窗口（5分钟 vs 10分钟），就得手动改 SQL 字符串再执行，没法做参数化对比。而 Zeppelin 的%flink.sql解释器天然支持${var}语法，配合 notebook 的段落（paragraph）机制，你能把“建表语句”“查询语句”“写入语句”拆成独立段落，每个段落可单独运行、可设置依赖关系（比如“写入段落”必须等“建表段落”成功后才触发），还能把${window_size}这种变量全局注入，一键切换实验参数。

再看协作层面。Flink CLI 是个人玩具，但产线任务是多人协作的。Zeppelin 的 notebook 支持多人同时编辑、历史版本回溯、段落级权限控制（比如只允许 BI 同学运行查询段落，禁止修改建表段落）。我们曾有个案例：风控团队和推荐团队共用一套用户行为 Kafka 主题，前者要统计异常点击率（需过滤 bot 流量），后者要计算活跃度（需保留全部流量）。通过 Zeppelin，我们为两个团队分别创建 notebook，底层共享同一个 Kafka 表定义，但各自 SQL 中用不同WHERE条件隔离逻辑，互不干扰，审计日志里还能清晰看到谁在什么时间执行了哪条 SQL。

最后是可观测性。Flink CLI 执行INSERT INTO后，你只能等 Kafka 消费端反馈，或者去 Flink Web UI 看作业状态，但看不到“这条 SQL 实际生成了多少条记录”“每条记录的字段值是否符合预期”。Zeppelin 的%flink.sql解释器会在执行后自动捕获前 100 条输出结果（如果启用了sink.print），并以表格形式渲染在 notebook 里。更重要的是，它能把 Flink 作业的 JobID、Checkpoint 路径、并行度等元信息直接写进段落结果区，点一下就能跳转到 Flink Web UI 查看详情。这种“所见即所得”的调试体验，是 CLI 根本无法提供的。

2.2 为什么 Kafka 是首选 Sink 目标？

可能你会疑惑：既然都用 Zeppelin 了，为啥不直接写入 MySQL 或 Elasticsearch？因为 Kafka 在实时链路中承担着不可替代的“缓冲区+协议转换器”角色。

首先看可靠性。MySQL 写入是强事务模型，Flink 的 Checkpoint 机制与 MySQL 的两阶段提交（2PC）耦合复杂，一旦下游 DB 出现主从延迟或连接抖动，Flink 作业会卡在 Checkpoint，导致整个流中断。而 Kafka 是异步批量写入，Flink 的 KafkaSink 内部维护一个内存缓冲区（默认 1MB），当缓冲区满或达到linger.ms（默认 0，即立即发送）时才批量刷盘。即使 Kafka 集群短暂不可用，Flink 会自动重试（默认 3 次），失败后抛出异常触发作业 failover，但不会阻塞上游数据处理——这意味着你的聚合逻辑依然在持续计算，只是结果暂时积压在 Flink TaskManager 的内存里，等 Kafka 恢复后自动续传。这种“削峰填谷”能力，是直连数据库做不到的。

其次看解耦性。一个典型的实时数仓架构里，Flink 聚合结果往往要被多个下游消费：BI 工具做实时看板、AI 平台取特征流、告警系统监听阈值突破。如果 Flink 直接写 MySQL，每个下游都要轮询查询，既增加 DB 压力，又无法保证数据新鲜度（轮询间隔决定延迟）。而 Kafka 天然支持“一对多”广播，Flink 只需写一次，所有消费者按需订阅，互不影响。我们曾做过压测：同一份用户停留时长聚合结果，写入 Kafka 后，3 个下游消费者（Flink CEP 告警、Spark Streaming 特征计算、Grafana 监控）同时消费，Kafka 集群 CPU 使用率仅上升 8%，而改用 MySQL 时，DB 连接数瞬间打满，查询延迟飙升至 2s+。

最后是序列化灵活性。Kafka 只管传输字节流，具体怎么序列化由生产者决定。Flink SQL 默认用JsonFormat，但你可以轻松切换成AvroFormat（需提前注册 Schema Registry 地址）或DebeziumJsonFormat（兼容 CDC 数据格式）。比如某次需求变更，下游要求字段类型严格校验，我们只需在CREATE TABLE语句中把FORMAT 'json'改成FORMAT 'avro'，并指定schema.registry.url参数，无需改动任何业务逻辑代码。这种“协议无关”的设计，让数据管道具备极强的演进韧性。

2.3 Zeppelin 解释器配置的关键权衡

Zeppelin 本身不执行 Flink 任务，它只是个调度和展示层，真正的计算由 Flink Cluster 完成。因此解释器配置决定了整个链路的健壮性。核心配置项有三个：Flink 集群模式、Checkpoint 语义、State Backend。

第一，集群模式选YARN_PER_JOB还是YARN_APPLICATION？前者为每个 notebook 段落启动独立 Flink 作业（JVM 进程），后者复用同一个 ApplicationMaster。我们强烈推荐YARN_PER_JOB，虽然启动稍慢（约 8-12 秒），但它实现了严格的资源隔离。想象一下：A 同学跑一个 100 并行度的聚合任务占满集群，B 同学同时执行一个简单SELECT COUNT(*)，如果共用 Application，后者会被前者饿死。而PER_JOB模式下，每个作业有独立的 TaskManager，资源配额（CPU、内存）可精确控制，YARN 调度器能公平分配。配置时需在 Zeppelin 的flink.interpreter.group设置yarn.per-job，并在flink.yarn.application.name中指定唯一作业名（建议加 notebook ID 前缀，如zeppelin-notebook-abc123-gmv-aggr）。

第二，Checkpoint 语义必须设为EXACTLY_ONCE。这是 Kafka Sink 可靠性的基石。Flink 默认是AT_LEAST_ONCE，意味着同一条记录可能被重复写入 Kafka（比如网络抖动导致 Flink 认为发送失败，实际 Kafka 已收到）。开启EXACTLY_ONCE后，Flink 会在 Kafka 中为每个分区维护一个__consumer_offsets主题的 offset 提交点，并在 Checkpoint 成功时原子性地提交 offset 和 Kafka producer 的事务状态。配置方法是在 Zeppelin 解释器设置里添加execution.checkpointing.mode=EXACTLY_ONCE和execution.checkpointing.interval=60000（60秒）。注意：这要求 Kafka 集群版本 ≥ 0.11.0，且transactional.id必须全局唯一（Zeppelin 会自动生成，格式为flink-sql-zeppelin-${noteId}-${paragraphId}）。

第三，State Backend 推荐RocksDB而非Memory或FsStateBackend。聚合操作（如COUNT、SUM）会产生大量中间状态，Memory会把所有状态存 JVM Heap，极易 OOM；FsStateBackend虽然落盘，但频繁读写本地磁盘影响性能。RocksDB是嵌入式 KV 存储，状态数据存本地 SSD，Checkpoint 时只把增量快照同步到 HDFS/S3。配置项为state.backend=rocksdb和state.checkpoints.dir=hdfs://namenode:8020/flink/checkpoints。我们实测过：处理 10 亿/天事件流时，RocksDB的平均 GC 时间比Memory低 73%，Checkpoint 完成率从 89% 提升至 99.98%。

提示：Zeppelin 解释器配置中有一项flink.sql.sink.parallelism易被忽略。它控制 Kafka Sink 的并行度，默认为 1。如果你的 Kafka 主题有 12 个分区，却只用 1 个 Sink Task 写入，会导致严重倾斜——所有数据挤在 1 个分区，其他 11 个分区空转。正确做法是设为 Kafka 分区数（如flink.sql.sink.parallelism=12），Flink 会自动按keyBy的 key 哈希分发到对应分区，实现负载均衡。

3. 核心细节解析与实操要点

3.1 Kafka 表定义的隐藏陷阱

Flink SQL 写 Kafka 的第一步是CREATE TABLE，但很多人卡在这一步就失败，原因全在表定义的细节里。我们逐字段拆解一个生产环境可用的模板：

CREATE TABLE kafka_sink ( user_id STRING, order_cnt BIGINT, window_start TIMESTAMP(3), window_end TIMESTAMP(3), proc_time AS PROCTIME() ) WITH ( 'connector' = 'kafka', 'topic' = 'dwd_user_order_agg_5min', 'properties.bootstrap.servers' = 'kafka-broker1:9092,kafka-broker2:9092', 'properties.group.id' = 'flink-sql-zeppelin-sink-group', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601', 'sink.partitioner' = 'round-robin', 'sink.semantic' = 'exactly-once', 'sink.parallelism' = '12' );

第一眼看着没问题，但实际部署时，80% 的失败源于这五个字段：

properties.group.id：这个参数名极具迷惑性！它不是Kafka Consumer 的 group.id，而是 Flink KafkaSink 内部用于事务管理的标识。Flink 要求每个EXACTLY_ONCE语义的 Sink 必须有唯一transactional.id，而这个group.id会被 Flink 自动转换为transactional.id（规则是group.id + "-tx-" + randomUUID）。如果多个 notebook 共用同一个group.id，会导致事务冲突——A 作业提交事务时，B 作业的 producer 被强制关闭。解决方案是：在 Zeppelin 的 notebook 设置里启用“动态变量”，把group.id设为${noteId}-${paragraphId}，确保全局唯一。

json.timestamp-format.standard：Flink 1.14+ 默认用SQL格式解析时间戳（如'2023-01-01 12:00:00'），但 Kafka 里的时间字段往往是 ISO 标准（'2023-01-01T12:00:00.123Z'）。如果不显式声明，Flink 会报Cannot parse timestamp错误。实测发现，ISO-8601能兼容带毫秒和时区的字符串，而SQL格式只认空格分隔、无时区的格式。

sink.partitioner：这是控制数据分发到 Kafka 分区的核心开关。默认值是default，它会把所有数据发到分区 0（如果没指定 key），造成单点瓶颈。round-robin是最安全的选择，它无视 key，轮询分发到所有分区，适合不需要严格顺序的场景（如聚合结果）。但如果下游要求“同一 user_id 的数据必须在同一个分区”，就得用key-hash，此时必须确保user_id字段在CREATE TABLE中声明为PRIMARY KEY，否则 Flink 会报Key is not defined错误。

sink.semantic：必须显式设为'exactly-once'。虽然 Flink 默认开启 EXACTLY_ONCE，但 Kafka connector 的sink.semantic参数是独立控制的。如果这里写成'at-least-once'，即使全局 Checkpoint 设为 EXACTLY_ONCE，Kafka 写入仍会重复。我们曾因漏配此项，导致风控告警重复触发 37 次，排查了两天才发现是这个参数惹的祸。

sink.parallelism：前面提过，它必须等于 Kafka 主题的分区数。但还有一个隐藏规则：如果 Flink 作业的并行度（parallelism.default）小于sink.parallelism，Flink 会自动扩容 Sink Task；反之，如果大于，则会合并。比如主题有 12 分区，但sink.parallelism=6，Flink 会用 6 个 Task，每个 Task 负责 2 个分区，导致吞吐下降。所以务必保持两者一致。

注意：PROCTIME()是处理时间函数，它返回 Flink TaskManager 的系统时间。不要用ROWTIME()（事件时间），因为 Kafka Sink 不支持事件时间语义——它只负责写入，不参与时间窗口计算。如果你的聚合逻辑依赖事件时间（如TUMBLING(event_time, INTERVAL '5' MINUTES)），window_start和window_end字段必须在上游SELECT中用TUMBLING函数生成，不能指望 Sink 自动填充。

3.2 聚合 SQL 的性能调优技巧

写入 Kafka 只是终点，但聚合过程才是性能瓶颈所在。一个常见的反模式是：SELECT user_id, COUNT(*) FROM events GROUP BY user_id，看似简单，实则暗藏杀机。

问题一：KeyBy 倾斜。user_id如果存在超级节点（比如user_id='system'占 30% 流量），所有匹配该 key 的数据都会路由到同一个 TaskManager，导致该节点 CPU 100%，其他节点闲置。解决方案是加盐（salting）：在GROUP BY前对 key 做哈希分桶。Flink SQL 不支持原生加盐语法，但可以用CASE WHEN模拟：

SELECT CASE WHEN MOD(HASH_CODE(user_id), 100) < 10 THEN CONCAT('salt1_', user_id) WHEN MOD(HASH_CODE(user_id), 100) < 20 THEN CONCAT('salt2_', user_id) ELSE user_id END AS salted_user_id, COUNT(*) as cnt FROM events GROUP BY CASE WHEN MOD(HASH_CODE(user_id), 100) < 10 THEN CONCAT('salt1_', user_id) WHEN MOD(HASH_CODE(user_id), 100) < 20 THEN CONCAT('salt2_', user_id) ELSE user_id END

这样就把热点 user_id 拆到多个 salt 前缀下，分散到不同分区。下游消费时，用SUBSTRING(salted_user_id, 6)去掉前缀即可还原原始 ID。

问题二：状态爆炸。COUNT(*)看似轻量，但如果GROUP BY字段组合过多（如GROUP BY user_id, product_id, category_id），状态会随维度组合数线性增长。Flink 默认状态 TTL 是无限期，意味着昨天的数据状态永远不清理。必须显式设置 TTL：在CREATE TABLE的WITH子句中添加'state.ttl' = '86400000'（24 小时毫秒值）。更激进的做法是用OVER WINDOW替代GROUP BY：

SELECT user_id, COUNT(*) OVER (PARTITION BY user_id ORDER BY proc_time ROWS BETWEEN 1000 PRECEDING AND CURRENT ROW) as recent_cnt FROM events

OVER WINDOW基于滑动窗口计算，状态只保留最近 1000 条记录，内存占用恒定，适合“最近 N 条行为统计”类场景。

问题三：JSON 序列化开销。默认jsonformat 会把所有字段转成字符串再拼 JSON，对BIGINT、DOUBLE等数值类型是浪费。可以改用avro格式，它用二进制编码，体积小 40%，序列化耗时降 60%。但需提前在 Confluent Schema Registry 注册 schema：

{ "type": "record", "name": "UserOrderAgg", "fields": [ {"name": "user_id", "type": "string"}, {"name": "order_cnt", "type": "long"}, {"name": "window_start", "type": "long"}, {"name": "window_end", "type": "long"} ] }

然后在CREATE TABLE中指定：

'format' = 'avro', 'avro.schema-registry.url' = 'http://schema-registry:8081'

3.3 Zeppelin 段落协同与变量注入实战

Zeppelin 的真正威力不在单条 SQL，而在多段落协同。我们以“实时 GMV 分钟级监控”为例，展示如何用 4 个段落构建闭环：

段落 1：动态参数定义（名称：Set Parameters）

%flink.sql -- 设置可变参数，支持 notebook 级别复用 SET 'pipeline.name' = 'gmv-monitor-1min'; SET 'kafka.topic' = 'dwd_gmv_1min'; SET 'window.size' = '1'; -- 单位：分钟

段落 2：Kafka 表创建（名称：Create Kafka Sink，依赖：Set Parameters）

%flink.sql CREATE TABLE kafka_sink ( gmv DECIMAL(18,2), window_start TIMESTAMP(3), window_end TIMESTAMP(3) ) WITH ( 'connector' = 'kafka', 'topic' = '${kafka.topic}', 'properties.bootstrap.servers' = 'kafka-broker1:9092', 'format' = 'json', 'sink.partitioner' = 'round-robin', 'sink.semantic' = 'exactly-once' );

段落 3：聚合逻辑（名称：Run Aggregation，依赖：Create Kafka Sink）

%flink.sql INSERT INTO kafka_sink SELECT SUM(price) as gmv, TUMBLING_START(proc_time, INTERVAL '${window.size}' MINUTES) as window_start, TUMBLING_END(proc_time, INTERVAL '${window.size}' MINUTES) as window_end FROM orders GROUP BY TUMBLING(proc_time, INTERVAL '${window.size}' MINUTES);

段落 4：结果验证（名称：Verify Output，依赖：Run Aggregation）

%flink.sql -- 自动消费刚写入的 Kafka 数据，验证前 5 条 SELECT * FROM kafka_sink LIMIT 5;

这种设计带来三大好处：

1. 可复用性：复制整个 notebook，只需改Set Parameters段落里的window.size，就能快速生成 5 分钟、10 分钟版本；
2. 可追溯性：每个段落执行后，Zeppelin 会记录JobID和Duration，点击JobID能跳转到 Flink Web UI 查看详细 Metrics；
3. 防误操作：Verify Output段落设置了依赖，只有Run Aggregation成功后才执行，避免查空数据。

实操心得：Zeppelin 的变量注入${var}不支持表达式，比如${window.size * 60}会报错。如果需要单位转换，得在SET语句里预先计算好，如SET 'window.ms' = '60000'。另外，LIMIT子句在INSERT INTO中无效，必须用独立的SELECT段落验证。

4. 实操过程与核心环节实现

4.1 从零搭建 Zeppelin+Flink+Kafka 环境

虽然标题聚焦“写入 Kafka”，但环境搭建是绕不开的第一步。我们跳过官网文档的泛泛而谈，直击生产环境必须处理的 5 个关键动作。

第一步：确认 Flink 版本兼容性
Zeppelin 0.10+ 官方支持 Flink 1.13~1.17，但实际使用中，Flink 1.15.4 是当前最稳的黄金版本。原因有三：

• 它修复了 1.14 中 Kafka connector 的EXACTLY_ONCE事务泄漏 bug（FLINK-25892）；
• 它对TUMBLING窗口的PROCTIME支持最完善，1.16+ 开始引入EVENT TIME优化，反而在某些边缘 case 下出现水位线延迟；
• 社区插件生态最丰富，比如flink-sql-connector-kafka_2.12-1.15.4.jar可直接下载，无需编译。

下载地址：https://archive.apache.org/dist/flink/flink-1.15.4/flink-1.15.4-bin-scala_2.12.tgz
解压后，将lib/flink-sql-connector-kafka_2.12-1.15.4.jar复制到 Zeppelin 的interpreter/flink/目录下（Zeppelin 0.10.1 默认路径）。

第二步：Zeppelin 解释器配置（关键参数清单）
进入 Zeppelin Web UI →Interpreter→ 找到flink解释器 →Edit，填写以下必填项：

注意：flink.execution.type=streaming这个配置容易被忽略。如果设成batch，Flink SQL 会尝试用批模式执行流式 SQL，导致TUMBLING窗口无法识别，报Unsupported window function错误。

第三步：Kafka 主题预创建与权限校验
不要依赖 Flink 自动创建主题！生产环境必须手动创建并设置合理参数：

# 创建主题，12 分区，3 副本 kafka-topics.sh --create \ --bootstrap-server kafka-broker1:9092 \ --topic dwd_user_order_agg_5min \ --partitions 12 \ --replication-factor 3 \ --config retention.ms=604800000 \ # 7天保留 --config segment.bytes=1073741824 # 1GB 分段

关键参数说明：

• retention.ms=604800000：7 天保留，避免聚合结果被过早清理；
• segment.bytes=1073741824：1GB 分段，减少小文件数量，提升消费吞吐；
• 必须禁用auto.create.topics.enable=true，否则 Flink 第一次写入时会创建主题，但默认只有 1 分区，导致后续扩容困难。

第四步：Zeppelin Notebook 初始化
新建 notebook 后，首段必须执行初始化 SQL，否则后续建表会失败：

%flink.sql -- 启用 Flink SQL 的 Hive Catalog（如果要用 Hive 维表关联） -- CREATE CATALOG hive_catalog WITH ('type' = 'hive', 'hive-conf-dir' = '/etc/hive/conf'); -- USE CATALOG hive_catalog; -- 设置默认并行度（避免单点瓶颈） SET 'parallelism.default' = '12'; -- 启用 checkpoint（Zeppelin 不自动继承全局配置） SET 'execution.checkpointing.enabled' = 'true'; SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE'; SET 'execution.checkpointing.interval' = '60000';

第五步：验证 Kafka 连通性
在 Zeppelin 中执行一条最简INSERT，验证基础链路：

%flink.sql -- 创建一个测试表 CREATE TABLE test_kafka ( msg STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'test_topic', 'properties.bootstrap.servers' = 'kafka-broker1:9092', 'format' = 'json' ); -- 写入测试数据 INSERT INTO test_kafka SELECT 'hello from zeppelin' as msg;

然后用 Kafka 命令行消费：

kafka-console-consumer.sh --bootstrap-server kafka-broker1:9092 \ --topic test_topic --from-beginning --max-messages 1

如果看到{"msg":"hello from zeppelin"}，说明环境已通。

4.2 完整聚合写入流程演示

现在我们把所有环节串起来，完成一个真实的“用户订单量 5 分钟滚动聚合”任务。

Step 1：准备源数据 Kafka 主题
假设已有ods_user_orders主题，每条消息是 JSON：

{ "order_id": "ORD-123456", "user_id": "U-7890", "price": 299.99, "event_time": "2023-10-01T12:00:05.123Z" }

Step 2：在 Zeppelin 中创建源表

%flink.sql CREATE TABLE ods_orders ( order_id STRING, user_id STRING, price DECIMAL(10,2), event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECONDS ) WITH ( 'connector' = 'kafka', 'topic' = 'ods_user_orders', 'properties.bootstrap.servers' = 'kafka-broker1:9092', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' );

注意WATERMARK定义：event_time - INTERVAL '5' SECONDS表示允许 5 秒乱序，这是事件时间窗口的基石。

Step 3：定义目标 Kafka 表（带完整容错配置）

%flink.sql CREATE TABLE dwd_user_order_agg_5min ( user_id STRING, order_cnt BIGINT, gmv DECIMAL(18,2), window_start TIMESTAMP(3), window_end TIMESTAMP(3), proc_time AS PROCTIME() ) WITH ( 'connector' = 'kafka', 'topic' = 'dwd_user_order_agg_5min', 'properties.bootstrap.servers' = 'kafka-broker1:9092,kafka-broker2:9092', 'properties.security.protocol' = 'PLAINTEXT', -- 如需 SASL，改为 SASL_PLAINTEXT 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601', 'sink.partitioner' = 'key-hash', -- 按 user_id 分区 'sink.semantic' = 'exactly-once', 'sink.parallelism' = '12', 'sink.buffer-flush.max-bytes' = '1048576', -- 1MB 缓冲区 'sink.buffer-flush.interval' = '1000' -- 1秒刷新一次 );

Step 4：执行聚合写入（核心 SQL）

%flink.sql INSERT INTO dwd_user_order_agg_5min SELECT user_id, COUNT(*) as order_cnt, SUM(price) as gmv, TUMBLING_START(event_time, INTERVAL '5' MINUTES) as window_start, TUMBLING_END(event_time, INTERVAL '5' MINUTES) as window_end FROM ods_orders GROUP BY user_id, TUMBLING(event_time, INTERVAL '5' MINUTES);

Step 5：实时验证结果
新开一个段落，用 Flink SQL 直接消费目标主题（需 Kafka connector 支持 consumer）：

%flink.sql -- 创建消费表（仅用于验证，不推荐长期运行） CREATE TEMPORARY TABLE verify_sink ( user_id STRING, order_cnt BIGINT, gmv DECIMAL(18,2), window_start TIMESTAMP(3), window_end TIMESTAMP(3) ) WITH ( 'connector' = 'kafka', 'topic' = 'dwd_user_order_agg_5min', 'properties.bootstrap.servers' = 'kafka-broker1:9092', 'format' = 'json', 'scan.startup.mode' = 'latest-offset' -- 从最新 offset 开始 ); -- 查询最近 10 条 SELECT * FROM verify_sink LIMIT 10;

执行后，Zeppelin 会以表格形式渲染结果，你能看到类似：

Step 6：监控与告警接入
Flink 作业启动后，Zeppelin 会显示JobID: 123e4567-e89b-12d3-a456-426614174000。复制此 ID，在 Flink Web UI（http://flink-jobmanager:8081）中搜索，进入作业详情页。重点关注：

• Checkpoint Status：绿色表示成功，红色表示失败；
• Metrics → numRecordsOutPerSecond：应稳定在预期吞吐（如 10000 rec/s）；
• Task Managers → Memory Usage：RocksDB 状态内存不应超过总内存的 70%。

如果发现numRecordsOutPerSecond骤降，大概率是 Kafka 写入阻塞，此时去 Kafka Broker 日