Spark 注文の欠損・重複・不正な記録を整える

はじめに

スクリプトが正常に実行されても、不正な記録をレポートに渡してしまうことがあります。今日届いた小売データには、注文行の重複、顧客の欠損、不正な数値文字列、正でない数量、負の価格、不正な日付があります。これらを自動的に0で埋めると、業務情報を作り出してしまいます。

判断を3つに分けます。元の証拠を失わずテキストを読み、目的の数値・日付フィールドを安全に解析し、定めたルールを満たす行だけを残します。その後、有効な同一行の重複を除きます。最後に、整えた Parquet データセットを保存して読み戻します。Parquet は型付きのバイナリ列指向ファイル形式です。ここでは最小限の読み書きだけを学び、パーティション分割した保存は後で扱います。

準備済み環境に、独立した2つのデータが用意されています。CSV 読み取り、Spark の列表現、フィルター、十進数の価格は既習とします。今回の安全な変換式は、準備済みの Spark バージョンで使えるので、エラー処理の設定を変える必要はありません。ターミナルを使います。以前の出力ファイル、Hadoop、Hive、サービスのインストールは不要です。

届いたデータを文字列のステージングとして保持する

このステップでは、値を変換する前に問題を観察します。ローカルの入力を調べます。

cd /home/labex/project

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

業務上の行は14です。ステージングの DataFrame は入力を文字列として保持し、two、unknown、not-a-date などを、有効な数値や日付と見なさずに確認できるようにします。CSV リーダーの既定の処理では、空の顧客フィールドは NULL になります。NULL は文字列の "NULL" ではなく欠損値です。

完全なステージングプログラムを保存して実行します。前と同じく、引用符付きの PYTHON ヒアドキュメントはソースをそのまま書きます。show(30, truncate=False) は最大30記録を表示し、フィールドのテキストを省略せず読みやすくします。

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
staging.printSchema()
staging.orderBy("order_id").show(30, truncate=False)
print("Staging rows:", staging.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/staging-orders.txt

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

14記録と Staging rows: 14 が期待されます。全列が文字列のはずです。b2 の数量 two と b4 の価格 unknown を探してください。推測した既定値で隠すべきエラーではなく、有用な証拠です。o2 は2回あり、o1 の状態には前後の空白があります。

この演習の元ファイルは読み取り専用です。届いた元データを確認用に残し、派生した DataFrame と出力ディレクトリを作ります。

既定値を作り出さず、安全に解析する

このステップでは、前後の空白を除く必要がある2つのテキストフィールドを整え、数値と日付の変換を試みます。trim は先頭と末尾の空白を除きます。欠けた顧客を推測したり、注文の支払い状態を変えたりはしません。

expr では、Spark の組み込み式をテキストで指定できます。try_cast(quantity AS INT) は整数変換を試し、変換できなければ NULL にします。同様の式で価格を DECIMAL(10,2)、日付を DATE にします。通常のキャストは Spark の ANSI 動作で失敗するとエラーになる場合があるため、明示的な安全な変換を使い、全体設定には依存しません。

型が変わっても行は残るため、解析の問題を観察できます。この版を実行します。

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
    staging
    .withColumn("customer_id", trim(col("customer_id")))
    .withColumn("order_status", trim(col("order_status")))
    .withColumn("quantity", expr("try_cast(quantity AS INT)"))
    .withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
    .withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
parsed.printSchema()
parsed.orderBy("order_id").show(30, truncate=False)
print("Parsed rows:", parsed.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/parsed-orders.txt

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

Parsed rows: 14 が期待されます。b2 の数量、b4 の価格、b6 の日付は NULL になるはずです。b3 の数量0と b5 の負の価格は、構文上は有効な数値ですが、次に適用する業務ルールには違反します。o1 の状態は空白のない paid になるはずです。

変換の成功と有効な注文は同じではありません。段階を分けると、解析に失敗したのか業務制約に違反したのかを説明できます。価格の欠損は0で埋めていないため、本当に価格0の注文 o7 と不明な価格を区別したままにできます。

妥当性と重複のルールを明示して適用する

このステップでは、有効な業務記録を残し、完全に同じ重複だけを除きます。次のルールを使います。

• customer_id が存在し、空白除去後に空でないこと。
• quantity の解析が成功し、0より大きいこと。
• unit_price の解析が成功し、0以上であること。
• order_date の解析が成功していること。
• 正規化した7フィールドすべてが同じ行は1つだけ残すこと。

ほかの ID フィールドは、このサンプルではすでに正しい形式です。同じ注文 ID の矛盾した版はないため、変化する状態のどれを採用するかは扱いません。ほかの条件を満たせば paid、pending、refunded はすべて残します。届いたデータを整えることと、レポートの支払い済み売上を選ぶことは別です。

isNotNull は存在を明示的に調べます。NULL を含む比較は真の一致にはならず、filter は真の条件だけを残します。& で組み合わせると方針が見える形になります。引数なしの dropDuplicates() は完全一致行を削除します。order_id だけを使うと、矛盾した情報を失いかねない別の方針になります。

次の版を実行します。

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
    staging
    .withColumn("customer_id", trim(col("customer_id")))
    .withColumn("order_status", trim(col("order_status")))
    .withColumn("quantity", expr("try_cast(quantity AS INT)"))
    .withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
    .withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
    col("customer_id").isNotNull()
    & (col("customer_id") != "")
    & col("quantity").isNotNull() & (col("quantity") > 0)
    & col("unit_price").isNotNull() & (col("unit_price") >= 0)
    & col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
print("Clean rows:")
clean.orderBy("order_id").show(30, truncate=False)
print("Staging count:", staging.count())
print("Valid count before deduplication:", valid.count())
print("Clean count:", clean.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/cleaning-result.txt

cat /home/labex/project/output/cleaning-result.txt

Staging count: 14、Valid count before deduplication: 8、Clean count: 7 が期待されます。b1～b6 はルールで除外され、繰り返された o2 が1つ除かれます。残る7つの ID は o1 から o7 です。

o7 の実際の価格0は、価格を非負とするルールでは有効です。o3 は支払い待ち、o5 は返金済みのままです。どちらの状態も元記録が不正だという意味ではありません。結果はこの方針に従って整えたデータであり、現実のあらゆる異常が解決したという主張ではありません。

保存して読み戻し、不正値を1つ追加して試す

このステップでは、整えたデータを再利用可能にし、別のデータにも同じ方針を適用します。write.mode("overwrite").parquet(path) はディレクトリに Parquet データセットを書き、そこにある以前の派生成果物を置き換えます。元入力とは別のディレクトリを使ってください。Spark は複数の part ファイルや完了マーカーを作れるので、特定の名前や個数には依存しないでください。

spark.read.parquet(path) は保存したデータセットと型を読みます。書く前の計画を表示するだけでなく、読み戻した DataFrame を調べることで実際の成果物を確認します。次の完全なプログラムでは、既習のロジックを Python 関数に置き、両方の入力で同じルールを使います。

def clean_orders(input_path): は関数を定義し、入力パラメーターに名前を付けます。インデントされた行は本体で、関数が呼ばれると実行されます。最初の呼び出しは元のファイル名、2つ目は追加データのファイル名を渡します。各呼び出しの中で input_path は渡された値を指します。return clean は結果の DataFrame を呼び出し元に返すだけで、保存はしません。インデントのない書き込み行が、別のアクションとして保存します。プログラム全体をコピーするとき、インデントを保ってください。

追加データは元の14行に、数量が負の行と、有効な価格0の注文の完全なコピーを加えたものです。結果を予想してください。どちらの追加も、残る7記録を変えないはずです。追加分を確認します。

tail -n 2 /home/labex/project/data/dirty-orders-extra.csv

完全な保存・読み戻しプログラムを実行します。

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

def clean_orders(input_path):
    staging = spark.read.option("header", True).csv(input_path)
    parsed = (
        staging
        .withColumn("customer_id", trim(col("customer_id")))
        .withColumn("order_status", trim(col("order_status")))
        .withColumn("quantity", expr("try_cast(quantity AS INT)"))
        .withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
        .withColumn("order_date", expr("try_cast(order_date AS DATE)"))
    )
    valid = parsed.filter(
        col("customer_id").isNotNull()
        & (col("customer_id") != "")
        & col("quantity").isNotNull() & (col("quantity") > 0)
        & col("unit_price").isNotNull() & (col("unit_price") >= 0)
        & col("order_date").isNotNull()
    )
    clean = valid.dropDuplicates()
    return clean

clean = clean_orders("/home/labex/project/data/dirty-orders.csv")
clean.write.mode("overwrite").parquet("/home/labex/project/output/clean-orders")
reloaded = spark.read.parquet("/home/labex/project/output/clean-orders")
print("Read-back clean rows:")
reloaded.orderBy("order_id").show(30, truncate=False)
reloaded.printSchema()
print("Read-back count:", reloaded.count())

practice = clean_orders("/home/labex/project/data/dirty-orders-extra.csv")
practice.write.mode("overwrite").parquet("/home/labex/project/output/practice-clean-orders")
practice_reloaded = spark.read.parquet("/home/labex/project/output/practice-clean-orders")
print("Extra-delivery clean rows:")
practice_reloaded.orderBy("order_id").show(30, truncate=False)
print("Extra-delivery count:", practice_reloaded.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/clean-parquet-check.txt

cat /home/labex/project/output/clean-parquet-check.txt

両方の読み戻した表に o1～o7 がちょうど1回ずつあり、件数はどちらも7のはずです。主出力は output/clean-orders で、practice-clean-orders は別の派生した実験用です。読み戻したスキーマでは、quantity の integer、価格の decimal、order_date の date が保たれるはずです。

これは最小限の Parquet の保存と読み戻しであり、速度比較や保存のチューニングではありません。別の Spark プロセスが読める小さな整ったデータと、新しい不正行や重複が黙って結果を変えない証拠ができました。

まとめ

元のテキストを保ち、安全な変換で不正値を見えるようにして、明示的な欠損・範囲の方針を適用し、完全一致の重複行を除きました。有効な価格0と未払い記録は残し、実際の型付き Parquet 出力と2つ目のデータを確認しました。次のチャレンジでは、これらのクレンジングと金額計算を独力で応用します。