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 的优势最好的方式是和其它方案摆在一起对比。我把常见做法整理成了一张表造数方案上手成本可控性无界流支持适合场景手写 Java Source高需编译打包强几乎一切可控强生产级自定义数据源静态 JSON 文件回放低弱量有限弱简单的 Demo 演示外部 Mock 工具如 Mockoon 等中需要部署服务中弱API 层面的模拟克隆线上数据 脱敏高权限和清洗成本大强中回归测试、性能仿真DataGen SQL Connector极低一条 SQL 建表中高强本地调试、压测、边界数据构造从表格能看出来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 的常用参数整理成表格按全局参数和字段级参数分开看结构会清晰很多参数层级参数名默认值作用说明全局connector无固定为 datagen全局rows-per-second10000每秒生成的最大行数全局number-of-rows无有限行数模式不配置则无限生成字段级fields.字段名.kindrandom生成策略random 或 sequence字段级fields.字段名.start0sequence 起始值字段级fields.字段名.endInt/Long 最大值sequence 结束值到达后回绕字段级fields.字段名.length4字符/字符串类型生成长度字段级fields.字段名.min0random 数值范围的下界字段级fields.字段名.max类型最大值random 数值范围的上界字段级fields.字段名.seed无随机数种子固定后可复现字段级fields.字段名.max-past0时间类型随机偏移范围单位毫秒重点看几个容易忽略的参数。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);这个查询会一直滚动验证窗口逻辑是否正常。写入 KafkaCREATE 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为 9223372036854775806end为 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 字段配了minFlink 在验证参数时会直接拒绝。解决办法是先确认字段类型对应的可配置参数不要照搬别人的 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一张表的时间就能感受到差距。