3-1. メダリオンアーキテクチャとデータクリーニング(ブロンズ→シルバー)

💡 具体例:null 処理の3パターン

from pyspark.sql import functions as F

df = spark.read.table("bronze.orders_raw")

# (1) キー列が null の行は削除(subset で対象列を限定するのが実務の基本)
df = df.dropna(subset=["order_id", "customer_id"])

# (2) 固定値で埋める:数量の null は 0、地域の null は "unknown"
df = df.fillna({"quantity": 0, "region": "unknown"})

# (3) coalesce:配送先住所がなければ請求先住所で補完
df = df.withColumn(
    "ship_address",
    F.coalesce(F.col("ship_address"), F.col("billing_address"))
)

-- SQL で同じことを行う場合
SELECT
  order_id,
  customer_id,
  COALESCE(quantity, 0)                      AS quantity,
  COALESCE(region, 'unknown')                AS region,
  COALESCE(ship_address, billing_address)    AS ship_address
FROM bronze.orders_raw
WHERE order_id IS NOT NULL
  AND customer_id IS NOT NULL;

df.dropna() を引数なしで呼ぶと「いずれかの列が null の行をすべて削除」になり、意図せず大量の行が消えることがあります。subset で対象列を明示するのが安全です。

💡 具体例:型変換の基本セット

from pyspark.sql import functions as F

df = (df
    # 文字列 → 数値(try_cast:変換できない値は null になる)
    .withColumn("amount", F.col("amount").try_cast("double"))
    .withColumn("quantity", F.col("quantity").try_cast("int"))
    # 文字列 → 日付/タイムスタンプ(フォーマットを明示)
    .withColumn("order_date", F.to_date(F.col("order_date"), "yyyy-MM-dd"))
    .withColumn("updated_at", F.to_timestamp(F.col("updated_at"),
                                             "yyyy-MM-dd HH:mm:ss"))
)

-- SQL 版:TRY_CAST と to_date / to_timestamp
SELECT
  TRY_CAST(amount AS DOUBLE)                              AS amount,
  TRY_CAST(quantity AS INT)                               AS quantity,
  to_date(order_date, 'yyyy-MM-dd')                       AS order_date,
  to_timestamp(updated_at, 'yyyy-MM-dd HH:mm:ss')         AS updated_at
FROM bronze.orders_raw;

Databricks Runtime 17.0 以降では ANSI モードが既定で有効になっており、cast は変換できない値(例:"abc" を数値へ)に対してエラー(CAST_INVALID_INPUT)で失敗します。不正値を null に落として処理を続けたい場合は try_cast(SQL では TRY_CAST(amount AS DOUBLE)、PySpark では F.col("amount").try_cast("double") や F.expr("try_cast(amount AS double)"))を使います。その場合も「try_cast 後に null が増えていないか」を確認する習慣が品質を守ります(検証クエリのパターンは 3-6 で扱います)。

💡 具体例:MERGE による冪等なシルバー更新

-- 変換結果を一時ビューとして登録(PySpark 側)
-- df_clean.createOrReplaceTempView("orders_clean")

MERGE INTO silver.orders AS t
USING orders_clean AS s
  ON t.order_id = s.order_id
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *;

# PySpark(Delta API)で同じ MERGE を書く場合
from delta.tables import DeltaTable

silver = DeltaTable.forName(spark, "silver.orders")
(silver.alias("t")
    .merge(df_clean.alias("s"), "t.order_id = s.order_id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

キー(ここでは order_id)が一致する行は最新の内容で更新され、存在しない行だけが挿入されます。同じ入力で何度実行してもテーブルの最終状態は変わらないため、リトライやバックフィル(過去分の再処理)に安全に対応できます。