目次 › 第3章 › 3-2

第3章 データ変換とモデリング / 想定学習時間:30〜40分 / 最終確認:2026年8月

3-2. DataFrameの結合(inner/left/broadcast/cross join・union)

1. 結合の基本形:join の種類

シルバー層のデータは複数のテーブルに分かれています(例:注文 orders と顧客 customers)。これらを突き合わせて1つの DataFrame にするのが結合(join)です。PySpark では df1.join(df2, 結合条件, 結合タイプ) の形で書きます。

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

# inner join(既定):両方に存在するキーの行だけ残る
df = orders.join(customers, on="customer_id", how="inner")

# left join:左(orders)は全行残り、相手がなければ null で埋まる
df = orders.join(customers, on="customer_id", how="left")

-- SQL 版
SELECT o.*, c.customer_name, c.region
FROM silver.orders o
INNER JOIN silver.customers c
  ON o.customer_id = c.customer_id;

SELECT o.*, c.customer_name
FROM silver.orders o
LEFT JOIN silver.customers c
  ON o.customer_id = c.customer_id;

# 複数キー結合:リストで渡す
df = orders.join(returns, on=["order_id", "line_no"], how="left")

# 列名が左右で異なる場合は条件式で書く
df = orders.join(
    customers,
    orders["customer_id"] == customers["cust_id"],
    "inner"
)

# cross join:全組み合わせ(明示的に crossJoin を使う)
calendar = dates.crossJoin(stores)   # 日付 × 店舗の全組み合わせ

2. broadcast join:小さいテーブルを配って高速化

通常の結合(shuffle join)では、同じキーの行を同じワーカーに集めるために、両方のテーブルをネットワーク越しに再配置(シャッフル)します。シャッフルはネットワーク転送とディスクI/Oを伴う高コストな処理です。

ここで片方のテーブルが十分小さい場合、小さいテーブル全体を全ワーカーにコピー(ブロードキャスト)してしまえば、大きいテーブルは動かさずその場で結合できます。これが broadcast join(ブロードキャスト結合)です。

from pyspark.sql.functions import broadcast

orders = spark.read.table("silver.orders")        # 数億行の大テーブル
regions = spark.read.table("silver.region_master") # 数百行の小テーブル

# 小さい側を broadcast() で包む
df = orders.join(broadcast(regions), on="region_code", how="left")

-- SQL ではヒント構文で指定
SELECT /*+ BROADCAST(r) */ o.*, r.region_name
FROM silver.orders o
LEFT JOIN silver.region_master r
  ON o.region_code = r.region_code;

3. union:縦方向の連結

join が「横に列を増やす」操作なら、union は「縦に行を積む」操作です。スキーマが同じ2つの DataFrame(例:2026年の注文と2025年の注文)を1つにまとめます。

df_all = df_2026.union(df_2025)

# 列の「順序」ではなく「名前」で対応付けたい場合
df_all = df_2026.unionByName(df_2025)

# 片方にしかない列を null で許容する場合
df_all = df_2026.unionByName(df_2025, allowMissingColumns=True)

-- SQL 版:UNION ALL は重複を残す、UNION は重複を除去する
SELECT * FROM sales_2026
UNION ALL
SELECT * FROM sales_2025;

4. 結合の落とし穴:行の膨張と null キー

4-1. 重複キーによる行の膨張

結合は「1対1」とは限りません。キーが左に m 行・右に n 行あると、結果は m×n 行になります。たとえば注文テーブルに customer_id=101 が3行、顧客テーブルに誤って同じ顧客が2行入っていると、結合後はその顧客の注文が 3×2=6 行に「膨張」します。集計すると売上が2倍にカウントされる、という典型的な事故です。

# 結合前:右テーブルのキーの一意性を確認する
customers.groupBy("customer_id").count().filter("count > 1").show()

# 一意でなければ先に重複排除(3-4 で詳述)してから結合する
customers_unique = customers.dropDuplicates(["customer_id"])
df = orders.join(customers_unique, on="customer_id", how="left")

# 結合後:行数が変わっていないか検算する
assert df.count() == orders.count()

4-2. null キーは一致しない

SQL の比較演算では NULL = NULL は真になりません。そのためキーが null の行は、どの行とも結合されません。inner join では null キーの行が黙って消え、left join では右側が全て null のまま残ります。「結合したのに件数が減った」「マスタの値が付かない行がある」ときは、キーの null を疑ってください(null 処理は 3-1 で学んだとおりです)。

# 結合前に null キーの件数を把握する
orders.filter(F.col("customer_id").isNull()).count()

# 要件に応じて:除外する / 代替キーで埋める / left join で残す
orders_valid = orders.filter(F.col("customer_id").isNotNull())

練習問題