3-3. 列・行・テーブル構造の操作(追加・削除・分割・rename・filter・explode)

💡 具体例:列操作の基本セット

from pyspark.sql import functions as F

df = spark.read.table("silver.orders")

df = (df
    # 追加:金額 = 単価 × 数量
    .withColumn("total_amount", F.col("unit_price") * F.col("quantity"))
    # 追加:条件分岐で区分列を作る(SQL の CASE 相当)
    .withColumn("size_class",
        F.when(F.col("total_amount") >= 10000, "large")
         .when(F.col("total_amount") >= 1000, "medium")
         .otherwise("small"))
    # 改名
    .withColumnRenamed("cust_id", "customer_id")
    # 削除(複数列まとめて)
    .drop("temp_col", "legacy_flag")
)

# select は「選ぶ」と同時に式・改名もできる
df_slim = df.select(
    "order_id",
    F.col("customer_id"),
    (F.col("unit_price") * F.col("quantity")).alias("total_amount"),
)

-- SQL 版:CASE 式と別名で同じ形を作る
SELECT
  order_id,
  cust_id AS customer_id,
  unit_price * quantity AS total_amount,
  CASE
    WHEN unit_price * quantity >= 10000 THEN 'large'
    WHEN unit_price * quantity >= 1000  THEN 'medium'
    ELSE 'small'
  END AS size_class
FROM silver.orders;

PySpark の when().otherwise() は SQL の CASE WHEN に対応します。otherwise を省略すると、どの条件にも合わない行は null になる点に注意してください。

💡 具体例:氏名列を姓と名に分割する

from pyspark.sql import functions as F

# full_name = "Yamada Taro" のような列を分割
df = (df
    .withColumn("name_parts", F.split(F.col("full_name"), " "))
    .withColumn("last_name",  F.col("name_parts").getItem(0))
    .withColumn("first_name", F.col("name_parts").getItem(1))
    .drop("name_parts")
)

-- SQL 版:split と添字アクセス(SQL の配列添字は 0 始まり)
SELECT
  split(full_name, ' ')[0] AS last_name,
  split(full_name, ' ')[1] AS first_name
FROM silver.customers;

第2引数は正規表現として解釈されます。ピリオドやパイプ(. や |)で区切る場合は \\. のようにエスケープが必要です。

💡 具体例:注文内の商品配列を行に展開する

from pyspark.sql import functions as F

# orders: order_id | items(array<struct<sku:string, qty:int>>)
df_items = (orders
    .select("order_id", F.explode(F.col("items")).alias("item"))
    .select(
        "order_id",
        F.col("item.sku").alias("sku"),
        F.col("item.qty").alias("qty"),
    )
)

-- SQL 版:explode をそのまま SELECT に書ける
SELECT
  order_id,
  item.sku,
  item.qty
FROM orders,
  LATERAL explode(items) AS t(item);

-- より簡潔に、SELECT 句内で使う書き方も可能
SELECT order_id, explode(items) AS item FROM orders;

展開前が3注文で items がそれぞれ 2, 1, 3 要素なら、展開後は 6 行になります。行数が変わる操作なので、後続の集計(3-4)の粒度が「注文単位」から「商品明細単位」に変わることを意識してください。

💡 具体例:JSON 文字列列のパース

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# payload 列 = '{"device":"sensor-1","temp":23,"loc":{"city":"osaka"}}'
schema = StructType([
    StructField("device", StringType()),
    StructField("temp", IntegerType()),
    StructField("loc", StructType([StructField("city", StringType())])),
])

df = (raw
    .withColumn("data", F.from_json(F.col("payload"), schema))
    .select(
        F.col("data.device").alias("device"),
        F.col("data.temp").alias("temp"),
        F.col("data.loc.city").alias("city"),
    )
)

-- SQL 版:スキーマは DDL 文字列で指定する
SELECT
  parsed.device,
  parsed.temp,
  parsed.loc.city
FROM (
  SELECT from_json(payload,
    'device STRING, temp INT, loc STRUCT<city: STRING>') AS parsed
  FROM bronze.events
);

スキーマ指定に対して JSON 側にないフィールドは null になり、パースに失敗した行も null になります。なお SQL では、JSON 文字列に対して payload:device のようなコロン記法(JSON パス)で場当たり的に値を取り出す方法もありますが、シルバー層で型を確定させるなら from_json でスキーマを与えるのが基本です。