Flink DataGen SQL Connector：一条SQL搞定测试数据生成与压测

做Flink开发这几年，最烦的事往往不是业务逻辑写不出来，而是没有数据可测。Kafka还没打通、业务库不能随便连、临时表还没就绪，但你已经急着验证一个窗口聚合、一条写入链路、或者一组规则的效果。这种时候，Flink DataGen SQL Connector 就是我最常用的“解药”。它只需要一条 CREATE TABLE 语句，就能按你指定的字段类型、取值范围、生成速率，源源不断地产出数据。

一句话概括它的价值：不需要写 Java 代码、不需要外挂 Mock 服务、不需要准备测试文件，在 Flink SQL 里用配置的方式就能完成造数这件事。这篇文章我会从四个方向讲透它：本地造数怎么做、压测参数怎么调、边界数据怎么构造、以及怎么把随机数据“伪装”成看起来像线上真实流量的数据。如果你是刚开始接触 Flink SQL，或者已经用了一段时间但对 DataGen 的理解只停留在“能出数”，那这篇文章正好适合你。

1. 把 DataGen 放在工具链的哪个位置：设计思路与选型对比

1.1 为什么需要 DataGen：手工造数的痛点

写测试数据的痛点，经历过的人都懂。手写一批 JSON 文件回放，数据量一上去就撑不住；用 Java 写一个 Source，每次改字段都要重新打包；从线上 Clone 数据，不仅涉及隐私和权限，还要清洗脱敏。更麻烦的是，很多场景需要的是无界流——一个窗口作业要连续跑半个小时才能观察到稳定的吞吐指标，手工准备的数据根本撑不了那么久。

我第一次接触 DataGen 的时候，第一反应是这东西能有多少人用？后来才发现，它恰恰解决的是“本地和测试环境没有上游数据源”这个最普遍的问题。它不是一个生产级的数据同步工具，而是一个为 Flink 作业本身服务的“测试数据生成器”。你把它的输出接到 Kafka、JDBC、文件系统，或者直接在 Flink 内部做后续计算，一条链路很快就能跑起来。

1.2 常见造数方案对比：为什么多数时候选 DataGen

要理解 DataGen 的优势，最好的方式是和其它方案摆在一起对比。我把常见做法整理成了一张表：

从表格能看出来，DataGen 最大的优势是低门槛和无界能力。它不需要额外服务，Flink 自带的连接器里就有；它作为流式 Source 天然持续运行，不会像静态文件那样几秒就结束。如果你的造数需求是“字段类型明确、范围可控、速率可调”，DataGen 就是第一选择。

1.3 它的边界在哪里：什么场景不该用

任何工具都有边界。DataGen 不适合用来模拟非常复杂的业务关联，比如多表之间强一致的主外键关系、跨字段的业务规则校验、复杂加密逻辑等。它的字段生成是相互独立的，虽然可以通过表达式做部分关联模拟，但本质上它不是业务仿真器。

另外，DataGen 不会真的读取外部数据，所以如果你要测的是“从 MySQL 上游同步的时效性”这类问题，那应该用 CDC 工具而不是 DataGen。我在实际项目里的分工很简单：验证 Flink 作业逻辑本身、做链路压测、构造边界输入，用 DataGen；验证上下游集成，用真实的连接器和真实环境。把这两件事分清楚，工具就不会被“用错地方”。

2. 把 DataGen 用明白：核心参数与字段类型逐个拆

2.1 最小可运行示例：先让数据跑起来

不看任何文档，先给你一段可以直接跑通的 SQL。假设你面前有一个 Flink SQL Client 或者某个支持 Flink SQL 的开发环境，执行下面这个语句：

CREATE TABLE datagen_basic ( id BIGINT, name STRING, score DOUBLE, ts TIMESTAMP(3) ) WITH ( 'connector' = 'datagen', 'rows-per-second' = '100', 'fields.id.kind' = 'sequence', 'fields.id.start' = '1', 'fields.id.end' = '1000', 'fields.name.length' = '8', 'fields.score.min' = '0', 'fields.score.max' = '100', 'fields.ts.kind' = 'random', 'fields.ts.max-past' = '3600000' );

然后执行简单的查询：

SELECT * FROM datagen_basic;

你会发现数据源源不断地刷新。id 从 1 开始递增到 1000 后回绕，name 是一串随机英文字符，score 在 0 到 100 之间浮动，ts 是过去一小时内的某个随机时间点。这个例子虽然短，却已经覆盖了 DataGen 最核心的两种字段生成方式：sequence 和 random。

2.2 参数速查表：哪些参数是你必须知道的

我把 DataGen 的常用参数整理成表格，按全局参数和字段级参数分开看，结构会清晰很多：

重点看几个容易忽略的参数。number-of-rows用于有限生成，比如你只想造 1000 行数据然后让任务自动结束，就可以设置它，否则任务会一直跑。seed字段级参数非常重要，设置了固定的 seed 之后，每次运行生成的随机序列完全相同，这个特性对测试复现极其有用，后面我会专门讲。

2.3 字段类型支持与生成策略的原理

DataGen 支持大部分 Flink SQL 常见类型，包括数值型、字符型、时间型、布尔型，以及一部分复杂类型。官方文档里的类型列表很长，实际用下来我总结成三条经验：

第一，整数和浮点类型都支持 min 和 max 范围控制，这对构造业务字段非常方便。第二，字符串类型不支持 min/max，只支持 length 控制长度，而且默认只生成字母和数字字符，需要中文或者字典值需要通过表达式字段解决。第三，时间类型比较特殊，random 模式下不会生成任意历史时间，而是围绕当前时间附近随机偏移，max-past控制最长回退时间，这个设计看似限制，实际对测试窗口聚合特别友好。

sequence 策略的原理则是从 start 递增到 end，到达 end 后重新回到 start。它不仅支持数值类型，也支持时间戳。比如你可以生成从 2024-01-01 00:00:00 开始递增的 ts，这样在测乱序和 watermark 的时候就非常可控。

2.4 全局参数和并行度：别把速率理解错了

DataGen 的rows-per-second是全局速率还是每个并行子任务的速率？这个问题在不同版本里表现有差异，但按官方语义理解，它控制的是整体产出速率。也就是说，你把并行度从 1 调到 8，不会让数据量变成 8 倍，而是由多个子任务分担这每秒 N 行的配额。

这里有一个我在实践中踩过的坑。当你开启很高的并行度时，sequence 字段的连续性不再像单并行度那样一目了然，因为多个子任务各自持有一段序列范围。如果你想用 DataGen 生成“全局唯一且连续”的编号，建议直接观察整体输出而不是依赖单个子任务的顺序。真数压测中需要模拟高并发写入时，我更倾向于把并行度和速率分开调：先固定速率，逐步加并行度观察吞吐变化，再反过来调整，这样才能找到当前资源下的最佳平衡点。

3. 实操：从启动 Flink 到造出一张“订单表”

3.1 快速起一个本地 Flink 环境

本地体验 DataGen 最省事的方式是直接跑 Flink SQL Client。你可以下载 Flink 发行版，在本地启动一个单机集群：

# 启动集群 ./bin/start-cluster.sh # 启动 SQL Client ./bin/sql-client.sh embedded

如果你不想折腾环境，用 Docker 也一样，官方镜像里已经自带了所有常用连接器。我平时会直接用 Flink SQL Client 来做快速验证，因为它启动快、没有 IDE 依赖，适合临时造数。有一点要注意：SQL Client 的默认依赖包含 datagen 连接器，但如果你使用自定义的 Flink 发行包，需要确认flink-table-planner-loader和连接器 Jar 是否齐全。遇到 ClassNotFound 之类的报错，优先检查 Jar 依赖。

3.2 定义一张接近真实业务的订单表

造数不能只求“有数据”，更要“像业务”。我用一个电商订单场景做例子，包含订单号、用户 ID、金额、订单状态、城市、优惠折扣、下单时间、支付时间。SQL 如下：

CREATE TABLE orders_datagen ( order_id BIGINT, user_id BIGINT, amount DOUBLE, status STRING, city STRING, discount DOUBLE, order_ts TIMESTAMP(3), pay_ts TIMESTAMP(3), is_paid BOOLEAN ) WITH ( 'connector' = 'datagen', 'rows-per-second' = '500', 'fields.order_id.kind' = 'sequence', 'fields.order_id.start' = '20240001', 'fields.order_id.end' = '20249999', 'fields.user_id.kind' = 'random', 'fields.user_id.min' = '10001', 'fields.user_id.max' = '30000', 'fields.amount.kind' = 'random', 'fields.amount.min' = '9.9', 'fields.amount.max' = '9999.0', 'fields.status.length' = '2', 'fields.city.length' = '5', 'fields.order_ts.kind' = 'random', 'fields.order_ts.max-past' = '86400000', 'fields.pay_ts.kind' = 'random', 'fields.pay_ts.max-past' = '86400000', 'fields.is_paid.kind' = 'random' );

跑起来之后你会发现，status 和 city 生成的是随机字符，看起来跟线上完全对不上。这是纯 random 的局限。先记住这个痛点，后面讲“像真数据”的时候我会专门给出解决方案。

3.3 把数据接到不同 Sink：验证完整链路

造数的最终目的是喂给下游逻辑。最常用的三种接法：

直接参与 Flink 计算：

SELECT TUMBLE_START(order_ts, INTERVAL '1' MINUTE) AS win_start, COUNT(*) AS cnt, SUM(amount) AS total_amount FROM orders_datagen GROUP BY TUMBLE(order_ts, INTERVAL '1' MINUTE);

这个查询会一直滚动，验证窗口逻辑是否正常。写入 Kafka：

CREATE TABLE kafka_sink ( order_id BIGINT, amount DOUBLE, status STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'orders_test', 'properties.bootstrap.servers' = 'localhost:9092', 'format' = 'json' ); INSERT INTO kafka_sink SELECT order_id, amount, status FROM orders_datagen;

写入 JDBC 表：你可以把 DataGen 的字段映射到 MySQL 表，重点测试批量写入、主键冲突、字段长度截断等情况。这一步对验证 Sink 的健壮性特别有用。

我提醒一下：如果只是验证 Flink 作业自身逻辑，不要轻易把数据发到外部系统再读回来，这样会让排查链路变长。优先在 Flink 内部完成“生成-计算-校验”，最后一步再打通外部 Sink。

3.4 数据分布的手工校验

造完数之后，怎么知道分布是否符合预期？如果只是SELECT *快速看一眼，大概率看不出毛病。我常用的做法是直接在 SQL 里做分布统计：

SELECT status, COUNT(*) AS cnt, ROUND(AVG(amount), 2) AS avg_amount, MIN(order_ts) AS min_ts, MAX(order_ts) AS max_ts FROM orders_datagen GROUP BY status;

如果预期是 status 大致均分，实际统计却严重偏斜，说明生成参数设置有问题。再看 min/max 的 ts 是否落在max-past范围内，就能验证时间字段的偏移是否符合预期。这里要记住一个经验：不要只看一行数据，要用聚合函数看数据分布。

4. 边界数据、压测与“像真数据”的生成技巧

4.1 边界数据怎么构造：空值、极值、特殊值全覆盖

测试中比正常数据更重要的，往往是边界数据。代码能不能抗住零值、负值、最大值、空串、NULL，才是稳定性的关键。DataGen 看起来很“随机”，但构造精确的边界值是有方法的。

数值边界场景：用 sequence 的 start 和 end 生成最大值附近的数值。比如测试 BIGINT 最大值场景，设置start为 9223372036854775806，end为 9223372036854775807，就能稳定产出接近溢出的数据。测试金额为 0 的场景，将 min 和 max 都设为 0，整个字段恒为 0。测试负数场景，设定 min 为 -1000、max 为 -1 即可。

空串和 NULL 场景：DataGen 本身生成的字符串不会是空串，但你可以用计算列来处理。Flink 的 CREATE TABLE 语法支持在字段名后加 AS 表达式，利用内置函数生成特殊值：

CREATE TABLE boundary_test ( id BIGINT, name STRING, name_null STRING, eof_flag STRING ) WITH ( 'connector' = 'datagen', 'fields.id.kind' = 'sequence', 'fields.id.start' = '1', 'fields.id.end' = '100' );

注意上面的 SQL 还没写完，因为我需要把特殊值逻辑放在源数据之后做二次处理。更常见的做法是建一张基础 DataGen 表，然后通过 INSERT INTO SELECT 或者流式查询，用 SQL 函数把边界值构造出来：

CREATE TABLE boundary_result AS SELECT id, CASE WHEN id % 10 = 0 THEN NULL ELSE CONCAT('user_', CAST(id AS STRING)) END AS name_with_null, CASE WHEN id % 10 = 0 THEN '' ELSE CONCAT('user_', CAST(id AS STRING)) END AS name_with_empty FROM boundary_test;

日期边界：Flink 的 DATE 类型有合法范围，你可以通过计算列直接生成DATE '9999-12-31'或DATE '0001-01-01'来测试下游解析逻辑。这种做法在测试 Hive Sink 和 JDBC Sink 时特别有价值。

4.2 压测的正确姿势：速率、并行度与输出端瓶颈

拿 DataGen 做压测，很多人第一反应就是调大rows-per-second，结果发现吞吐上不去，就开始怀疑 DataGen。实际上问题多半出在输出端。DataGen 在无输出或输出为 Blackhole 模式下，速率可以冲得很高；一旦接了 Kafka、JDBC 这些外部 Sink，瓶颈很可能是 Sink 的写入能力或者网络带宽。

我有一个自己常用的压测流程，分享给你参考：

第一步，先用默认配置跑起来，确认任务本身能稳定运行。第二步，调大rows-per-second，观察 Flink Web UI 中的 Records Sent 指标。如果 DataGen 产生速率上不去，检查任务并行度和 CPU 占用。第三步，把 Sink 从 Kafka 临时换成 Blackhole 连接器，对比同样速率下的表现，判断瓶颈在 Source 还是 Sink。第四步，Sink 恢复为真实目标，逐步增大速率，找到一个在可接受延迟下不会堆积记录的临界值。

Blackhole 连接器的建表方式很简单：

CREATE TABLE blackhole_sink ( order_id BIGINT, amount DOUBLE, status STRING ) WITH ( 'connector' = 'blackhole' );

压测中还要留意一个隐蔽问题：Flink 的反压机制。当 Sink 写不过来了，DataGen 的速率也会跟着降下来，这在 Web UI 上看起来就像“DataGen 不行了”，实际是下游保护机制生效了。所以判断压测结果时，一定要结合 Sink 侧写入延迟来看，不要只盯着 Source 侧速率。

4.3 让随机数据“像真数据”：分布、字典、依赖字段与时间漂移

这一节是全文我最想分享的部分。DataGen 默认产出的随机数据一眼就能看出来是假的，但稍微动几个脑筋，你完全能让它“像那么回事”。

字典字段的真值模拟。之前提到city.length生成的是乱码，解决方法是用计算列映射。比如城市字段，理想情况下应该大致符合业务分布：

CREATE TABLE order_with_dict ( order_id BIGINT, user_id BIGINT, amount DOUBLE, city STRING, status STRING, order_ts TIMESTAMP(3) ) WITH ( 'connector' = 'datagen', 'fields.order_id.kind' = 'sequence', 'fields.order_id.start' = '1', 'fields.order_id.end' = '1000000', 'fields.user_id.kind' = 'random', 'fields.user_id.min' = '1', 'fields.user_id.max' = '100000', 'fields.amount.kind' = 'random', 'fields.amount.min' = '10', 'fields.amount.max' = '1000', 'fields.order_ts.kind' = 'random', 'fields.order_ts.max-past' = '604800000' ); CREATE TABLE order_enriched AS SELECT order_id, user_id, CAST(10 + RAND() * 990 AS DECIMAL(10, 2)) AS amount_precise, CASE WHEN RAND() < 0.45 THEN '上海' WHEN RAND() < 0.7 THEN '北京' WHEN RAND() < 0.85 THEN '广州' ELSE '成都' END AS city, CASE WHEN RAND() < 0.6 THEN 'CREATED' WHEN RAND() < 0.8 THEN 'PAID' WHEN RAND() < 0.95 THEN 'SHIPPED' ELSE 'FINISHED' END AS status, order_ts FROM order_with_dict;

这样生成出来的数据，城市有明确的业务占比，状态机也像是一条正常订单路径。关于RAND()我在实际使用中发现，它会为每一行计算一次，因此在同一行里如果多次调用RAND()，结果是不相关的。如果你希望两个字段的字典分布保持一定的对应关系，比如“北京的订单比例 45%”，上面这种方式没问题；但如果你希望“城市为北京的同时支付渠道是微信”，就需要用同一个字段做基准，而不要再依赖新的RAND()。

关联字段的一致性模拟。这是个更高级的技巧。假设你要模拟“大用户的订单更多”这类幂律分布。可以先用 user_id 作为随机基准值，再基于它计算是否为大客户：

CREATE TABLE order_user_attr AS SELECT user_id, CASE WHEN user_id % 100 = 0 THEN 'VIP' WHEN user_id % 10 = 0 THEN 'POTENTIAL_VIP' ELSE 'NORMAL' END AS user_level FROM order_with_dict;

这样 user_level 和 user_id 之间的对应关系是稳定且可复现的，测试关联规则时就特别扎实。时间是另一个高频痛点。默认max-past只能控制随机时间偏移，如果要模拟“上周的数据”或“昨天 20:00 到 22:00 高峰期的数据”，用计算列做时段加权会比较方便。

4.4 可复现性：固定 seed 与稳定性测试

测试最怕的是什么？不是数据量太大，而是同一个任务这次跑出一个结果、下次跑出另一个结果。DataGen 的 random 模式默认没有固定种子，每次运行随机序列都不一样。如果你在调试窗口聚合、CEP 规则，会发现经常出现“上次还能命中的规则，这次复现不了”的情况。

解决方式就是给字段设置 seed。以 user_id 为例：

CREATE TABLE stable_datagen ( user_id BIGINT, amount DOUBLE ) WITH ( 'connector' = 'datagen', 'fields.user_id.kind' = 'random', 'fields.user_id.min' = '1', 'fields.user_id.max' = '100000', 'fields.user_id.seed' = '42', 'fields.amount.kind' = 'random', 'fields.amount.min' = '10', 'fields.amount.max' = '1000', 'fields.amount.seed' = '7' );

设置了 seed 之后，只要并行度、参数不变，每次运行生成的随机序列就是一致的。这在构造回归测试用例时价值巨大。我通常会给所有字段都显式配上 seed，这看起来是小事，但在团队协作中能避免很多“我这边正常、他那边的数据对不上”的争议。还要注意一点：seed 是字段维度而不是全局维度，不同字段之间用不同 seed 不影响整体复现，只要每个字段的 seed 固定即可。

5. 常见问题与排查实录

5.1 任务“不产数”或者数据不动

新手最容易遇到的情况是：CREATE TABLE 成功了，SELECT 也执行了，但结果久久不出来。大概率是rows-per-second设置得太小，比如设了 1，那每秒只有一行，窗口聚合半天统计不到多少数据；还有一种可能是任务还在初始化，Flink 控制台第一次展示结果需要一点时间。

排查思路很简单：先在 DataGen 表上执行SELECT COUNT(*) OVER ()或者直接SELECT *限制输出，把速率单独调大，比如 1000、10000，确认 DataGen 本身没问题，再逐步调回目标速率。另外要留意浏览器端的输出缓冲区，这个我之前被坑过：数据其实一直在生成，只是 SQL Client 的显示端刷新慢了。

5.2 字段类型不匹配与计算列引用限制

DataGen 对字段类型非常挑剔。常见报错是类型不匹配，比如给 DOUBLE 字段配了length，给 STRING 字段配了min，Flink 在验证参数时会直接拒绝。解决办法是先确认字段类型对应的可配置参数，不要照搬别人的 SQL，因为字段类型不同，同一个参数名的意义可能完全不同。

计算列是另一个隐藏坑。Flink SQL 中一个计算列不能引用另一个计算列，也就是说你在建表时写了A AS ...，后面的B AS A + 1是不被允许的。我一开始没注意这个限制，后来发现这种写法会被直接拒绝。替代方案是：把被依赖的字段保留为 DataGen 生成的物理字段，再在查询层用视图或 INSERT INTO 做二次加工，而不是在同一个 CREATE TABLE 里串接计算列。

5.3 随机性带来的 Flaky 测试

这里说的 Flaky 是指测试结果不稳定。比如你写了一个窗口聚合，预期每分钟数据分布基本均匀，但因为 DataGen 的随机序列没有种子，每次运行分布都不同，导致断言偶尔失败。这个问题的最佳解法就是前面说的固定 seed。如果你发现固定 seed 之后结果仍然不稳定，要检查并行度是否发生了变化，因为同一个 seed 在不同并行度下的数据分配方式可能不同。

对于压测场景，还有一个小技巧：不要只依赖 DataGen 的随机速率来衡量真实压力。如果下游处理逻辑有去重、聚合等状态操作，随机重复数据会影响状态大小。建议在 DataGen 基础上通过计算列人为控制重复率，比如user_id用rand()四舍五入到百位，造出粒度更粗的用户 ID，从而产生更多状态条目，更贴近真实场景。

5.4 不要把 DataGen 当成生产数据源

最后一条是原则性的：DataGen 是调试工具，不是生产数据源。有些人会在测试环境长期挂一个 DataGen 任务给下游 Demo 消费，这在可控环境下没问题，但要注意两个隐患：一是 DataGen 数据没有业务语义和关联约束，下游如果依赖它做积压测试，指标会有偏差；二是无边界的 DataGen 任务会持续占用资源，如果长时间不清理，多个任务累积起来可能拖垮整个 Session。我习惯的做法是：每次用完任务及时停止；需要长期提供测试数据时，把 DataGen 结果导入 Kafka，再让下游消费 Kafka，而不是让每个下游都直连 DataGen。

最后分享两个我自己的使用习惯

第一个习惯是维护一份datagen_templates.sql。我会把订单、用户、日志、点击流这些常用表结构都整理成模板，字段名、类型、取值范围、字典表达式、seed 全部统一。下次要做相关测试，直接复制模板改几个数字就行。团队里其他人要用同一套数据，也不用各自从头造。

第二个习惯是永远先确认“数据是给谁用的”。如果只是验证 SQL 语法，随便造数就行；如果要做性能压测，必须关心速率、并行度、状态大小；如果要做规则回归，优先保证可复现性和分布稳定。搞清楚了目的，DataGen 的参数配置就不会乱。如果你现在还在手工拼测试数据，或者被“没有上游数据”卡住，建议立刻开一个 SQL Client 试试 DataGen，一张表的时间就能感受到差距。