databricks-synthetic-data-gen

目录和架构始终由用户提供，绝不能使用默认值。如果用户尚未提供，请询问。对于任何 UC 写入操作，始终先创建不存在的架构，再写入数据。

Databricks 合成数据生成

使用 Spark + Faker + Pandas UDFs（强烈推荐）为 Databricks 生成逼真且具有故事性的合成数据。

数据必须讲述业务故事

合成数据应展示 Databricks 如何帮助解决真实的业务问题。

模式： 出现问题 → 产生业务影响（$）→ 分析根本原因 → 找出受影响的客户 → 修复并预防。

关键原则：

• 问题 → 影响 → 分析 → 解决方案 — 包含会造成可衡量业务影响的事故、异常或问题。数据应能帮助你找到根本原因并采取行动。
• 贴合行业但保持简单 — 使用领域术语（例如“SLA 违约”“客户流失”“缺货”），同时保持架构易于理解。表格数量少，关系清晰。
• 体现 $ 影响的业务指标 — 收入、MRR、成本、转化率。每个故事都需要有美元金额，说明其重要性。
• 表之间相互解释 — 工单激增？事件表应显示故障。收入下降？客户流失表应显示哪些客户离开以及原因。所有数据都相互关联。
• 可执行的见解 — 数据应能回答：发生了什么？谁受到了影响？造成了多少损失？如何预防？

为什么不能使用平坦分布： 均匀分布的数据没有故事——没有峰值、异常、群组、二八分布、偏态，也没有任何值得调查的内容。它无法体现 Databricks 在根因分析方面的价值。

参考资料

时机	指南
用户提到机器学习模型训练或复杂的时间模式	references/1-data-patterns.md — 适用于机器学习的数据、时间乘数、行一致性
生成期间出现错误	references/2-troubleshooting.md — 修复常见问题

关键规则

1. 数据要讲述一个故事 — 出现问题，造成 $ 影响，并且可以分析和修复。展现 Databricks 的价值。
2. 所有数据都要服务于故事 — 每个表和列都必须连贯，并且可用于仪表板或机器学习模型。不要有孤立数据或随机噪声——如果某项数据无法帮助解释、绘制未来仪表板或进行预测，就不要生成。
3. 使用行业术语，保持架构简单 — 使用领域专用词汇，同时确保易于理解（表格数量少，关系清晰）。
4. 绝不使用均匀分布 — 类别应有偏态，金额应呈对数正态分布，并体现 80/20 模式。平坦分布 = 没有故事 = 毫无用处。
5. 数据量要足以呈现趋势 — 主要表格约需 100K+ 行，让聚合后仍能看出模式。
6. 询问目录/架构 — 绝不使用默认值；生成前始终确认。
7. 先提交计划供用户批准 — 编写代码前展示表格、分布和假设。
8. 先生成主表 — 生成父表并写入 Delta，然后创建具有有效 FK 的子表。
9. 使用 Spark + Faker + Pandas UDFs — 可扩展且支持并行。仅当用户明确要求在本地运行且数据少于 30K 行时，才使用 Polars。
10. 默认使用 Databricks Connect Serverless 生成数据 — 如有需要，在 python 3.12 上更新 databricks-connect（除非另有指示不要使用 Databricks Connect，否则避免使用 execute_code）。
11. 不要使用 .cache() 或 .persist() — Serverless 不支持。写入 Delta，再读回以进行联接。
12. 不要使用 Python 循环或 .collect() — 使用 Spark 并行处理。不要在驱动端迭代，并避免 Pandas 与 Spark 之间的转换。

生成规划工作流

生成任何代码之前，必须先提交计划供用户批准。

⚠️ 必须执行：继续之前确认目录

必须明确询问用户要使用哪个目录。 不要自行假设，也不要在未确认的情况下继续。

向用户提问的示例：

“要将此数据写入哪个 Unity Catalog？”

提交计划时，务必醒目地显示所选目录：

📍 Output Location: catalog_name.schema_name
   Volume: /Volumes/catalog_name/schema_name/raw_data/

这样用户可以轻松发现并纠正目录（如有需要）。

第 1 步：收集需求

询问用户：

• 目录/架构——要使用哪个目录？
• 领域——电子商务、支持工单、IoT、金融？（使用行业术语）

**如果用户没有指定故事：**提出一个故事。不要生成平淡无奇的数据——建议一个能体现 Databricks 价值的事件、异常或趋势（例如：“我会加入一次系统中断，导致工单激增和客户流失——这样你就能演示根因分析。”）

第 2 步：展示包含故事的计划

清晰地展示规格说明，并明确说明业务故事和你的假设：

📍 Output Location: {user_catalog}.support_demo
   Volume: /Volumes/{user_catalog}/support_demo/raw_data/

📖 Story: A payment system outage causes support ticket spike. Resolution times
   degrade, enterprise customers churn, revenue drops $2.3M. With Databricks we
   identify the root cause, affected customers, and prevent future impact.

表	描述	行数	关键假设
customers	包含层级和 MRR 的客户资料	10,000	企业客户占 10%，但贡献 60% 的收入
tickets	包含优先级和解决时间的支持工单	80,000	系统中断期间工单激增，SLA 违约
incidents	系统事件（中断、部署）	50	月中发生支付中断
churn_events	包含流失原因的客户取消记录	500	糟糕的支持体验后出现流失高峰

业务指标：

• customers.mrr——风险收入（$）
• tickets.resolution_hours——SLA 表现
• churn_events.lost_mrr——流失影响（$）

这些数据讲述的故事：

• incidents 表显示 3 月 15 日发生支付中断
• 中断期间工单量激增至 5 倍，解决时间从 4 小时延长至 18 小时
• SLA 违约的企业客户在 3 周后流失
• 总影响：流失的 MRR 达 230 万美元，可追溯至单次事件
• **Databricks 的价值：**根因分析、识别有流失风险的客户、构建告警

询问用户：“这个故事合适吗？需要做任何调整吗？”

第 3 步：询问数据特性

• 偏斜（非均匀分布）——默认启用
• 联接（参照完整性）——默认启用
• 注入不良数据（用于数据质量测试）
• 多语言文本
• 增量模式（追加而非覆盖）

生成前检查清单

• 已确认目录——用户已明确批准要使用的目录
• 计划中显著展示输出位置（方便查看/更改）
• 已展示并获批表规格
• 已确认关于分布的假设
• 用户已确认计算偏好（推荐使用无服务器上的 Databricks Connect）
• 已选择数据特性

在用户批准计划（包括目录）之前，不要开始生成代码。

生成后验证

使用 databricks experimental aitools tools query 验证生成的数据（行数、分布、参照完整性）。直接查询 parquet 文件：

databricks experimental aitools tools query --warehouse $WAREHOUSE_ID "
SELECT COUNT(*) FROM parquet.\`/Volumes/CATALOG/SCHEMA/raw_data/customers\`
"

完整的验证示例请参阅 references/2-troubleshooting.md。

使用 Databricks Connect Spark + Faker 模式

from databricks.connect import DatabricksSession
from pyspark.sql import functions as F
from pyspark.sql.types import StringType
import pandas as pd

# Setup serverless Spark session
spark = DatabricksSession.builder.serverless(True).getOrCreate()

# Pandas UDF pattern - import lib INSIDE the function (libs must be installed locally)
@F.pandas_udf(StringType())
def fake_name(ids: pd.Series) -> pd.Series:
    from faker import Faker  # Import inside UDF
    fake = Faker()
    return pd.Series([fake.name() for _ in range(len(ids))])

# Generate with spark.range, apply UDFs
customers_df = spark.range(0, 10000, numPartitions=16).select(
    F.concat(F.lit("CUST-"), F.lpad(F.col("id").cast("string"), 5, "0")).alias("customer_id"),
    fake_name(F.col("id")).alias("name"),
)

# Write to Volume as Parquet (default for raw data)
# Path is a folder with table name: /Volumes/catalog/schema/raw_data/customers/
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {CATALOG}.{SCHEMA}")
spark.sql(f"CREATE VOLUME IF NOT EXISTS {CATALOG}.{SCHEMA}.raw_data")
customers_df.write.mode("overwrite").parquet(f"/Volumes/{CATALOG}/{SCHEMA}/raw_data/customers")

按规模划分分区： spark.range(N, numPartitions=P)

• 少于 100K 行：8 个分区
• 100K–500K 行：16 个分区
• 500K–1M 行：32 个分区
• 1M 行及以上：64 个或更多分区

输出格式：

• Parquet 到 Volume（默认）：df.write.parquet("/Volumes/.../raw_data/table")——用于管道的原始数据
• Delta 表：df.write.saveAsTable("catalog.schema.table")——如果用户需要可查询的表
• JSON/CSV：小型维度表，用于复刻旧系统

性能规则

生成的脚本必须具有高性能。绝不要采用以下做法：

反模式	慢的原因	应改用
在驱动程序上使用 Python 循环	单线程运行，无法并行	使用 spark.range() + Spark 操作
.collect() 后再迭代	将所有数据加载到驱动程序内存	将数据保留在 Spark 中，使用 DataFrame 操作
Pandas → Spark → Pandas	序列化开销，无法发挥分布式处理的优势	保持使用 Spark，仅在 UDF 中使用 pandas_udf
读取/写入临时文件	不必要的 I/O	链接 DataFrame 转换
标量 UDF	逐行处理	使用 pandas_udf 批量处理

良好模式： spark.range() → Spark 转换 → 使用 pandas_udf 生成 Faker 数据 → 直接写入

常见模式

加权类别（切勿使用均匀分布）

F.when(F.rand() < 0.6, "Free").when(F.rand() < 0.9, "Pro").otherwise("Enterprise")

对数正态金额（在 pandas UDF 中）

使用 np.random.lognormal(mean, sigma)——结果始终为正，且有长尾：

• Enterprise：lognormal(7.5, 0.8) → 中位数约为 1800 美元
• Pro：lognormal(5.5, 0.7) → 中位数约为 245 美元
• Free：lognormal(4.0, 0.6) → 中位数约为 55 美元

日期范围（最近 6 个月）

END_DATE = datetime.now()
START_DATE = END_DATE - timedelta(days=180)

基础设施（始终在脚本中创建）

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {CATALOG}.{SCHEMA}")
spark.sql(f"CREATE VOLUME IF NOT EXISTS {CATALOG}.{SCHEMA}.raw_data")

参照完整性（外键模式）

先将主表写入 Delta，再读回以进行外键联接（无服务器环境中不要使用 .cache()）：

# 1. Write master table
customers_df.write.mode("overwrite").saveAsTable(f"{CATALOG}.{SCHEMA}.customers")

# 2. Read back for FK lookup
customer_lookup = spark.table(f"{CATALOG}.{SCHEMA}.customers").select("customer_idx", "customer_id")

# 3. Generate child table with valid FKs via join
orders_df = spark.range(N_ORDERS).select(
    (F.abs(F.hash(F.col("id"))) % N_CUSTOMERS).alias("customer_idx")
)
orders_with_fk = orders_df.join(customer_lookup, on="customer_idx")

设置

需要 Python 3.12 和 databricks-connect>=16.4。使用 uv：

uv pip install "databricks-connect>=16.4,<17.4" faker numpy pandas holidays

相关技能

• databricks-unity-catalog — 管理目录、架构和卷
• databricks-dabs — 用于生产部署的 DAB

常见问题

问题	解决方案
ModuleNotFoundError: faker	在本地安装：uv pip install faker，并在 UDF 内导入
Faker UDF 运行缓慢	使用 pandas_udf 进行批处理
内存不足	增加 spark.range() 中的 numPartitions
参照完整性错误	先将主表写入 Delta，再读回以进行外键联接
PERSIST TABLE is not supported on serverless	在无服务器环境中切勿使用 .cache() 或 .persist()——先写入 Delta 表，再读回
F.window 与 Window 混淆	对 row_number()、rank() 等使用 from pyspark.sql.window import Window。F.window 仅用于流处理。
不支持广播变量	在无服务器环境中切勿使用 spark.sparkContext.broadcast()

完整的故障排除指南请参见 references/2-troubleshooting.md。