はじめに
役立つバッチアプリケーションは、以前のセッションに残った DataFrame や一時ビューに依存せず、保存したプログラムと明示した入出力で実行できるべきです。準備済みの日次売上プログラムを別プロセスに投入し、保存結果と実行の証拠を確認します。
spark-submit は Spark のアプリケーション起動ツールです。実行用 master、Python プログラム、プログラムの入出力引数を渡します。このスクリプトでは、外部の master 設定が実行を制御するようにしてあります。レポート処理は既習の型付き読み取り、paid フィルター、十進数金額、集計です。引数解析は提供済みなので、投入と安全な再実行に集中できます。
この独立した実験には、有効な2つの CSV と完全なプログラムがあります。ログと終了コードを調べ、入力パスを変え、入力不足と既存出力の競合を区別します。リモートクラスター、パッケージ化、スケジューラー、Hive、以前のプロセス状態は不要です。ターミナルを使ってください。
引数を明示して準備済みプログラムを投入する
このステップでは、最初のデータを独立したアプリケーションで処理します。入力とプログラムを調べます。
cd /home/labex/project
cat /home/labex/project/data/orders.csv
cat /home/labex/project/sales_batch.py
パーサーは input_path、output_path、任意の --overwrite を受け取るよう実装済みです。argparse を作る必要はありません。位置引数の最初は常に入力、2つ目は出力です。SparkSession の builder は master を呼ばないため、起動ツールの --master が実行を制御します。
コマンドを左から読んでください。spark-submit のオプションは Python ファイル名より前に置き、その後の引数はアプリケーションに渡されます。local[2] を引用符で囲み、シェルに角括弧をそのまま扱わせます。> は標準出力を保存し、2>&1 はエラー出力も同じログに送ります。
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-1 > /home/labex/project/output/run-1-job.log 2>&1
ほかのターミナルコマンドを実行する前に、直ちに終了状態を保存します。$? は直前のコマンドの状態で、0は成功、0以外は失敗です。printf で値と改行をローカルの状態ファイルに書きます。
printf '%s\n' "$?" > /home/labex/project/output/run-1-exit-status.txt
cat /home/labex/project/output/run-1-exit-status.txt
0が期待されます。ドライバーログの末尾を読みます。
tail -n 30 /home/labex/project/output/run-1-job.log
Master: local[2]、実際のアプリケーション ID、選択した入出力パス、レポートの日付を探します。ID は変わります。レポートは10月14日 / 2 / 45.00 と10月15日 / 2 / 80.00のはずです。終了コード0と、正しい実際の Parquet 結果を合わせると、安心できそうなログ1行より強い証拠になります。
プログラムは独自にセッションを作り、停止します。前の実験のビューや Python 変数は再利用しません。
新しいプロセスで入力を変える
このステップでは、指定したパスから結果が導かれることを確かめます。2つ目の入力を読みます。
cat /home/labex/project/data/new-orders.csv
10月16日の支払い済み2注文、金額30.00と25.00、さらに pending が1注文あります。レポート1行、件数2、売上55.00を予想します。最初の出力を残し、新しいファイルを別の出力先に投入します。
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/new-orders.csv /home/labex/project/output/run-2 > /home/labex/project/output/run-2-job.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/run-2-exit-status.txt
cat /home/labex/project/output/run-2-exit-status.txt
tail -n 30 /home/labex/project/output/run-2-job.log
状態0と、予想した10月16日の結果が期待されます。実際のアプリケーション ID は最初のプロセスと異なります。プログラムが新しいパスを読み、独自のセッションを作るので、残っているビューは不要です。
run-1 と run-2 を両方残すと、各入力と結果の関係を確認できます。記憶した答えを出すために元ファイルを変更しないでください。
入力不足と既存出力を別々に診断する
このステップでは、データを削除せず、2種類のパスの失敗を保存します。まず、存在しない入力を投入します。次のコマンドは失敗するのが期待される動作です。
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/missing-orders.csv /home/labex/project/output/unused-missing-input > /home/labex/project/output/missing-input.log 2>&1
ログを調べる前に、直ちに状態を保存します。
printf '%s\n' "$?" > /home/labex/project/output/missing-input-exit-status.txt
cat /home/labex/project/output/missing-input-exit-status.txt
tail -n 25 /home/labex/project/output/missing-input.log
0以外の状態と、missing-orders.csv を示すパス不在の診断が期待されます。入力パスを直すことが対処であり、master の変更やクラスターの再作成では直りません。
スクリプトの既定の errorifexists モードは、既存の出力先も保護します。既知の入力を、完成済み run-1 パスに対して試します。次のコマンドは失敗するのが期待される動作です。
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-1 > /home/labex/project/output/existing-output.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/existing-output-exit-status.txt
cat /home/labex/project/output/existing-output-exit-status.txt
tail -n 25 /home/labex/project/output/existing-output.log
0以外の状態と、出力がすでにあるという診断が期待されます。入力不足とは異なり、以前成功したレポートはそのまま残るはずです。
保持を優先する修復として、正しい入力と新しい出力ディレクトリを使います。
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-3 > /home/labex/project/output/run-3-job.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/run-3-exit-status.txt
cat /home/labex/project/output/run-3-exit-status.txt
tail -n 25 /home/labex/project/output/run-3-job.log
成功と、元の2日分の合計が期待されます。新しい出力先なら、元ファイルや広い出力ツリーを削除せず、以前の両結果と失敗の証拠を保持できます。
意図した派生スナップショットを明示的に再構築する
このステップでは、既知の完全なスナップショットに対し、置き換えを意図的に選びます。任意の --overwrite 引数は実装済みで、このプログラムの出力モードだけを overwrite に変えます。入力、ほかの結果パス、外部指定の master は変えません。
同じ入力から指定した派生結果全体を再構築したいときに使います。今回は run-1 がその既知のスナップショットで、ほかの実行と元入力は別に保ちます。
spark-submit --master 'local[2]' /home/labex/project/sales_batch.py /home/labex/project/data/orders.csv /home/labex/project/output/run-1 --overwrite > /home/labex/project/output/overwrite-job.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/overwrite-exit-status.txt
cat /home/labex/project/output/overwrite-exit-status.txt
tail -n 30 /home/labex/project/output/overwrite-job.log
状態0、Write mode: overwrite、同じ2行が期待されます。この新プロセスは、日付の重複を加えずスナップショットを置き換えます。アプリケーション ID は最初の実行と異なり、ログも別名です。
既存出力への有効な2対応を練習しました。両版を残したいなら新しいパスを使い、置き換えが目的なら既知の派生スナップショットを明示的に再構築します。入力パスの誤りには、依然として正しい入力が必要です。上書きはデータ不足の修復ではありません。これらの投入と再実行の習慣が、最後の再利用可能な売上パイプラインのチャレンジにつながります。
まとめ
引数付き Spark アプリケーションを独立したプロセスで起動し、実際の状態とログを記録して、異なる入力の結果を確認しました。入力不足と既存出力の競合を区別し、新しい出力先か、明示的なスナップショットの置き換えを選びました。最後のチャレンジでは、投入の習慣にクレンジング、情報追加、集計、分割出力を組み合わせます。
