はじめに
小売業者には、正しい型を持つ支払い済み注文の結果がすでにあります。今度は別のアプリケーションが、後で結果を読み、1日を選び、エクスポートの繰り返しで注文が気付かないうちに重複していないと信頼できる必要があります。保存も処理の取り決めの一部であり、スクリプトの最後の1行というだけではありません。
order_date によるディレクトリのパーティション分割で Parquet を書きます。データセットのルートには日付名の子ディレクトリができ、Spark はルートを読むと日付列を復元します。この保存用ディレクトリは、作業の実行に使う計算用パーティションとは別です。実行は次の実験で調べます。
この実験では、十進数金額を持つ支払い済み4注文を、ローカル Parquet 入力に独立して用意します。DataFrame の読み取り、フィルター、最小限の上書き Parquet は既習とします。Hive カタログ、HDFS、SQL、インストール、以前の出力は不要です。ターミナルを使用してください。
日付別データセットを書き、読み戻す
このステップでは、準備済みの結果を日付ディレクトリに整理します。data/prepared-orders には4注文があり、数量、十進数の価格、日付、十進数の order_amount は以前と同じ型です。Parquet はバイナリなので、cat で CSV が見えると考えず、Spark で調べます。
作業ディレクトリから始めます。
cd /home/labex/project
partitionBy("order_date") は、日付名のディレクトリに行を分けるようライターに指定します。mode("overwrite") は、指定の出力パスにある既存の派生データセットを置き換えます。入力を保持するため、出力パスは input_path と別にしてください。
完全なプログラムを保存して実行します。
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/partitioned-read-back.txt
cat /home/labex/project/output/partitioned-read-back.txt
復元された4記録、10月14日の o1 と o2、10月15日の o4 と o6 が期待されます。金額は20.00、25.00、50.00、30.00です。Main rows は4のはずです。quantity が integer、価格と金額が decimal、order_date が date と確認します。
パーティション列は、各ファイルの通常フィールドではなくディレクトリ名から復元されるため、読み戻すと位置が変わる場合があります。列の名前と値がその列を識別します。後続の利用者が列順を要求するなら select を明示してください。表示順にも orderBy が必要です。
ディレクトリを調べ、1日を選ぶ
このステップでは、ディレクトリ構成と復元された日付列を結び付けます。出力を再帰的に一覧表示します。-R は日付の子ディレクトリとそのファイルも含めます。
ls -R /home/labex/project/output/orders-by-date
order_date=2026-10-14 と order_date=2026-10-15 が期待されます。part ファイル名、個数、隠れたメタデータは変わるため、特定の1ファイルをデータセットと決めつけないでください。意味を持つルートは orders-by-date です。
ルートを読むことで、Spark がパーティション値を発見できます。このファイル配置の検出には Hive のパーティションやテーブルの登録は不要です。次の独立したスクリプトはルートを読み、DateType 列と有効な ISO 日付文字列を比べ、選んだ日の記録を別に保存します。
cat > /home/labex/project/inspect_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("InspectDateDataset").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
root = "/home/labex/project/output/orders-by-date"
df = spark.read.parquet(root)
selected = df.filter(col("order_date") == "2026-10-14")
selected.orderBy("order_id").show(truncate=False)
selected.write.mode("overwrite").parquet("/home/labex/project/output/selected-day")
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_sales_data.py > /home/labex/project/output/selected-date.txt
cat /home/labex/project/output/selected-date.txt
10月14日の o1 と o2 だけが、金額20.00と25.00で出るはずです。別の日付は主データセットに残ります。フィルターした読み取りは返す記録を変えるだけで、選ばれないディレクトリは削除しません。
保存の構成と実際のクエリ結果を確認できました。この小さなサンプルの経過時間では、ある形式や配置が速いとは判断できません。目的は、後の読み手が使える、正しく検出可能なデータセットです。
上書きの繰り返しと追記を安全に比較する
このステップでは、正しい主データセットを保ち、別のディレクトリで実験します。練習用の件数を予想してください。同じ4行を2回上書きしても毎回4行で、その後に全体を追記すると8行になります。増えた行は新しい注文ではなく、エクスポートの重複です。
完全な版を実行します。主パスは上書きし、意図的な追記は write-mode-demo だけに行います。
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
practice_path = "/home/labex/project/output/write-mode-demo"
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("First overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("Second overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("append").partitionBy("order_date").parquet(practice_path)
print("After append rows:", spark.read.parquet(practice_path).count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Main rows: 4、First overwrite rows: 4、Second overwrite rows: 4、After append rows: 8 を探します。練習用には意図的に各注文が2コピー入り、完全なスナップショットの追記が主エクスポートに適さないことを示します。
保存したプログラムを再実行し、全体の再現性も確認します。
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
同じ件数になるはずです。1回のプログラム全体で主データを再構築し、意図的な追記の前に練習用をリセットするため、12や16行には増えません。追記は、本当に新しく重複しないデータのための別の方針です。今回の完全な派生スナップショットでは上書きを選びます。
元の prepared-orders は引き続き利用できます。日付ディレクトリは保存値を表し、Spark の実行時計算パーティションはジョブの作業分割を表します。次に、この2つ目のパーティションの意味を、計画と実際の実行に結び付けます。
まとめ
日付別の Parquet を書き、ルートから型付き記録を復元して、ほかの日を削除せず1日を選びました。上書きの繰り返しで主スナップショットを保ち、練習用の追記でコピーの重複を確認しました。次は実行計画と実際のイベントから、計算用パーティション、ジョブ、ステージ、タスクを学びます。
