PySpark の注文を顧客と商品の参照データに結合する

はじめに

注文には顧客と商品の ID があり、別の参照テーブルには顧客の居住地域と商品の分類が記録されています。結合はキーを通じて記録をつなぎ、地域別や商品分類別の売上といった疑問に答えられるようにします。

参照テーブルを1つずつ使って支払い済み注文に情報を加え、曖昧さのない出力列を選び、合計を照合します。顧客が見つからない場合の INNER JOIN と LEFT JOIN も比較します。用意された顧客キーと商品キーは一意なので、一致する各注文は各参照テーブルに1記録を見つけるはずです。参照キーが重複すると結合行と売上が増える可能性があり、意図した関係の理解が重要です。

この独立した実験では、正常な注文、2顧客、3商品、一致しない例を別に準備します。型付きの読み取り、フィルター、十進数金額、集計、最小限の Parquet 出力は既習とします。Hive、SQL、Hadoop、pandas、クラスター設定は不要です。ターミナルを使ってください。

明示したキーで顧客の地域を加える

このステップでは、4つの支払い済み注文を、一意の顧客記録に結合します。元データの関係を確認します。

cd /home/labex/project

cat /home/labex/project/data/orders.csv

cat /home/labex/project/data/customers.csv

複数の注文が customer_id を共有できますが、参照テーブルには c1 と c2 が1行ずつあります。同じ ID で記録をつなぎます。行の位置を合わせるのは有効なルールではありません。c1 は East、c2 は West を提供します。

DataFrame の join は、相手の DataFrame、一致条件、結合の種類を受け取ります。inner は一致する記録だけを残します。DataFrame の alias("o") と alias("c") は、入力に短い参照名を付けます。col("o.customer_id") は注文側、col("c.customer_id") は参照側の列を指します。

両入力に customer_id があるため、select で注文側のキーと必要な各フィールドを明示的に選び、曖昧な重複名を避けます。DataFrame の別名は入力参照に、列の別名は結果フィールドに名前を付けます。リーダーと金額計算は既習のまま、次の完全な版を保存して実行します。

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
with_customers.orderBy("order_id").show(truncate=False)
with_customers.write.mode("overwrite").parquet("/home/labex/project/output/orders-with-regions")
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/customer-join.txt

cat /home/labex/project/output/customer-join.txt

情報が加わった支払い済み4行が期待されます。o1 と o4 は East、o2 と o6 は West で、金額は20.00、25.00、50.00、30.00のままです。4入力すべてに一意の一致があれば、結合後も4行です。Parquet 出力は型付きのフィールドを保って region を加え、元 CSV は変更しません。

商品分類を加えて売上を照合する

このステップでは、product_id で2つ目の参照データを加えます。商品の対応を確認します。

cat /home/labex/project/data/products.csv

p1 と p2 は Office、p3 は Home です。この表に customer_id で結合するのは誤った関係です。中間の情報追加済み注文フィールドを保ち、商品側からは category だけを選びます。

この完全な版は、最初の結合を残して2つ目を加えます。射影を明示し、同名キーを2つ残さず product_id を1つにします。その後、既習の groupBy と合計で分類別の売上を求めます。

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
enriched = (
    with_customers.alias("e")
    .join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
    .select(
        col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
        col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
        col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
        col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
        col("e.region").alias("region"), col("p.category").alias("category"),
    )
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/product-join.txt

cat /home/labex/project/output/product-join.txt

情報追加後も支払い済み注文は4件で、region と category が加わるはずです。分類別レポートは Home 30.00、Office 95.00で、合計は同じ支払い済み売上125.00です。

行数4の保持と売上の照合は、有用な2つの確認です。説明列を加えただけで、どちらも増えるべきではありません。このデータのキーは一意ですべて一致します。次は主な売上合計を変えず、独立した例で一致しない場合を調べます。

顧客が見つからない場合の INNER と LEFT を比較する

このステップでは、参照先が一致しない場合に各結合が何をするかを観察します。独立した例を確認します。

cat /home/labex/project/data/example-orders.csv

demo1 は c1、demo2 は customers にない c_missing を指します。まず INNER JOIN を予想します。残るのは demo1 だけのはずです。LEFT JOIN は左側の例をすべて残し、不足する右側のフィールドを NULL にします。

主な売上スクリプトとデータを変えずに保つため、小さな別プログラムを保存します。

cat > /home/labex/project/compare_joins.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName("CompareCustomerMatches").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
examples = spark.read.option("header", True).schema("order_id string, customer_id string").csv("/home/labex/project/data/example-orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
inner = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
left = examples.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "left").select(col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"), col("c.region").alias("region"))
print("Inner matches:")
inner.orderBy("order_id").show(truncate=False)
print("Left matches:")
left.orderBy("order_id").show(truncate=False)
inner.write.mode("overwrite").parquet("/home/labex/project/output/example-inner")
left.write.mode("overwrite").parquet("/home/labex/project/output/example-left")
spark.stop()
PYTHON

bigdata-python /home/labex/project/compare_joins.py > /home/labex/project/output/join-comparison.txt

cat /home/labex/project/output/join-comparison.txt

Inner matches は demo1/c1/East です。Left matches にはその行に加え demo2/c_missing/NULL があるはずです。右側の地域がなくても、左側の customer_id は残ります。NULL は一致する地域の欠如であり、NULL という名前の地域ではありません。

完全な一致を確認済みのレポートなら INNER JOIN が適切かもしれません。LEFT JOIN は品質チェックや意図した業務方針のため、未解決の記録を残せます。結合の種類は、表示の好みではなく、どの記録が残るかを変える選択です。

加えたフィールドで地域別に比較する

このステップでは、主な情報追加済みデータに、既習のフィルターと集計を適用します。全分類の地域別売上は East 70.00、West 55.00のはずです。Office だけの結果を予想してください。East は70.00のまま、West は Home の30.00が除かれて25.00になります。

2つの派生レポートを加えた完全な主スクリプトを実行します。結合は同じまま、分類条件は2つ目の集計に入る注文行だけを変えます。

cat > /home/labex/project/enrich_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("EnrichOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
products = spark.read.option("header", True).schema("product_id string, category string").csv("/home/labex/project/data/products.csv")

with_customers = (
    paid.alias("o")
    .join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner")
    .select(
        col("o.order_id").alias("order_id"), col("o.customer_id").alias("customer_id"),
        col("o.product_id").alias("product_id"), col("o.quantity").alias("quantity"),
        col("o.unit_price").alias("unit_price"), col("o.order_status").alias("order_status"),
        col("o.order_date").alias("order_date"), col("o.order_amount").alias("order_amount"),
        col("c.region").alias("region"),
    )
)
enriched = (
    with_customers.alias("e")
    .join(products.alias("p"), col("e.product_id") == col("p.product_id"), "inner")
    .select(
        col("e.order_id").alias("order_id"), col("e.customer_id").alias("customer_id"),
        col("e.product_id").alias("product_id"), col("e.quantity").alias("quantity"),
        col("e.unit_price").alias("unit_price"), col("e.order_status").alias("order_status"),
        col("e.order_date").alias("order_date"), col("e.order_amount").alias("order_amount"),
        col("e.region").alias("region"), col("p.category").alias("category"),
    )
)
category_sales = enriched.groupBy("category").agg(spark_sum("order_amount").alias("revenue"))
enriched.write.mode("overwrite").parquet("/home/labex/project/output/enriched-orders")
category_sales.write.mode("overwrite").parquet("/home/labex/project/output/category-sales")
enriched.orderBy("order_id").show(truncate=False)
category_sales.orderBy("category").show(truncate=False)
regional = enriched.groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
office = enriched.filter(col("category") == "Office").groupBy("region").agg(spark_sum("order_amount").alias("revenue"))
regional.write.mode("overwrite").parquet("/home/labex/project/output/regional-sales")
office.write.mode("overwrite").parquet("/home/labex/project/output/office-regional-sales")
print("All categories by region:")
regional.orderBy("region").show(truncate=False)
print("Office category by region:")
office.orderBy("region").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/enrich_orders.py > /home/labex/project/output/regional-comparison.txt

cat /home/labex/project/output/regional-comparison.txt

予想した全分類と Office の地域別結果になるはずです。Office の地域別合計は95.00で、前のステップの Office 小計と一致します。全地域を足すと125.00です。

これらの照合で、結合関係と選択した業務上の範囲の両方を説明できます。主な情報追加済み Parquet には、正しい型の支払い済み4注文が残り、一致しない例は別に保たれています。次は、DataFrame のビューに対して Spark SQL で同等の操作を表します。

まとめ

一意の顧客キーと商品キーを明示して注文に情報を加え、曖昧でない出力列を選び、分類別と地域別の合計を照合しました。INNER と LEFT は、一致しない場合に意図どおり異なる動作をしました。次は、同じ構造化データに対する Spark SQL と DataFrame 式を比較します。

Spark の注文と商品・顧客データを結合する

はじめに

明示したキーで顧客の地域を加える

商品分類を加えて売上を照合する

顧客が見つからない場合の INNER と LEFT を比較する

加えたフィールドで地域別に比較する

まとめ