databricks-synthetic-data-gen

カタログとスキーマは必ずユーザーが指定します。デフォルト値は決して使用しないでください。ユーザーが指定していない場合は確認してください。UC への書き込みでは、データを書き込む前に必ずスキーマが存在することを確認し、存在しない場合は作成してください。

Databricks の合成データ生成

Spark + Faker + Pandas UDF（強く推奨）を使用して、Databricks 向けのリアルでストーリー性のある合成データを生成します。

データでビジネスストーリーを伝える

合成データでは、Databricks が現実のビジネス課題の解決にどう役立つかを示す必要があります。

パターン： 問題が発生する → ビジネスへの影響（$） → 根本原因を分析する → 影響を受けた顧客を特定する → 修正して再発を防ぐ。

主な原則：

• 問題 → 影響 → 分析 → 解決策 — 測定可能なビジネス上の影響を引き起こすインシデント、異常、または問題を含めます。データを使って根本原因を特定し、対処できるようにします。
• 業界に即しつつ、シンプルに — ドメイン用語（例：「SLA 違反」「解約」「在庫切れ」）を使いながら、スキーマは理解しやすく保ちます。テーブルは少数にし、関係性を明確にします。
• $ の影響を示すビジネスメトリクス — 売上、MRR、コスト、コンバージョン率などを含めます。すべてのストーリーで、その重要性を示すドル記号が必要です。
• テーブル同士が互いを説明する — チケットが急増した場合、インシデントテーブルで障害を示します。売上が減少した場合、解約テーブルで離脱した顧客とその理由を示します。すべてのデータを関連付けます。
• 実行可能なインサイト — データから次の問いに答えられるようにします。何が起きたのか？誰が影響を受けたのか？どれだけのコストがかかったのか？どうすれば再発を防げるのか？

一様分布を避ける理由： 一様なデータにはストーリーがありません。急増、異常、コホート、20/80 の偏り、歪みなど、調査すべきものが何もありません。根本原因分析における Databricks の価値を示せません。

参考資料

タイミング	ガイド
ユーザーがML モデルのトレーニングや複雑な時間パターンに言及した場合	references/1-data-patterns.md — ML 対応データ、時間倍率、行の一貫性
生成中にエラーが発生した場合	references/2-troubleshooting.md — よくある問題の解決方法

重要なルール

1. データでストーリーを伝える — 問題が発生し、金銭的な影響があり、分析して修正できるようにします。Databricks の価値を示します。
2. すべてのデータがストーリーに役立つ — すべてのテーブルと列に一貫性があり、ダッシュボードや ML モデルで利用できる必要があります。孤立したデータやランダムなノイズは含めません。将来のダッシュボードで説明や可視化に役立たず、予測にも使えないデータは生成しないでください。
3. 業界用語を使い、スキーマはシンプルに — ドメイン固有の語彙を使いつつ、理解しやすく保ちます（テーブルは少数にし、関係性を明確にします）。
4. 一様分布は使わない — 偏りのあるカテゴリ、対数正規分布の金額、80/20 パターンを使います。一様な分布 = ストーリーなし = 役に立たないデータです。
5. 傾向を捉えるのに十分なデータ量を確保する — パターンが集計後も残るよう、主要なテーブルには約 10 万行以上を用意します。
6. カタログとスキーマを確認する — デフォルト値は使わず、生成前に必ず確認します。
7. 承認を得るために計画を提示する — コードを書く前に、テーブル、分布、前提条件を提示します。
8. マスターテーブルを先に作成する — 親テーブルを生成して Delta に書き込み、その後、有効な外部キーを持つ子テーブルを作成します。
9. Spark + Faker + Pandas UDF を使う — スケーラブルな並列処理が可能です。ユーザーがローカル実行かつ 3 万行未満を明示的に希望する場合に限り、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：要件を収集する

ユーザーに次の点を確認します。

• カタログ／スキーマ — 使用するカタログはどれですか？
• ドメイン — Eコマース、サポートチケット、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	優先度と resolution_time を含むサポートチケット	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 万ドルの損失。原因は 1 件のインシデントまで追跡できます
• 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 以上のパーティション

出力形式：

• Volume への Parquet 出力（デフォルト）：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 変換 → Faker 用の pandas_udf → 直接書き込み

一般的なパターン

重み付けカテゴリ（一様にしない）

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")

参照整合性（FK パターン）

マスターテーブルを先に Delta に書き込み、その後読み戻して FK 結合に使用します（サーバーレスでは .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 に書き込み、読み戻して FK 結合に使用します
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 を参照してください。