Spark の注文データを選択・絞り込み・変換する

はじめに

正しい型を持つ注文表は出発点であり、まだ答えではありません。同僚が必要とするのは、少数の列、支払い済み注文だけ、数量と単価から計算した金額かもしれません。元の記録を利用できるまま、これらの判断を Spark DataFrame の操作で表します。

列表現は、Spark が各行に適用する計算や条件を記述します。フィールドの選択、条件の組み合わせ、十進数の金額の追加、分かりやすい結果列名の指定を行います。これらの操作は新しい DataFrame を作り、show などのアクションが実際の結果を求めます。この違いから、まだスケジューリングの内部を学ばずに、Spark の遅延実行を理解できます。

この実験では正常な小売注文6件を独立して用意します。前の実験の明示的な CSV スキーマと Python の基礎を理解していることを想定します。各実行版に完全なリーダーと SparkSession が含まれるため、以前のプロセスやファイルは不要です。ターミナルと準備済みの bigdata-python 環境を使います。

疑問に必要な列を選ぶ

このステップでは、6件の注文をすべて残しながら、表の列を絞ります。元データから始めます。

cd /home/labex/project

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

select は、新しい DataFrame にどの列をどの順で公開するかを選びます。行は絞り込みません。次の完全なスクリプトは、既習の明示的なリーダーを繰り返し、注文 ID、数量、価格、状態を選びます。日付と顧客のフィールドは selected には出ませんが、df に残っています。

引用符付きのヒアドキュメントで保存して実行します。> はファイルを作るか前の版を置き換え、最後の PYTHON がソースを終えます。

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

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt

cat /home/labex/project/output/selected-orders.txt

6つすべての注文 ID が、4列で表示されるはずです。orderBy が表示順を予測可能にし、show が行を返すために必要な作業を起動します。これは列の選択で、射影とも呼ばれます。元ファイルは変わらず、注文も削除されていません。

条件を組み合わせて行を選ぶ

このステップでは、数量が2以上の支払い済み注文を残します。col("order_status") は Spark の列を参照し、== "paid" との比較は、Python の単一の True や False ではなく、Spark 用の条件を作ります。>= は数量の条件を作ります。

Spark の列条件を AND の意味で組み合わせるには & を使います。Python の演算子の優先順位が関係するので、各比較を括弧で囲んでください。通常の Python の and は、Spark Column オブジェクトを組み合わせる演算子ではありません。

対象の ID を予想します。o1、o4、o6 が両条件を満たします。o2 は支払い済みですが数量1です。完全なスクリプトを置き換えて実行します。

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

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt

cat /home/labex/project/output/eligible-orders.txt

結果には、その3つの ID だけが、それぞれ数量2で含まれるはずです。filter は選ぶ行を変え、select は表示する列を決めます。両操作を組み合わせても、df は全6行の入力を表したままです。

この数量のしきい値は実演用の条件で、売上の一般的な定義ではありません。次の計算は、各支払い済み注文が売上に寄与するため、o2 も含む4注文すべてを使います。

十進数の注文金額を計算して名前を付ける

このステップでは、支払い済みの各注文について、数量と十進数の単価を掛けて金額を求めます。withColumn は、新しい DataFrame に名前付きの列を加え、同名の列があれば置き換えます。ここでは Spark 組み込みの乗算式で order_amount を計算します。

続く select は注文 ID を残し、alias で計算列の表示名を amount にします。別名は結果列名を変えるだけで、CSV ヘッダーは編集しません。4件の支払い済み注文すべてを計算に含め、pending と refunded は除外します。この単純化したルールでは、返金を負の売上にはしません。

次の完全な版を実行します。

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

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt

cat /home/labex/project/output/paid-amounts.txt

o1/20.00、o2/25.00、o4/50.00、o6/30.00 が期待されます。末尾の0がさらに付く場合もあります。スキーマでは amount が十進数型になっているはずです。Spark は乗算時に十進数の精度を広げる場合があります。重要なのは、価格を二進浮動小数点にせず、十進数の計算を保持することです。

filter、withColumn、select、orderBy は変換を記述し、新しい DataFrame を返します。この連結を定義しても、すべての中間テーブルがすぐ実体化されるわけではありません。show は、必要な作業を実行して答えを表示するよう求めるアクションです。printSchema は列の構造を示します。後で、この連結の背後にある実際の実行計画を調べます。

状態を1つ変え、効果を予想する

このステップでは、支払い済みの結果を残し、支払い待ちの独立した練習用ビューを加えます。実行前に予想しましょう。pending は o3 だけで、数量1、価格10.00なので金額は10.00です。

新しい pending_result は、同じ式の連結で状態値だけを変えます。すでに絞られた paid_result ではなく df から始めるので、支払い待ちの記録も利用できます。

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

spark = SparkSession.builder.appName("TransformOrders").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),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

pending_result = (
    df.filter(col("order_status") == "pending")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt

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

Paid orders の下には同じ4件の金額が残り、Pending practice の下には o3/10.00 だけが期待されます。paid のビューを pending で絞ると、その記録はすでに除かれているため0行になります。適切な入力から始めることが重要です。

どちらの連結も Spark の組み込み式で評価します。データ全体を Python リストに集めてループすることはしていません。これにより、Spark の実行エンジンに適した処理の記述を保ちます。最終スクリプトには支払い済みの計算と小さな比較練習が残り、元の CSV にも6注文すべてが残っています。

まとめ

列の選択、Spark 式による行条件の組み合わせ、十進数の注文金額の計算、出力列の命名を行いました。支払い済みと支払い待ちのビューを比べ、適切な入力から変換を始める理由と、アクションが結果を要求する仕組みを確認しました。次は、既習の読み取りと選択の考え方を JSON 入力に適用します。