做数据开发这几年我发现自己最常干的工作一半是写业务逻辑一半是“造数”。给下游团队演示报表要一批数据给算法模型跑通流程要一批数据给新集群做性能压测还要一批数据。真实数据牵扯内部权限和隐私管控不能随手导出能用的测试样本又往往太小几百万行的分布式任务根本撑不起来。这种时候最省事的办法就是自己在 PySpark 里生成合成描述性数据。所谓合成描述性数据就是看起来像业务数据的虚构数据有订单编号、有商品分类、有金额、有日期甚至还有一段能读的商品描述。它是有业务语义、有取值范围、有分布倾向的“假数据”不是那种拿rand()拍出来的裸随机数。这篇文章把我反复在各种模拟项目里验证过的生成思路、完整代码和踩坑经验一次性整理出来适合数据工程师、算法工程师以及每一位需要给 PySpark 作业准备大规模测试数据的开发者。1. 为什么要在 PySpark 里造数据三个真实场景1.1 数据权限受限时的替身数据数据开发者应该都经历过这种尴尬数仓链路搭好了报表模板画好了却找不到一份可以公开演示的样本数据。真实业务表里的用户标识、地址、交易流水都属于敏感信息任何一次不经意的查询和拷贝都可能踩到数据治理红线。在这种背景下合成数据几乎是唯一合理的替身方案。合成数据和“脱敏数据”不是一回事很多人容易混。脱敏是把真实数据做变换比如把姓名替换成随机别名、把手机号打码本质上仍然保留着原始数据的分布和部分关联关系。合成数据则是从零开始生成一批虚构记录只是“长得像”业务数据里面没有任何真实个体的痕迹。我的建议是如果公司允许脱敏优先用脱敏数据如果权限拿不到样本或者样本量太小撑不起测试再转向合成生成。我在实际项目里因为一开始没区分清楚被内部审计同事约谈过一次后来就养成了习惯任何造数脚本都会在文件头写明“数据来源全合成无真实个人信息”。给下游团队演示只是其中一种场景。更多时候合成数据用于打通开发环境和预发环境的联调开发环境里没有真实数据权限但代码逻辑又必须完整跑一遍。这时候造一份和真实 schema 完全一致、数据量可大可小的合成表开发进度才不会被卡住。1.2 压测与性能基线第二种高频场景是性能压测。你写了一个复杂的多表 join想知道它在 5000 万行规模下到底跑多久真实业务表根本不敢随便跑全量复制一份也要占用大量存储。合成数据的优势在这里非常明显生成规模可以精确控制想要 1000 万行就传一个参数想要 1 亿行也就是改一个数字的事。更重要的是压测需要极端场景。真实数据里很难找到“某个分类占比 90%”的倾斜状况但合成数据可以轻易构造出来。只要调整类别权重就能制造数据倾斜用来验证你的 join 策略、分区策略和内存参数是否足够健壮。同样重复率、空值率、异常值比例都可以人为设定。一套可控的合成生成器本质上就是一个压测场景构造器。1.3 算法模型的冷启动数据第三种场景来自算法团队。模型训练 pipeline 从数据读取、特征工程到训练评估链路非常长。在真实数据权限还没申请下来或者样本集还在脱敏处理的时候算法同学往往需要一批结构一致的数据先把流程跑通。如果 pipeline 在第一步就因为 schema 不匹配或者类型问题报错后面所有的调试都是在浪费时间。我遇到过好几次这种情况模型代码写好了特征工程也写好了就是没有一份合适的训练样本。最后我用 PySpark 生成了一份包含用户特征、行为标签、目标列的数据集字段类型和数量都与线上表完全一致。算法同学拿这份数据把整个训练流程跑通了后来换上真实数据代码几乎不用改。这里的关键在于合成数据的 schema 必须与真实表严格对齐类型要一致列名要一致甚至数值范围都要贴近业务逻辑否则 pipeline 试跑没有意义。2. 核心设计原则与选型动手前先想清楚2.1 合成数据不等于随机数新手最容易犯的错误是觉得造数就是给每个字段塞一个random()。实际上业务数据有它内在的逻辑约束。订单金额不可能出现负数也不会动辄几百万订单日期一定落在某个业务周期内商品分类虽然多样但大类数量是有限的。如果你的合成数据是“裸随机数”下游做数据质量校验时瞬间就会被打回。合成数据至少要满足三个层次的要求。第一是取值范围金额要在业务可解释的区间内年龄要在合法区间内日期要在项目周期内。第二是分布形态真实订单金额往往集中在某一区间长尾部分也有少量大额订单用正态分布近似效果会比均匀分布好。第三是列间弱关联比如高金额订单的购买数量通常不会太大某些分类的复购率天然更高。这些弱关联不需要非常精确但至少要能“骗过”肉眼观察和基础统计校验。2.2 三个设计原则规模可调、分布可控、结果可复现我建议任何造数项目都遵循这三个原则缺一个后面都会难受。规模可调是最基础的。用一个统一的row_count参数控制总行数而不是在代码里硬编码。测试时用 1 万行验证逻辑压测时改成 1000 万行同一个脚本直接复用。如果造数脚本本身还要花时间修改那它就没有价值了。分布可控意味着每个字段的生成逻辑要可配置。类别的权重、金额的区间、日期的跨度最好都通过参数或配置表指定。我习惯在脚本开头集中定义这些参数后面所有列都引用它们。结果可复现是最容易被忽视的。造数有两种“复现”需求第一种是固定随机种子让同一份代码每次跑出来结果一致方便追踪问题第二种是无论执行计划怎么变同一行数据永远生成同样的值这需要用行号而不是随机种子来驱动生成。第二种方式在第四章会详细解释这里先记住一个结论在分布式环境里单纯靠rand(seed)并不可靠靠确定性函数如md5(row_id)才真正可靠。2.3 为什么选 PySpark 而不是单机脚本你可能会问造数据用 Python 加random库甚至用现成的 Faker 库不就行了吗当然行但有一个分界线——数据量。数据规模Pandas / 单机脚本PySpark1 万行很快代码简单杀鸡用牛刀启动都要几秒100 万行可以但内存吃紧轻松分布式生成5000 万行大概率 OOM轻松只要集群资源够5 亿行基本不可能可以受限于集群规模核心区别在于数据是否放得进单机内存。Pandas 在构造 DataFrame 时会把整个数据加载到内存中5000 万行几十个字段内存轻松超过几十 GB单机脚本基本没戏。PySpark 通过spark.range(n)生成行号时数据天然分布在多个分区每个 worker 只负责自己那部分计算内存压力被摊开了。还有一类折中方案是把 Faker 这类单机库包进 UDF 里在 Spark 的每个 worker 上分布式调用。这种做法适合生成地址、姓名这类需要现成语料的内容但性能不如纯 SQL 表达式后面我会对比这两种路线的取舍。2.4 随机函数的选择rand、randn、hashPySpark 里常用的随机源主要有三类它们的用途完全不同。rand(seed)生成 [0, 1) 之间的均匀分布随机数适合做类别抽样、日期偏移、编号计算。randn(seed)生成标准正态分布随机数适合模拟金额、评分这类在均值附近集中的数据但它的输出没有上下界可能出现极端值之后通常要加裁剪逻辑。md5()、sha2()、hash()属于确定性哈希函数输入相同则输出永远相同适合生成业务编码或者行级的稳定伪随机值。这里有一个新手很容易踩的点rand()表达式每次被调用都会独立采样。如果你在一个when条件里写三个rand()它们各自返回不同的随机值导致条件判断彼此矛盾。正确的做法是先用一个withColumn把随机值固定下来存成一列后续逻辑全部引用这一列。记住这条规则后面很多奇奇怪怪的问题都能避免。3. 实操从零生成一张完整业务表3.1 地基用 spark.range 分布式铺开行号先准备好 SparkSession 和几个会用到的函数from pyspark.sql import SparkSession from pyspark.sql.functions import ( col, rand, randn, lit, date_add, to_date, expr, when, concat, md5, substring, round ) spark SparkSession.builder.appName(synthetic_desc_data).getOrCreate() row_count 1_000_000 base spark.range(row_count).withColumnRenamed(id, row_id)spark.range(n)返回一个只有一列id的表从 0 到 n-1。这个 Range 节点在 Spark 里是分布式生成的不会在 driver 端构造数据也不会产生 shuffle所以生成几亿行也很轻量。把id改名为row_id是为了后面做确定性随机值得时候有一个稳定的输入。小提示如果你后续会反复复用这张基础表建议先执行一次base.cache().count()把数据物化到内存里避免每次withColumn都去重新扫描 Range 节点。但数据量特别大时cache 也可能挤占执行内存可以先不 cache观察执行计划再决定。3.2 数值列金额、数量、评分数值列是最直观的。以订单表为例价格、数量、评分是三个常见字段df base.withColumn(price, round(rand(11) * 8900 100, 2)) df df.withColumn(quantity, (rand(12) * 20 1).cast(int)) df df.withColumn(score, (randn(13) * 10 75).cast(int)) df df.withColumn( score_clipped, when(col(score) 60, 60).when(col(score) 100, 100).otherwise(col(score)) )price用rand(11) * 8900 100生成 100 到 9000 之间的金额round(..., 2)把精度限制到小数点后两位。这个细节很重要如果表结构里金额是DECIMAL(10,2)你生成一个带五位数小数的浮点数写入时轻则报错重则精度截断导致下游数据对不上。score用randn(13)生成标准正态分布乘 10 加 75 后整体落在 75 分附近。正态分布没有自然边界所以用when把结果裁剪到 60 到 100 分之间。这个裁剪操作就是 2.2 里说的“分布可控”——你要让假数据看起来像真实评分就不能出现 135 分这种业务上不可能的数值。如果你想要金额也呈现“多数集中、少数大额”的长尾形态可以视情况用randn()加指数变换或者简单地把大额区间概率调低。作为演示均匀分布乘以范围已经够用真正要贴近业务分布时再引入权重也不迟。3.3 类别列等概率抽样和带权重抽样类别字段是所有描述性数据的骨架。生成类别列有两种典型写法分别对应等概率和带权重的需求。等概率写法可以完全不用 UDF用 SQL 表达式的element_at加数组字面量就能实现df df.withColumn( category, expr(element_at(array(电子产品,家居用品,办公设备,运动户外,其他), cast(rand()*5 as int) 1)) )element_at(array, index)在 Spark SQL 里是按 1 开始的索引访问数组元素。rand()*5得到 0 到 5 之间的数转成整数后是 0 到 4加 1 变成 1 到 5正好对应数组里五个类别。每次调用rand()都是独立抽样所以分类大致均匀分布。带权重的写法需要先把随机数固定到一列再按累积区间切分df df.withColumn(cat_bucket, rand(20) * 100) df df.withColumn( category, when(col(cat_bucket) 15, 电子产品) .when(col(cat_bucket) 35, 家居用品) .when(col(cat_bucket) 55, 办公设备) .when(col(cat_bucket) 85, 运动户外) .otherwise(其他) )这段代码的意思是15% 的概率落在“电子产品”接下来 20% 落在“家居用品”依此类推。累积区间加起来要等于 100最后一个otherwise兜底。关键点是必须先把rand(20) * 100存成cat_bucket列再用它做多次when判断。如果每个when里都直接写rand() * 100你等于给同一行做了好几次独立的随机抽样几个分支可能同时满足也可能一个都不满足生成结果完全不可控。3.4 日期时间列固定基线加随机偏移日期字段的生成逻辑很直白选定一个业务起始日在它上面加一个随机天数或者随机秒数。df df.withColumn( order_date, date_add(to_date(lit(2024-01-01)), (rand(30) * 730).cast(int)) ) df df.withColumn( order_ts, expr(cast(from_unixtime(unix_timestamp(2024-01-01 00:00:00) cast(rand(31) * 31536000 as int)) as timestamp)) )第一段代码生成 2024 年 1 月 1 日起 730 天内的日期rand(30) * 730转换成整数就是随机的偏移天数。date_add接受一个日期列或日期字面量所以先用to_date把字符串转成日期类型。第二段代码生成的是带时分秒的时间戳。思路是把起始时间转换成 Unix 秒数加上 0 到 31536000 秒正好一年的随机偏移再用from_unixtime转回时间格式。这样生成的时间戳完全落在 2024 年这一年里含义清晰且类型正确。如果你要模拟“近期活跃”的冷热分布可以把均匀分布的随机数做一次变换让更多数据集中在最近 30 天。最简单的做法就是用when加权重70% 的概率落在最近 30 天30% 落在更早的时间区间。分布可控的价值就在这里体现。3.5 ID 与编号列单调自增与伪随机编码业务表一般都有主键或业务单号。这里有两个常用套路df df.withColumn( order_no, expr(concat(SO, lpad(cast(row_id % 100000000 as string), 8, 0))) ) df df.withColumn( sku_code, substring(md5(concat(col(row_id).cast(string), lit(|salt2024))), 1, 12) )order_no用row_id % 100000000限制编号在 8 位以内再用lpad前补零拼上SO前缀得到类似SO00001234的单号。这个做法可以让你从订单号直接反推出它在全表里的位置调试和抽查都非常方便。sku_code走的是另一条路把row_id拼上一个固定的盐值再用md5哈希取前 12 位作为商品编码。md5是确定性函数同一行无论跑多少遍、无论分区数怎么变生成的sku_code永远一样。这个方法同时具备两个优点看起来随机实际上完全可复现。它比rand()更适合作为业务编码的生成器因为业务编码通常要求稳定。3.6 描述性文本列两种路线实现商品描述这是标题里“描述性数据”的核心。我遇到最多的问题就是如何在大规模分布式环境里生成一段“像人写的”商品描述。这里给出两种路线按需选择。路线一纯 SQL 表达式。不要 UDF直接拼接各词语池df df.withColumn( description, expr( concat( 【, element_at(array(睿景,蓝诺,科凌,迈克,扬帆), cast(rand()*5 as int) 1), 】, element_at(array(铝合金,ABS,不锈钢,硅胶,原木), cast(rand()*5 as int) 1), 材质, element_at(array(轻便易携带,防磕碰设计,低功耗长续航,多场景通用,免工具安装), cast(rand()*5 as int) 1), 适合, element_at(array(办公室,居家,户外,差旅,校园), cast(rand()*5 as int) 1), 场景使用。 ) ) )这个写法本质和 3.3 里的element_at一样只是把多个独立抽样的词语用concat拼起来。每个rand()单独抽样所以品牌、材质、卖点、场景的组合是随机的理论上能组合出几百种不同的描述。它的优势在于全程 JVM 内计算没有 Python 序列化和跨进程通信速度极快千万级数据也是秒级完成。路线二Python UDF。适合更复杂的模板或者需要在描述里体现列间关联的场景import random from pyspark.sql.functions import udf from pyspark.sql.types import StringType udf(returnTypeStringType()) def gen_desc(seed_int): random.seed(seed_int % (2**32)) brand random.choice([睿景, 蓝诺, 科凌, 迈克, 扬帆]) material random.choice([铝合金, ABS, 不锈钢, 硅胶, 原木]) feature random.choice([轻便易携带, 防磕碰设计, 低功耗长续航, 免工具安装, 多场景通用]) scene random.choice([办公室, 居家, 户外, 差旅, 校园]) return f{brand}推出的{material}材质新品主打{feature}专为{scene}使用场景打造。 df df.withColumn(description, gen_desc(col(row_id).cast(int)))UDF 第一个要注意的点random.seed()必须每行都不同。如果你在函数里写random.seed(42)同一个 worker 内的所有行都会用同一个种子去调 Python 的随机模块结果就是它们生成一模一样的产品描述。大数据量下这种“整段重复”很容易被下游发现所以我把row_id作为种子传入每行都有不同的初值。第二个注意点是性能。Python UDF 会把每个row_id从 JVM 序列化传到 Python 进程计算完再把结果传回来开销比 SQL 表达式大很多。在我的实测中百万行以内还能接受千万行就要谨慎。如果你只是做简单词语拼接优先选路线一只有当你需要中文语料、复杂条件分支、或者调用外部文本模板时才选择 UDF。顺便提一句如果你在 UDF 里封装 Faker 这类库来生成姓名、地址、公司名思路是可行的但要注意线程安全和重复率。合成数据本身不要求全库唯一但如果下游需要唯一键记得把row_id拼进去做兜底。3.7 组装完整业务表并落盘前面每一列都是基于同一张df不断叠加最后用一次select把需要的列挑出来写入存储final_df df.select( row_id, order_no, sku_code, category, price, quantity, score_clipped, order_date, order_ts, description ) final_df.write.mode(overwrite).parquet(hdfs:///tmp/synthetic_desc_data)写入格式我一般优先选 Parquet因为它自带 schema日期和时间戳列在读出时仍然是原始类型。如果下游明确要求 CSV可以这样写final_df.write.mode(overwrite).option(header, True).option(encoding, utf-8).csv(hdfs:///tmp/synthetic_desc_data_csv)但要注意CSV 没有 schema日期列和时间戳列读出时会变成字符串下游需要自己做类型转换。另外尽量把生成逻辑封装成一个带参数的函数行数、权重、日期范围都通过参数传入这样下次需求变化时不需要改代码只要改配置。这看起来很小但长期维护成本差别很大。4. 性能调优与常见问题排查实录4.1 Python UDF 到底慢在哪里很多人用 Spark 造数时都会顺手写 UDF写完发现速度慢得离谱。原因在于 PySpark 的 UDF 不是每行调用一次 Python 函数那么简单——每一行数据都要从 JVM 序列化给 Python 解释器处理完再序列化回来。这个来回的代价远高于在 JVM 内部直接执行一条 SQL 表达式。我在一个项目里对比过10 万行的商品描述生成用element_at加concat的 SQL 表达式整体耗时不到 1 秒改成 Python UDF 后耗时增长到 20 秒以上。差距随着数据量线性放大。所以我的经验法则是能写成 SQL 表达式表达的逻辑绝不写成 Python UDF只有逻辑复杂到 SQL 写不清楚比如需要调用外部语料库或做多分支语义组合再考虑 UDF。如果 UDF 不可避免可以尝试用 pandas UDF 或者mapInPandas这类向量化方案按批次而不是按行传输数据性能会有明显提升。4.2 并行度、cache 与内存的取舍spark.range生成数据的并行度由集群配置决定但后面一旦有重分区或 shuffle 类操作要留意目标分区数。生成超大表时如果默认并行度太低某些 worker 会长时间空转如果太高每行计算的开销会被放大。我一般会先用df df.repartition(分区数)显式控制并行度具体数值根据集群核心数来定比如 50 个 executor、每 executor 4 核就设在 200 左右。cache()和checkpoint()的选择也常被忽略。如果基础表只有几十万行cache 非常合适后续多次withColumn不会重复扫描。但如果是几亿行的大表cache 本身就会挤占大量执行内存反倒拖慢全流程。大表场景下我更推荐先用final_df.write.format(parquet).save(...)落一次盘然后从 Parquet 重新读入用磁盘换内存稳定。还有个细节不要为了看数据就调用df.collect()。几千万行数据全部拉到 driver 端内存瞬间爆炸。想看前几行用df.limit(10).collect()想看统计用df.describe()或者df.groupBy().count()这些都是分布式执行不会把全量数据拷到 driver。4.3 随机种子失效的真相这是我在实际工作中被问得最多的问题代码里明明写了rand(42)为什么换一台机器跑结果就变了要理解这个现象得知道 Spark 里rand(seed)的生成机制。每个分区在初始化随机流的时候使用的是“种子 分区序号”的组合。也就是说rand(42)生成的结果严重依赖执行计划里的分区结构。同样的代码在同一集群、同一分区数下每次跑结果一致但只要你调整了并行度、改了spark.sql.shuffle.partitions或者把任务提交到不同核数的集群同一行对应的随机值就完全变了。如果你只追求“同一环境下的结果一致”rand(seed)够用。但如果你希望生成结果与执行计划完全无关即“无论怎么重分区同一行永远产生同一个随机值”就必须绕开rand()用确定性函数从行号派生df df.withColumn(stable_rand, expr(pmod(hash(row_id), 100) / 100.0))pmod(hash(row_id), 100)对行号做哈希后取 0 到 99 的余数再除以 100 得到 [0, 1) 的稳定随机值。它不依赖分区顺序只依赖row_id本身所以无论执行计划怎么变化同一行永远落在同一个“随机”位置上。配合 3.5 里的md5(row_id)技巧整张表的可复现性就从“受环境影响的复现”升级成了“全局确定性复现”。另外把rand()写在when多分支里导致的重复抽样问题在前面 3.3 已经提到过。这两个坑几乎是每个新手必踩的组合拳后面再做造数脚本时务必先固定随机数再做条件判断再谈可复现。4.4 中文乱码与文件倾斜处理中文数据有两个容易踩的坑。第一个是编码。CSV 写出时没有显式指定编码某些环境下会默认输出为平台默认字符集读出来就是乱码。解决方式是写出时加上option(encoding, utf-8)读入时也同样指定。Parquet 格式自带编码信息基本不用操心这也是我推荐 Parquet 的原因之一。第二个是数据倾斜。如果你用category做partitionBy写文件而类别的生成权重又严重不均某一个分类的文件会特别大其他分类则是大量小文件。小文件在后续任务里会带来很大的调度开销。我的建议是除非你有明确的按分类分目录读取需求否则在合成表场景下不要用partitionBy直接一个目录写若干个等大的 Parquet 文件就可以。如果后续高频按某个字段做 join可以用bucketBy做分桶而不是目录分区。4.5 常见问题速查表现象常见原因解决思路随机种子相同但两次结果不同分区数量或执行计划改变改用pmod(hash(row_id), 100)等确定性函数UDF 生成的中文描述整段重复random.seed()使用了固定种子用row_id作为每行的随机种子同一次生成中两个分类条件同时失效或重叠when分支里直接调多次rand()先用一列固定随机值再基于该列做判断生成的金额精度对不上下游表没有限制小数位用round(x, 2)并贴合DECIMAL精度日期字段变成字符串写了 CSV 导致 schema 丢失优先 Parquet或读 CSV 后做显式 cast数据量一大就 OOMdriver 端collect()或单机脚本生成分布式生成以limit代替collect抽查写入后大量小文件对倾斜列做了partitionBy去掉目录分区改用bucketBy或直接并行写这张表是我近几次造数项目里问题的集中总结。看起来都是小问题但在分布式环境下每一个都可能让任务跑好几小时然后失败。最后再分享一个我自己的操作习惯不管生成多少行数据正式大批量写盘之前一定会先用row_count为 1000 跑一遍小样本df.show(10)看一下每一列的值是否符合预期再用df.groupBy(category).count().show()看一眼类别分布是否接近设定的权重。这一步只需要几秒钟但能拦住绝大多数低级错误。等小样本验证通过后再把row_count调到真实需要的规模同一套代码直接套用。这比在大表上反复调试要高效得多。合成数据毕竟是“假数据”尽量让它在被消费前就看起来足够真实才是这套方案能长期用下去的关键。