Spark の実行とよくあるエラーを理解する

はじめに

これまで Spark の変換を記述し、アクションを起動してデータを保存しました。今度は、それらを実行の証拠に結び付けます。作業を説明する計画と、ジョブ・ステージ・タスクを記録した実際のイベントログを使います。目的は内部のタスク数の暗記ではなく、動作を説明し、有用なエラーを読むことです。

ドライバーはアプリケーションを実行し、作業を調整します。クラスターでは Executor が割り当てられたリソースでタスクを実行します。local[2] はこの1台で2つの実行スレッドを使います。ジョブはステージに分かれ、各ステージが計算用パーティションに対するタスクを実行します。特にクエリ計画や適応動作により、1アクションから複数ジョブが起きることもあり、正確な件数は変わります。

この独立した実験では、正常な注文、顧客、意図的に壊した2つの式、読み取り専用のイベント要約ツールを用意します。フィルター、集計、結合、Parquet 出力は既習とします。リソース調整、データ全体を Python ループへ集める処理、クラスター構築は行いません。ターミナルを使ってください。

既習のクエリを実行計画に結び付ける

このステップでは、結果を要求する前に、日次・地域別の支払い済み売上クエリを調べます。リーダー、フィルター、金額、顧客結合、集計は既習の操作です。

ここでは groupBy("order_date", "region") が2つのキーを組み合わせ、日付と地域の異なる組ごとにグループを作ります。10月14日の East は、10月15日の East とも10月14日の West とも別です。件数と合計は各組の中で計算します。以前の日次と地域別の疑問を1つのレポートにまとめます。explain(mode="formatted") は、結果行ではなく計画された演算子を表示します。

SparkSession は output/spark-events に実際のイベントログも有効にします。アプリケーション停止後に学習用ログを読めるよう、圧縮を無効にします。この設定は実行を記録するもので、作り物のジョブ記録ではありません。保存先ディレクトリは準備済みです。

完全な確認スクリプトを作って実行します。

cd /home/labex/project

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

spark = (
    SparkSession.builder.appName("InspectRetailExecution").master("local[2]")
    .config("spark.eventLog.enabled", "true")
    .config("spark.eventLog.dir", "file:///home/labex/project/output/spark-events")
    .config("spark.eventLog.compress", "false")
    .config("spark.eventLog.rolling.enabled", "false")
    .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")

customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
matched = paid.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(
    col("o.order_date").alias("order_date"), col("c.region").alias("region"), col("o.order_amount").alias("order_amount")
)
report = matched.groupBy("order_date", "region").agg(
    count("*").alias("paid_order_count"), spark_sum("order_amount").alias("revenue")
)
print("Report execution plan:")
report.explain(mode="formatted")
print("Report action result:")
report.orderBy("order_date", "region").show(truncate=False)
report.write.mode("overwrite").parquet("/home/labex/project/output/daily-regional-sales")
print("Computation partition IDs represented by source rows:")
orders.select(spark_partition_id().alias("compute_partition")).dropDuplicates().orderBy("compute_partition").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/inspect_execution.py > /home/labex/project/output/execution-plan-and-result.txt

cat /home/labex/project/output/execution-plan-and-result.txt

Report execution plan で、フィルター、射影、結合、集計、計算用パーティション間で記録を移す Exchange の役割を探します。最適化によって名前や構造は変わります。Exchange は実行の境界であり、新しい出力ディレクトリではありません。

Report action result の下には、10月14日 / East / 1 / 20.00、10月14日 / West / 1 / 25.00、10月15日 / East / 1 / 50.00、10月15日 / West / 1 / 30.00 が期待されます。合計は引き続き125.00です。show とライターが記述された作業の実行を要求しますが、計画や入力メタデータの処理も準備作業を行う場合があります。

最後は spark_partition_id を使い、元の行が属する計算パーティション ID を表示します。dropDuplicates は観察した各 ID を1回だけ出します。ID と件数は変わり、空のパーティションは出ない場合があります。実行時のこのグループは、前の実験で partitionBy が書いた日付ディレクトリとは異なります。一方は作業の構成、もう一方は保存データの配置です。

実際のジョブ・ステージ・タスクのイベントを読む

このステップでは、実行したアプリケーションのイベントログを調べます。用意されたツールは、そのディレクトリの JSON イベントを読み、実際のイベントを要約します。ログ解析プログラムの作成や、バージョン固有のファイル名の選択は不要です。

イベントディレクトリを引数にして要約ツールを実行します。シェルの > で確認用の出力を残します。

bigdata-python /home/labex/project/tools/summarize-events.py /home/labex/project/output/spark-events > /home/labex/project/output/event-summary.txt

cat /home/labex/project/output/event-summary.txt

実際のアプリケーション識別子、正の Jobs started、Stages submitted、Tasks ended の件数、ジョブからステージへの参照が期待されます。数は実行に依存し、最適化、ファイル分割、Spark バージョンで変わります。正確に再現すべき目標ではなく、観察した証拠です。

アクションが答えを要求すると、Spark は必要な作業を複数のジョブにする場合があります。ジョブはステージを参照し、ステージには分割した作業のタスクがあります。シャッフルがステージを分ける一方、複数の変換が1ステージにまとまることもあります。local[2] は同時実行スレッドを制限するもので、どのアプリケーションにもジョブ・ステージ・タスクがちょうど2つあるという意味ではありません。

小さなデータでは、計算自体に比べて JVM の起動、スケジュール、ファイル処理が目立つ場合があります。小さな時間差からは信頼できる調整の結論は出せません。まず計画とイベントの証拠で作業を理解しましょう。

不明な列を診断して修正する

このステップでは、意図的な名前解決のエラーを読んでから原因を直します。準備済みの例を調べます。

cat /home/labex/project/tools/broken-column.py

スキーマは quantity を宣言していますが、最後の select は quanity を求めています。実際の失敗出力を保存します。> は標準出力をログへ送り、2>&1 は標準エラーも同じ場所へ送るので、トレースバックも残ります。次のコマンドは失敗するのが期待される動作です。

bigdata-python /home/labex/project/tools/broken-column.py > /home/labex/project/output/column-error.log 2>&1

tail -n 20 /home/labex/project/output/column-error.log

解決できない列名と、有効な名前の候補を探してください。長いトレースバックには実装詳細もありますが、列の参照が修正箇所を示します。入力の再インストールや、Python ループへの変更は不要です。

壊れた学習用ファイルを比較用に残し、宣言された列名で修正版を作ります。

cat > /home/labex/project/fixed_column.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("OrderExpressionExample").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")
fixed = orders.select("order_id", "quantity")
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-quantities")
spark.stop()
PYTHON

bigdata-python /home/labex/project/fixed_column.py > /home/labex/project/output/fixed-column-result.txt

cat /home/labex/project/output/fixed-column-result.txt

6つすべての ID と数量2、1、1、2、1、2が期待されます。今度は実際のスキーマで列が解決し、結果が保存されます。つづりも含めた列名は、データの取り決めの一部です。

真偽値でないフィルター条件を修正する

このステップでは、有効な整数列と、有効なフィルター条件を区別します。2つ目の例を確認します。

cat /home/labex/project/tools/broken-filter.py

filter は、行ごとの Boolean 式を必要とします。col("quantity") だけでは整数であり、真偽の判定ではありません。意図的な型の不一致を記録します。次のコマンドは失敗するのが期待される動作です。

bigdata-python /home/labex/project/tools/broken-filter.py > /home/labex/project/output/filter-error.log 2>&1

tail -n 20 /home/labex/project/output/filter-error.log

フィルター条件と、Boolean ではない整数型を探してください。修復には既習の列演算子を使い、意図した比較 quantity > 0 を表します。修正済みの完全なプログラムを実行します。

cat > /home/labex/project/fixed_filter.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("OrderExpressionExample").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")
fixed = orders.filter(col("quantity") > 0)
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-orders")
spark.stop()
PYTHON

bigdata-python /home/labex/project/fixed_filter.py > /home/labex/project/output/fixed-filter-result.txt

cat /home/labex/project/output/fixed-filter-result.txt

正常な6行はすべて数量が正なので、すべて残ります。負または0の数量はこの条件に合いません。この修正自体では、別の支払い済み売上ルールは適用しません。

これで、意図した式とスキーマ、実際の計画・実行、対象を絞って修正した後の正しい保存結果という3段階の証拠ができました。実行を説明するときも、保存した日付ディレクトリと計算用パーティションを区別してください。次は、入力と出力のパラメーターを明示した独立バッチスクリプトを投入します。

まとめ

既習の変換を実際の計画とイベントログに結び付け、アプリケーション・ジョブ・ステージ・タスクの件数を固定せず証拠を観察し、計算用パーティションと日付ディレクトリを区別しました。スキーマとエラーを根拠に、解決できない列と Boolean でないフィルターを修正しました。次は投入時のパラメーターを明示し、以前のプロセスに依存しないバッチアプリケーションにします。