Flink SQL + Kafka 实时统计实战：从建表、窗口聚合到踩坑排查

做实时统计这件事，我从 DataStream API 一路写到现在，说实话早期被各种乱序、延迟、状态恢复的问题折磨得不轻。后来 Flink SQL 慢慢成熟，尤其是和 Kafka 搭在一起用，一条从消息队列到实时指标的链路变得非常干净：Java 工程里起一个 Flink SQL 作业，用 DDL 定义好 Kafka 来源表和结果去向表，剩下的统计逻辑全交给 SQL。今天这篇文章就把我实际搭建这套程序的过程完整拆开，从版本选型、建表 DDL、窗口聚合、结果 Sink，到踩过的坑和排查思路，全部按干货讲。适合正在学 Flink SQL 的 Java 后端，也适合已经在用 Kafka 做数据管道、想快速产出实时统计结果的团队参考。

为什么强调“Java 版”，是因为很多同学搜到教程可能是 Python 脚本或者 Scala Demo，真正落到 Java 工程里要配 connector、要处理依赖冲突、要设置状态后端，细节全在工程代码里。我这里不会只贴一个 SQL 片段，而是把整个作业从零到落地怎么组织讲清楚，你跟着搭一遍，基本就能迁移到自己业务上。

1. 为什么我选择“Flink SQL + Kafka”这套组合

1.1 实时统计最朴素的链路

先还原一个最常见的场景：订单系统或者埋点日志系统，每时每刻都在往 Kafka 里写消息。业务方想实时知道的往往就是这几类问题：现在每分钟有多少笔订单、累计成交额是多少、哪个品类卖得最好、今天有多少活跃用户。

这些指标如果靠传统 Java 后端做定时任务去数据库里扫，数据量一大根本扛不住，而且统计口径分散在业务代码里，改一个指标要重新发版。所以需要一个独立于业务系统之外的实时计算层，Kafka 做数据通道，Flink SQL 做计算引擎。

这条链路有固定套路：Kafka topic 进，Flink 作业算，结果写回 Kafka 或者 MySQL/ES。我长期用这套方案的原因很朴素。第一，Kafka 作为消息缓冲层可以削峰，业务端只管发送，不关心下游怎么算；第二，Flink SQL 的语义足够直观，一个窗口聚合就是一条 CREATE TABLE 加一条 INSERT INTO SELECT，不需要像手写 DataStream 那样处理底层各类细节；第三，状态和容错是引擎自带的，作业重启后能从上一次 checkpoint 恢复，不用自己维护中间结果。

1.2 Flink SQL 相比 DataStream API 的核心优势

早期很多团队排斥 Flink SQL，觉得灵活性不够，只能写 DataStream API 才叫“真实时计算”。实际上，实时统计类的需求有非常明显的模式化特征：窗口聚合、多维计数、去重、TopN、累计值，这些用 SQL 表达几乎是标准答案。

但也不是完全替代。当业务需要多流按业务 ID 匹配、复杂状态机、或者要实时调用外部服务做特征加工时，还是要回到 DataStream API。以我自己的经验来看，SQL 覆盖了简单统计和报表类 80% 的场景，剩下 20% 需求写成 UDF 嵌入 SQL 也不难。我的建议是：先 SQL，SQL 表达不了再加 UDF，最后才考虑 DataStream，不要一上来就用低级 API 堆逻辑。

2. 环境准备与工程项目搭建

2.1 版本选型

Flink 生态的版本兼容是一件很头疼的事，版本选不对，后面连 ClassNotFound 都算轻的，还可能遇到 connector 和引擎之间方法签名对不上的情况。我这里给出一套我实际跑通的组合：

• JDK：8 或 11 都行，我建议 JDK 11，长期维护，GC 表现也好一些；
• Flink：1.17.2，这是目前线上用得比较稳的版本，新特性不少，社区踩坑资料也全；
• Kafka： 3.4.0，2.8 以上的版本基本都兼容；
• Flink SQL Kafka Connector：flink-sql-connector-kafka 1.17.2，注意是带 sql 前缀的 fat jar；
• JDBC Connector：flink-connector-jdbc 3.1.2-1.17，用于写 MySQL。

Java 环境变量配置属于最基础的前置工作，这里默认你已经做好了。如果你还没配好 JAVA_HOME，先去把 JDK 装好，否则起 Flink 作业第一步就会卡在环境上。

2.2 Maven 依赖怎么写

工程用 Maven 管理依赖，pom.xml 里核心依赖如下：

<properties> <flink.version>1.17.2</flink.version> </properties> <dependencies> <!-- Flink Table API --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-loader</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <!-- Flink SQL Kafka Connector，需要打包进作业 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-sql-connector-kafka</artifactId> <version>1.17.2</version> </dependency> <!-- JDBC Sink --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-jdbc</artifactId> <version>3.1.2-1.17</version> </dependency> <!-- MySQL 驱动 --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.33</version> </dependency> <!-- RocksDB 状态后端，建议加上 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-statebackend-rocksdb</artifactId> <version>${flink.version}</version> </dependency> </dependencies>

注意几个细节。

flink-table-api-java-bridge 和 flink-table-planner-loader 的 scope 是 provided，因为 Flink 集群本身已经有这些类，打进去反而容易和集群冲突。flink-sql-connector-kafka 必须打进作业 jar，因为它不是 Flink 发行版自带的，需要单独准备。flink-connector-jdbc 同理，也需要打进去。

打包用 maven-shade-plugin，别用 assembly，shade 会正确重定位部分依赖类，减少冲突。如果你的作业是在 Flink SQL Client 里跑，也可以不用 Java 工程，直接把 connector jar 放到 Flink 的 lib 目录，然后写纯 SQL 脚本。但 Java 工程的好处是便于接入公司内部的配置中心、统一日志、监控上报，以及写自定义 UDF。

2.3 连接 Kafka 集群的前置检查

很多同学一上来直接跑 Flink 作业，结果报错，然后对着日志一头雾水。我的习惯是先用命令行确认 Kafka 本身是通的，再启动 Flink 作业，这样可以快速隔离问题。

先建 topic 并手动发一条消息验证：

kafka-topics.sh --bootstrap-server kafka-1:9092 --create --topic order_topic --partitions 6 --replication-factor 1 kafka-console-producer.sh --bootstrap-server kafka-1:9092 --topic order_topic > {"order_id":"o_10001","user_id":"u_8888","product_id":"p_233","category":"数码","amount":1299.00,"ts":"2024-06-01 10:15:30"}

然后另开一个终端消费确认消息能读到：

kafka-console-consumer.sh --bootstrap-server kafka-1:9092 --topic order_topic --from-beginning

如果生产和消费都正常，说明 Kafka 集群对外服务没问题，问题只可能出在 Flink 作业的配置上。如果这里就报错，先解决 Kafka 侧的网络、权限、topic 分区等问题，再往下走。

“error while fetching metadata”这类报错，最常见原因有三个：bootstrap.servers 配置不对、advertised.listeners 和客户端不在同一个网段、安全组拦截了 9092 端口。排查顺序是：先 telnet 看端口通不通，再看 advertised.listeners 配置的地址从客户端能否访问，最后确认 topic 是否存在。尤其在 Docker 部署 Kafka 的场景，容器内部监听地址和宿主机外部访问地址经常不一致，这是超高频踩坑点。

3. 用 Flink SQL 把 Kafka 数据读进来

3.1 先定数据模型与消息格式

Flink SQL 建表前，一定要先想清楚 Kafka 里的消息长什么样。以订单为例，我常用的 JSON 消息格式是这样的：

{ "order_id": "o_10001", "user_id": "u_8888", "product_id": "p_233", "category": "数码", "amount": 1299.00, "ts": "2024-06-01 10:15:30" }

字段类型上注意几点。amount 建议用 DECIMAL(10, 2)，不要用 DOUBLE，避免金额精度问题。ts 时间字段映射成 TIMESTAMP(3)，毫秒精度。user_id、order_id 这类 ID 字段用 STRING，别用 BIGINT，因为很多 ID 在前端会被处理成字符串，强行转 BIGINT 遇到脏数据会直接导致作业失败。

Kafka 消息格式用 JSON 还是 CSV，取决于上游系统。JSON 可读性好、字段扩展方便，缺点是消息体积比 CSV 大，吞吐会受一些影响。如果对性能要求极高，可以考虑 Avro 加 Schema Registry，但维护成本会上升，小团队没必要一上来就上 Avro。

3.2 创建 Kafka Source 表

在 Flink SQL 里读 Kafka，本质就是定义一张带 connector 属性的表。DDL 如下：

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

这个 DDL 里最值得展开的是 WATERMARK 那一行。实时数据从业务系统发到 Kafka，再被 Flink 消费，过程中不可避免会有网络抖动、业务系统处理延迟，导致消息到达 Flink 的顺序和实际发生时间不一致。如果不处理乱序，直接按事件的 ts 去开窗口做聚合，结果会偏差很大。水印就是用来告诉 Flink：“我最多容忍 5 秒的乱序，超过这个迟到范围的数据，窗口就不等了。”

WATERMARK FOR ts AS ts - INTERVAL '5' SECOND 的具体含义是：当前水印时间 = 已收到消息中最大的 ts 减去 5 秒。窗口触发条件是水印时间 >= 窗口结束时间。所以一个 10:15:00 到 10:16:00 的窗口，要等到水印推进到 10:16:00 才会触发计算，也就是要收到一个 ts 至少为 10:16:05 的消息。

如果你不需要精确的事件时间统计，只想看个大概，可以完全不用事件时间，建表时不声明 ts 和 WATERMARK，窗口直接用处理时间 PROCETIME()。但做实时统计，尤其是涉及金额、订单量这类要和业务对账的指标，我强烈建议用事件时间加水印，结果才可信。

3.3 消费位点与并行度注意点

scan.startup.mode 决定 Flink 作业启动时从 Kafka 的哪个位置开始消费，常用取值有这几个：

• earliest-offset：从头开始消费，适合首次上线想回补历史数据；
• latest-offset：只消费作业启动后新到的数据，适合不关心历史、只要增量；
• timestamp：从指定时间戳之后开始消费，适合想从某个业务时间点补数据；
• specific-offsets：从指定分区的指定 offset 开始消费，一般用得少。

线上首次上线一个统计作业，如果 topic 里已经有堆积数据，我又想快速看到结果，一般用 timestamp，指定到当前时间往前推 5 分钟，既不会重复消费太多历史，也不会漏掉启动瞬间产生的消息。如果数据量小、可以接受重新计算历史，用 earliest-offset 最省事。

并行度方面，Flink SQL 作业的 source 并行度理论上可以大于 Kafka 分区数，但没意义，多的并行度只会空闲。合理的做法是并行度取分区数的整数倍，比如 topic 6 个分区，作业并行度设 6 或者 12，调度最均衡。如果你用 Flink SQL Client 直接跑，可以用 SET 'parallelism.default' = 6 来控制。

4. 实时统计逻辑怎么用 SQL 表达

4.1 最常用的几个实时指标模板

实时统计的需求翻来覆去就那么几类，我把常用的指标模板整理出来，对应到 SQL 上就是固定套路：

• 每分钟订单数和 GMV：用 TUMBLE 滚动窗口，按分钟切分，窗口结束触发输出；
• 近 1 小时每 10 分钟更新的滚动统计：用 HOP 滑动窗口，窗口长度 1 小时，滑动步长 10 分钟；
• 全量累计 PV/UV：用 COUNT 和 COUNT(DISTINCT)，但要小心状态膨胀；
• 品类销量 TopN：窗口内先聚合，再用 ROW_NUMBER 分组排序；
• 会话级统计：用 SESSION 会话窗口，按用户空闲时间切分会话。

刚开始做实时统计，建议先从 TUMBLE 滚动窗口入手，因为它最简单直观，窗口之间互不重叠，理解成本最低。跑通了再加 HOP 和 SESSION，理解窗口的触发机制后，其他窗口只是切分规则不同。

4.2 事件时间窗口聚合完整示例

下面这个例子是典型的“每分钟统计每个品类的订单数和 GMV”，并且直接把结果写入 MySQL。先建结果表：

CREATE TABLE order_stats_sink ( window_start TIMESTAMP(3), window_end TIMESTAMP(3), category STRING, order_cnt BIGINT, gmv DECIMAL(14, 2), PRIMARY KEY (window_start, window_end, category) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/stats', 'table-name' = 'order_stats', 'username' = 'root', 'password' = '123456' );

然后提交计算逻辑：

INSERT INTO order_stats_sink SELECT TUMBLE_START(ts, INTERVAL '1' MINUTE) AS window_start, TUMBLE_END(ts, INTERVAL '1' MINUTE) AS window_end, category, COUNT(*) AS order_cnt, SUM(amount) AS gmv FROM order_source GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), category;

这里有一个非常关键的细节：GROUP BY 后面跟着窗口函数和 category，说明每个品类在每个分钟窗口内都会产出一条结果。如果 category 的取值很多，比如几百个品类，一分钟就会产生几百条结果，下游存储压力不小，设计结果表时要有这个预期。

TUMBLE_START 和 TUMBLE_END 是窗口开始和结束时间，必须显式查询出来，否则下游看到的只有聚合值，不知道对应哪个时间窗口。用 PRIMARY KEY NOT ENFORCED 声明主键，是为了让 JDBC sink 在写入时能按主键做更新，而不是每条都追加。这个声明不会在 Flink 侧强制校验，只影响 sink 的写入语义。

4.3 窗口 TopN 和累计去重

订单量和 GMV 只是基础指标，业务方很快会追问“哪个品类卖得最好”。这就是 TopN 需求。Flink SQL 的窗口 TopN 写法如下：

SELECT window_start, window_end, category, gmv FROM ( SELECT window_start, window_end, category, gmv, ROW_NUMBER() OVER ( PARTITION BY window_start, window_end ORDER BY gmv DESC ) AS rn FROM ( SELECT TUMBLE_START(ts, INTERVAL '5' MINUTE) AS window_start, TUMBLE_END(ts, INTERVAL '5' MINUTE) AS window_end, category, SUM(amount) AS gmv FROM order_source GROUP BY TUMBLE(ts, INTERVAL '5' MINUTE), category ) ) WHERE rn <= 10;

注意这里用的是窗口 TopN，不是普通 TopN。普通 TopN 在无界流上会一直维护全局排名，数据量一大状态就爆炸。窗口 TopN 按 window_start 和 window_end 分区，每个窗口只保留前 10 名，窗口结束状态就可以清理，逻辑和性能都可控。

再讲 UV 类指标。最直接的写法是 COUNT(DISTINCT user_id)，但 Flink SQL 默认的 COUNT(DISTINCT) 是精确去重，状态里会保存每一个出现过的 user_id，用户量一大，状态后端会撑不住。如果 UV 量级在百万以下，精确去重可以接受；如果到千万级以上，建议用近似去重方案，比如自定义 HyperLogLog 的 UDAF，或者把去重窗口缩短，比如按小时而不是按天去重。实时报表的 UV 本来就允许一定误差，1% 以内的偏差业务方几乎感知不到。

5. 结果写到哪里：Sink 设计与落地

5.1 写回 Kafka 做下游复用

实时统计结果最常见的去向之一，是再写回 Kafka。这样下游的 SpringBoot 服务、ES、数据仓库都可以各自消费结果 topic，互不影响。这个思路和“实时数仓分层”很像，中间结果层先落到 Kafka，下游再按需取用。

写回 Kafka 的 DDL 非常简单：

CREATE TABLE result_kafka ( window_start TIMESTAMP(3), category STRING, order_cnt BIGINT, gmv DECIMAL(14, 2) ) WITH ( 'connector' = 'kafka', 'topic' = 'order_stats_topic', 'properties.bootstrap.servers' = 'kafka-1:9092,kafka-2:9092', 'format' = 'json' ); INSERT INTO result_kafka SELECT TUMBLE_START(ts, INTERVAL '1' MINUTE), category, COUNT(*), SUM(amount) FROM order_source GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), category;

消费端如果是 SpringBoot，配置好 Kafka 的 consumer 就能拿到 JSON 格式的统计结果，直接推页面或者做阈值告警都行。这种链路我实际用下来最大的好处是解耦，统计结果和业务系统之间没有强依赖，下游哪怕挂了，结果还在 Kafka 里躺着，恢复后重新消费就行。

5.2 JDBC 写入 MySQL 的参数推荐

如果统计结果要直接对接到报表系统或者管理后台，写到 MySQL 是最省事的方式。除了 4.2 里的 JDBC 建表 DDL，有几个参数我建议你重点关注：

• sink.buffer-flush.max-rows：默认 100，建议调大到 1000，减少 flush 次数；
• sink.buffer-flush.interval：默认 1s，建议设成 2s，让数据攒一攒再批量写；
• sink.parallelism：不要设置太高，2 到 4 就够了，太高会给 MySQL 造成不必要的连接和写入压力。

JDBC sink 默认的写入语义是 At Least Once，作业重启时可能出现重复写入。要保证下游数据最终一致，一个简单可靠的做法是 MySQL 结果表设计联合唯一键，比如把 window_start、window_end、category 三个字段建唯一索引，然后底层用 INSERT ON DUPLICATE KEY UPDATE 的语义去更新。Flink JDBC connector 本身不直接支持这种 upsert 语法，所以很多团队会把结果先写到 Kafka，再用一个轻量消费者异步写入 MySQL，这是另一个话题了，但思路可以记下来。

5.3 Checkpoint 和状态后端：不配置等于白做

Flink 的容错全部依赖 checkpoint，这句话我在多个场合强调过。如果 checkpoint 没开，作业一重启，状态全丢，Kafka offset 也要从配置的起始位置重新消费，结果就是重复计算、数据错乱。实时统计作业，checkpoint 是必须配置的。

Java 代码里配置如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 每 60 秒触发一次 checkpoint env.enableCheckpointing(60_000); // 精确一次语义 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 两次 checkpoint 之间最少间隔 30 秒，避免频繁快照 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000); // checkpoint 存储位置，生产环境建议 HDFS env.getCheckpointConfig().setCheckpointStorage("file:///opt/flink/checkpoints"); env.setStateBackend(new EmbeddedRocksDBStateBackend()); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

RocksDB 状态后端是我上线 Flink SQL 作业的默认选择。相比 HashMapStateBackend，RocksDB 把状态存储从 JVM 堆内挪到了堆外磁盘，状态量再大也不会 OOM，只是读写性能比纯内存差一些。实时统计作业的状态通常不小，用 RocksDB 更稳。

SQL 侧还可以设置状态 TTL，把过期状态清理掉：

SET 'table.exec.state.ttl' = '2h';

这个配置对 GROUP BY 和 JOIN 都生效。尤其对大窗口、大 key 数量的统计，TTL 能明显控制状态大小。但注意别设置太短，TTL 太短会导致窗口还没结束，状态已经被清理，统计结果瞬间变错。我一般按窗口长度的 3 到 5 倍来设置。

6. 我踩过的坑和排查实录

6.1 时间字段与时区差 8 小时

这个坑几乎所有用 Flink SQL 的人都躲不过。业务库里的时间字段通常是北京时间，但 Flink 的 TIMESTAMP 类型默认按 UTC 处理，窗口边界如果按 UTC 计算，你看到的统计分钟窗口会比真实时间慢 8 小时。

解决办法是把时间字段定义成 TIMESTAMP_LTZ(3)，并设置 Flink 的本地时区：

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

建表时这样声明：

CREATE TABLE order_source ( ... ts TIMESTAMP_LTZ(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND ) WITH (...);

TIMESTAMP 和 TIMESTAMP_LTZ 的区别，简单记就是：TIMESTAMP 不带时区，看到什么就是什么；TIMESTAMP_LTZ 带时区语义，Flink 会根据本地时区转换后参与窗口计算。如果 Kafka 里存的是 epoch 毫秒数，用 TIMESTAMP_LTZ 会非常合适。存的是格式化字符串时间，建议先用 TO_TIMESTAMP_LTZ 转成 TIMESTAMP_LTZ 再参与计算。

6.2 Kafka 消费延迟高，怎么定位

消费延迟高是实时统计最常见的运维问题。表现就是作业还在跑，但结果指标明显落后于当前时间。定位延迟第一件事不是看 Flink 日志，而是看 Kafka 消费组的 LAG：

kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --group flink-order-stat-group --describe

输出里能看到每个分区的 LOG-END-OFFSET 和 CURRENT-OFFSET，两者差值就是积压量。如果 LAG 持续增长，说明消费速度跟不上生产速度。常见原因按概率排序：sink 写入 MySQL 太慢、并行度不足、状态太大导致 checkpoint 耗时变长、GC 频繁。排查顺序先看下游，再看并行度，最后才看状态。

曾经遇到过一次统计作业 LAG 飙升，查到最后是 MySQL 结果表的一个索引失效，导致 sink 单条写入越来越慢。去掉索引并调大 buffer-flush 参数后，LAG 立刻降下来。实时作业的下游存储性能，往往是最先出问题的环节。

6.3 状态无限膨胀导致作业 OOM

COUNT(DISTINCT)、大窗口聚合、普通 TopN 都会让状态越来越大。我踩过一次 UV 状态把 RocksDB 撑到几十 GB 的坑，起因是对用户 ID 做了精确去重，且没有设置状态 TTL，状态只增不减。

应对策略有三层。第一层，合理使用窗口化统计，让状态随着窗口结束自动清理；第二层，设置 table.exec.state.ttl，把超过窗口周期的旧状态清掉；第三层，对去重类指标改用近似算法，比如 HyperLogLog，牺牲少量精度换状态量级的下降。

这里特别提醒一点：不要盲目调大 JVM 堆内存来扛状态。状态用 RocksDB 落到磁盘后，内存不够时可以用磁盘存状态，但磁盘也不是无限的，最终还是要从业务逻辑上减少状态量。实时统计的常态是“能用窗口表达就别用全局累加”。

6.4 Kafka 连接报错速查表

整理了我在实操中遇到过的几个高频 Kafka 连接问题，放在一起方便排查：

遇到 Kafka 连接问题，先确认基础网络，再往上查配置。很多人一上来就翻 Flink 日志，其实 Flink 的报错信息已经明确写了“Disconnected”或者“Timeout”，根因往往就是网络或者 listeners 配置，和 Flink 本身没关系。

做实时统计这么久，我个人的体会是：Flink SQL 把计算逻辑的门槛降得很低，难点已经从“怎么写聚合”转移到了“怎么把链路的每个环节配置正确”。从 Kafka 消费的位点、水印的设定、窗口的选择，到 sink 的幂等、checkpoint 的恢复，每一环都要对背后的原理有清晰认知，不然作业跑起来是一回事，跑得稳不稳定、数据准不准确又是另一回事。

最后再分享一个落地建议。第一次上手，不要追求指标多，先拿一个订单 topic 做分钟级 GMV 统计，把这个链路完整跑通，再逐步加窗口 TopN、UV、多维对比。同时把 checkppoint、状态后端、日志监控这些基础设施在第一个作业就配好，后面加作业就只是加 DDL 和 SQL 模板的事。这套东西一旦跑顺，后面新需求基本就是流水线作业了。