做数据开发和算法工程这些年我越来越觉得“造数据”是一项被严重低估的能力。团队里经常遇到这种情况下游报表要开发但上游数据还没接入模型训练需要脱敏样本但生产库不能直接碰性能压测要造千万行数据手工写脚本又慢又不真实。这时候在 PySpark 里生成一批合成描述性数据就是最顺手的解决方案。这篇文章我就把自己完整跑过的一套思路和代码整理出来从随机数、字典映射到 Faker UDF、分布控制再到性能优化和问题排查一步步拆开讲清楚。不吹不黑这些操作我都在真实项目里验证过照着改就能用。1. 为什么要生成合成描述性数据场景与方案选型1.1 什么算“描述性数据”先统一一下概念。这里说的描述性数据不是指一个孤立的数值列而是描述某个业务实体完整画像的数据。比如用户订单场景里描述性字段包括姓名、性别、年龄、城市、邮箱、商品类别、商品名称、订单金额、下单时间、会员等级这些。它跟纯随机数的区别在于字段之间要有业务逻辑关系值的分布要符合常理文本内容要“看起来像人写的”而不是一坨无意义的字符串。生成这类数据的价值主要落在三个地方。第一是做报表和 BI 验证你用可视化工具连上一张假表如果数据毫无业务语义看板效果根本没法演示第二是做数据管道联调上游没就绪时下游 ETL 需要一批结构完整的数据来验证字段映射、空值处理和异常告警第三是做隐私保护和脱敏真实数据不能外泄就把分布特征迁移到合成数据上让算法同学放心折腾。1.2 为什么非 PySpark 不可可能有人会问几百行数据我用 Python 写个循环不行吗用 Pandas 不香吗说实话小数据量完全行。但一旦量级上升到百万、千万甚至亿行单机方案就会暴露两个硬伤。一是内存瓶颈Pandas 处理几千万行时内存占用常常飙到几十个 GB普通开发机直接卡死二是分布式思维缺失很多数据管道本身就是跑在 Spark 上的你造的数据如果没经过分区的锤炼下游可能根本压不出性能问题。PySpark 的方案解决的就是这两个问题。它把数据生成的过程分布式化了——每个 Executor 负责生成一部分数据最后合并成一张完整的 DataFrame。这不仅是性能上的提升更重要的是你可以在生成数据的同一套代码里直接完成所有的 ETL 验证、特征工程测试和倾斜场景模拟。可以说PySpark 造数据不只是为了数据本身更是为了模拟生产环境的真实压力。1.3 三条技术路线怎么选做合成数据我用过三种思路各有优劣这里直接给结论。纯 Spark SQL / DataFrame 函数用 rand、randn、when、element_at 这些内置函数生成随机数、做字典映射。优点是极致快几亿行也能跑缺点是生成文本类的“人话”字段能力弱姓名地址这种得靠拼接。字典表 广播连接先把城市、商品、邮局域名这些枚举值做成小表再通过 join 关联到大数据上。优点是字段间的关联关系能做得非常干净符合真实业务逻辑缺点是写起来比纯函数略繁琐。Faker 库 Python UDF用第三方库模拟姓名、地址、手机号、邮箱真实感最强。缺点是普通 UDF 性能差必须改造成 pandas UDF 或者尽量少用否则大数据量会打到崩溃。我实际的组合策略是数值和类别字段交给内置函数和字典表文本类字段用 pandas UDF 收尾。这样既有速度又有真实感下面逐步展开。2. 核心实现用 PySpark 内置函数造出用户订单描述数据2.1 基础环境和数值型字段的生成先用一段最基础的代码把环境搭起来。我习惯把 SQL shuffle 分区数显式设置小一点避免生成数据时产生大量无意义的小文件。from pyspark.sql import SparkSession from pyspark.sql.functions import ( rand, randn, floor, lit, concat, col, md5, when, expr, element_at, array, from_unixtime, unix_timestamp, round as spark_round, concat_ws ) spark SparkSession.builder \ .appName(synth_descriptive_data) \ .master(local[4]) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate()然后先生成一百万行的基础用户表。注意spark.range是生成分布式自增 ID 最直接的方式它会严格保证每个 ID 不重复天然适合做主键。user_df spark.range(0, 1000000).withColumn(user_id, concat(lit(U-), md5(col(id).cast(string))))这里有一个实用细节user_id 如果用简单的 U-000001虽然可读性好但如果在后续 join 中暴露了行数信息就太“假”了。用 md5 加工之后用户 ID 看起来像是脱敏过的真实 ID演示效果更好。当然代价是长度会变长如果业务要求定长 ID可以 substring 截取前 16 位。接着生成年龄和性别。这一步的核心坑点是直接写多个withColumn(age, ...)和withColumn(gender, ...)每次都会调用全新的 rand()导致同一行的年龄和性别彼此独立这没问题但不可复现。如果要可复现就需要给 rand 传入同一个种子。user_df user_df \ .withColumn(rnd_age, rand(seed42)) \ .withColumn(rnd_gender, rand(seed43)) \ .withColumn(age, floor(col(rnd_age) * 60 18)) \ .withColumn(gender, when(col(rnd_gender) 0.5, M).otherwise(F))这样生成的年龄在 18 到 77 之间连续均匀分布。为什么不用randn来生成年龄因为现实中的用户年龄更接近偏态分布大量集中在 18 到 40 之间而不是均匀铺开。如果你要更逼真可以在基础均匀值上做一个量级变换比如floor((col(rnd_age) ** 1.5) * 35 18)这样低龄段密度会更高。这里先保持简单分布控制放到第 3 节专门讲。2.2 类别字段字典数组与 element_at类别字段是描述性数据里最能让数据“有业务感”的部分。城市、性别、会员等级、支付方式都属于这一类。最简单的写法是 CASE WHEN 一长串但城市几十个的时候代码丑到不想维护。我推荐用element_at配合数组字面量把字典映射变成一行代码。city_list [北京, 上海, 广州, 深圳, 杭州, 成都, 武汉, 西安, 南京, 重庆] user_df user_df \ .withColumn(rnd_city, rand(seed44)) \ .withColumn(city, element_at( array([lit(c) for c in city_list]), (col(rnd_city) * len(city_list)).cast(int) 1 ))解释一下这里的原理array([lit(c) for c in city_list])创建了一个包含所有城市名的数组列element_at(arr, idx)负责按照下标取值。注意 Spark 数组的下标从 1 开始所以要通过1把 0 到 N-1 的随机索引转换为 1 到 N。为什么这样比 cross join 字典表或者 UDF 更好因为element_at是内置函数在 Executor 内部直接完成取值操作没有 shuffle也没有 Python worker 的序列化开销。实测在千万行规模下这一段基本上只是一次全表扫描的代价。同样的套路可以直接套用到支付方式、季节、渠道来源等所有枚举型字段上。如果你有一批字典数据存在文件里也可以先读成 Python list 再塞进数组字面量效果完全一样。2.3 文本描述字段组合词库生成商品名真实场景里商品名这类描述性文本往往带有“品牌 品类 型号 颜色”的组合规律。直接用 Faker 生成商品名反而失真因为真实商品名更像是模板拼接。我的做法是先维护几个词库然后随机组合brands [云雀, 极光, 摩登, 蓝鲸, 方舟] categories [无线耳机, 机械键盘, 智能手表, 便携音箱, 运动手环] modifiers [Pro, Lite, Max, 青春版, 专业版] colors [曜石黑, 珍珠白, 远峰蓝, 暗夜绿, 樱粉] product_df spark.range(0, 1000000) \ .withColumn(rnd_brand, rand(seed50)) \ .withColumn(rnd_cat, rand(seed51)) \ .withColumn(rnd_mod, rand(seed52)) \ .withColumn(rnd_color, rand(seed53)) \ .withColumn(product_name, concat_ws( , element_at(array([lit(b) for b in brands]), (col(rnd_brand) * len(brands)).cast(int) 1), element_at(array([lit(c) for c in categories]), (col(rnd_cat) * len(categories)).cast(int) 1), element_at(array([lit(m) for m in modifiers]), (col(rnd_mod) * len(modifiers)).cast(int) 1), element_at(array([lit(c) for c in colors]), (col(rnd_color) * len(colors)).cast(int) 1) ))这样生成出来的商品名实际效果是“极光 机械键盘 Max 曜石黑”看起来非常真。如果你需要 SKU 级别的唯一标识可以在 product_name 后面再拼一个md5(col(id).cast(string))的前 8 位作为货号就形成了“产品描述 唯一规格”的真实结构。这里的细节是品牌和品类之间其实存在真实的业务约束比如“蓝鲸”这个品牌可能只做耳机不做键盘。如果你要完美复现这种约束不能靠独立随机列而要先根据品牌或者用同一个随机列派生再去限制品类的候选集合。绝大多数造数需求不需要做到这一步但需要知道方向在哪。2.4 时间字段制造一段业务时间窗时间字段是描述性数据里最容易出错的地方。很多人直接current_timestamp()一把梭结果所有数据都是同一时刻看起来非常假。更合理的方式是生成一个业务时间范围内的随机时间戳。start_ts unix_timestamp(lit(2021-01-01 00:00:00)) end_ts unix_timestamp(lit(2023-12-31 23:59:59)) order_df product_df \ .withColumn(rnd_time, rand(seed60)) \ .withColumn(order_time, from_unixtime(col(rnd_time) * (end_ts - start_ts) start_ts).cast(timestamp) )原理不复杂unix_timestamp 把日期字符串转成自 1970 年以来的秒数rand 生成 0 到 1 的浮点数乘上时间窗口的秒数跨度就得到了窗口内的一个随机秒偏移量再加回起点最后 from_unixtime 转回可读格式。这里有个我没少踩的坑如果直接用rand() * 78840000一年约 3153 万秒这种魔法数字代码可读性会很差。用两个变量把 start_ts 和 end_ts 算出来再参与运算审代码的人一眼就能看懂时间范围。如果想让时间呈现“白天单量多、凌晨单量少”的规律可以再加一个正弦变换来模拟波动或者直接用当天的 hour 字段做重采样。我先不展开这部分属于第 3 节分布控制的延伸。3. 让数据更像真实业务分布控制与字段关联3.1 均匀分布的局限性如果所有随机字段都是均匀分布数据会显得非常“假”。真实业务里订单金额一定是小额多、大额少用户年龄一定是年轻和中青年占大头注册时间往往随活动波峰波谷变化。均匀分布适合造测试数据但不适合造演示和模型训练数据。PySpark 提供了两类随机函数rand()生成 0 到 1 的均匀分布randn()生成标准正态分布。均匀分布可以直接线性映射到任意区间正态分布适合建模身高、体重、考试分数这类聚集在均值附近但存在长尾的特征。而订单金额通常更靠近对数正态分布或幂律分布直接套正态分布会让均值附近堆积过多数据大额订单的尾部完全出不来。3.2 用同一个随机列派生关联字段字段关联是整个合成数据里最值得花心思的地方。真实订单里VIP 等级越高的用户订单金额通常越大一线城市的客单价通常高于三四线城市老用户的复购间隔往往更短。如果每个字段都独立随机这些关联关系就全部消失了数据看起来“字段都对逻辑全无”。核心技巧非常简单让多个字段共享同一个随机列或用同一个随机列做分段映射。举个例子user_order_df spark.range(0, 1000000) \ .withColumn(rnd, rand(seed70)) \ .withColumn(vip_level, when(col(rnd) 0.3, 普通) .when(col(rnd) 0.6, 银卡) .when(col(rnd) 0.85, 金卡) .otherwise(黑金卡)) \ .withColumn(amount_multiplier, when(col(vip_level) 普通, expr(round(rand() * 100 20, 2))) .when(col(vip_level) 银卡, expr(round(rand() * 300 100, 2))) .when(col(vip_level) 金卡, expr(round(rand() * 800 300, 2))) .otherwise(expr(round(rand() * 2000 800, 2))))这里 vip_level 由rnd一个随机列决定所以它能被稳定复现金额的基准范围又由 vip_level 决定这就形成了“VIP 等级高 → 金额高”的业务逻辑。注意我在每个区间内部又用了独立的 rand()这样等级与金额有关联但同等级内的金额依然有随机波动不会显得死板。同样的思路可以用来造“城市与客单价”关联先随机一个城市再根据城市所处的城市等级走不同的金额基准。这需要多维护一张城市等级映射表但实现成本很低。3.3 幂律和偏态分布在金额上的应用如果要生成更贴近真实电商的订单金额我推荐直接用指数分布或者对数正态分布。指数分布天然符合“大量小单 少量大单”的形态。借助 PySpark 的内置数学函数可以这样构造from pyspark.sql.functions import exp, log amount_df spark.range(0, 1000000) \ .withColumn(rnd_exp, rand(seed80)) \ .withColumn(amount, spark_round( exp((1 - col(rnd_exp)) * 6) * 10, 2 ) )解释一下1 - rand()把均匀随机数翻转成接近 0 到 1 但偏向小的数值乘以 6 之后再取指数得到的数值大部分集中在几十到几百但尾部可以看到几千甚至上万的大单分布形态非常接近真实的客单价。乘以 10 是做一个量级缩放让均值落在合理范围。如果你要严格的对数正态分布可以用exp(randn(seed) * sigma mu)。我用 sigma0.8、mu4 的话生成金额的 90% 会落在几十到几百元区间头部少量几千元整体观感很舒服。调参技巧是mu 决定整体量级sigma 决定长尾的厚薄想让大单更极端就调大 sigma。3.4 空值与异常值注入完全干净的数据也不是真实的。生产环境里的数据管道最常见的问题就是空值、脏字符和超范围值所以造数时必须有意识地把这些“坏数据”混进去。空值注入不建议直接随机把某列置为 null那样会破坏字段关联。我的做法是单独准备一个data_quality随机列用它控制每行哪些字段参与空值生成bad_data_df raw_df \ .withColumn(rnd_dq, rand(seed90)) \ .withColumn(email, when(col(rnd_dq) 0.02, lit(None)) .otherwise(concat( lower(col(user_id)), lit(example.com) )))这样可以精确控制空值率比如邮箱字段的缺失率约为 2%手机号字段缺失率约为 0.5%地址字段缺失率约为 5%。把这些缺失率设置成和线上统计值一致下游管道在处理时就能提前暴露问题。异常值的生成更讲究技巧不应该随机生成超范围的值而应该针对特定字段制造典型脏数据。比如年龄字段混入负数和 200 这样的非法值订单金额混入负数和超大量级。特别是金额字段可以混入少量负数代表退款单这在业务上是合理的存在报表处理时必须过滤或转换。4. 规模放大与性能优化从百万行到亿行4.1 用宽表和 union 放大行数百万行很多时候不够用。下游压测往往要上亿行这时候有两条路线。一条是直接改spark.range(0, 100000000)让 Spark 自然分布式生成另一条是先造一百万行的“模板”再通过 union 放大。直接改 range 最省事但模板方案有一个好处可以精确控制每百万行模板的字段分布保证多轮生成的用户画像一致。放大行数的另一个技巧是利用交叉维度生成宽表。比如你有 10 万个用户、1000 个商品理论上订单明细就是用户 × 商品的一个关系子集。用 PySpark 生成这种交叉数据时可以把用户表和商品表都广播出去之后做 join控制概率就能得到真实的订购关系比单纯造一亿行更有业务价值。4.2 UDF 与 pandas UDF 的取舍文本类字段如果全用内置函数拼接真实感始终有限。姓名、地址、邮箱这种字段最省心的替代方案是 Faker 库。但普通 UDF 的性能确实一言难尽每行都要 Python 解释器参与在千万行量级下会慢到怀疑人生。我实测下来的经验是尽量用pandas_udf替代普通 UDF。pandas UDF 基于 Apache Arrow打包一批数据进 Python再用 Pandas 向量化操作性能差距可达几十倍。代码写起来差别不大from pyspark.sql.functions import pandas_udf import pandas as pd pandas_udf(string) def fake_name_udf(s: pd.Series) - pd.Series: local_fake Faker([zh_CN]) return pd.Series([local_fake.name() for _ in range(len(s))]) user_names user_df.withColumn(user_name, fake_name_udf(col(user_id)))这里有个细节在 pandas UDF 内部创建 Faker 实例每个批次的 worker 都会执行一次。如果一批里调用次数很多在函数内部Faker(zh_CN)会因为反复初始化而变慢。更优的写法是把 Faker 实例提到函数外面用全局变量但 Spark 的 Python worker 进程是复用的全局变量在同一个进程内不会反复创建。实测提升不明显但还是建议提出来。4.3 分区、缓存与写出 Parquet生成大数据时最容易踩的是分区数配置不当。默认spark.sql.shuffle.partitions是 200生成上亿行数据后写 Parquet会产生大量小文件下游读取慢且 NameNode 压力大。我一般会把生成阶段的分区数和最终输出的目标文件数结合考虑假设要输出 64 个 Parquet 文件就可以repartition(64)后再写既保证每个文件不至于太小又能充分并行。缓存的使用要克制。cache()适合同一份数据在后续被多个操作反复引用的情况。比如先构造好用户维表缓存后再分别去生成订单明细、退款单、物流信息这样能避免重复执行整条生成链路。但如果只是一次性生成后马上写文件缓存反而占用内存没必要。用完记得spark.catalog.clearCache()释放内存。写出 Parquet 时如果后续查询经常按时间过滤可以按order_year或order_month分区。分区键的基数要低否则会产生大量小分区效果适得其反。order_df.write.mode(overwrite) \ .partitionBy(order_year) \ .parquet(/tmp/synthetic_orders)5. 常见问题与排查实录5.1 UDF 返回 null 或序列化错误这是 Faker 类 UDF 最容易碰到的问题。常见场景是明明 Python 函数返回了字符串Spark 的最终结果里却有大量 null。原因多半是 Faker 在某些语言环境下返回了 None或者 pandas UDF 返回的 Series 长度和输入不一致。排查思路是先单独跑一小批数据打印出 pandas UDF 的输入输出确认长度一致、元素类型正确。修复手段是在 UDF 内部加兜底逻辑比如local_fake.name() or 未知避免 None 进入 Spark 端。序列化错误的典型报错是PickleException或ArrowTypeError。这通常是 pandas UDF 返回的 Series 类型与声明的 Spark 类型不匹配。比如声明了string但 Faker 返回的是 datetime 对象就会报错。解决方法是先统一str()转换再返回 Series。5.2 随机数多列不匹配问题生成多个字段时如果每个字段各自调用rand()并且没有显式种子会有两个后果一是结果不可复现每一次运行的数据都不一样二是字段之间的随机值完全独立看不出业务关系。这在联调阶段很致命——你改了代码后下游看到的测试数据完全变了没法对比。我的应对策略是每个随机维度显式使用不同种子比如rand(seed42)、rand(seed43)、rand(seed44)。如果确实需要多个字段关联就把种子相同的随机列先赋值给一个新列再从新列派生子字段。这样即使后续代码调整只要种子不变同一行的关联关系就不变。5.3 数据倾斜与 OOM生成数据时如果用了广播 join 或 cross join很容易在某些 Executor 上堆积过多数据导致 OOM。一个防护办法是不要做无条件的笛卡尔积而是先给数据打一个分桶键再 join。比如生成 1 亿行时可以先生成一个bucket_id两个 DataFrame 都按照 bucket_id 关联让 Spark 自然处理局部关联避免全网广播。另一个常见问题是内存不足尤其是用了cache()之后。排查时先用spark.sql(SET spark.sql.adaptive.enabledtrue)开启动态调整再看 Executor 内存和堆外内存配置。生成量级较大时把单个 Executor 的内存调大但并发度调小往往比盲目提高并行度更有效。5.4 时区与编码的坑时间字段生成后写 Parquet 再读回来时间戳可能会偏移 8 小时。原因是 Spark 默认使用 JVM 时区或spark.sql.session.timeZone的配置。我习惯在建会话时统一设置spark.conf.set(spark.sql.session.timeZone, Asia/Shanghai)否则后续在 SQL 里做to_date、date_trunc等操作都会出现“我觉得没问题但结果就是差 8 小时”的诡异问题。编码问题主要出现在中文城市和商品名称上。如果 Python 源文件不是 utf-8 编码或者集群默认编码不是 utf-8中文文本写入 Parquet 后再读回会有乱码。最稳妥的做法是代码里保留英文内部标识例如 city_code 用 110000、120000然后通过字典表关联出中文名称。这既避免了编码问题也让字段结构更像生产环境。最后分享一个我实际项目里养成的习惯生成合成数据的过程本身也应该写成一套带参数的脚本而不是一次性代码。把行数、种子、字段分布比例、空值率都做成参数这样每次生成不同口径的数据只需要改配置不需要改逻辑。数据管道联调时这种“参数化造数”能力能省下大量时间也减少了很多来回沟通的成本。
