Spark & PySpark は何をしますか？
遅いSparkジョブのほとんどは、回避可能なシャッフル、スキューしたキー、またはPythonが1行ずつデータを処理することの3つの原因のいずれかによって遅くなっています。よくある間違いは、実行計画を読む前にクラスターサイズを調整することです。より多くのエグゼキュータにお金を払って同じ無駄の多いDAGをより速く実行することになります。
運用手順
ステップ1: 入力を収集する
- ジョブコード、および遅いジョブの場合はSpark UIの遅いステージのビュー（タスク期間分布、シャッフル読み取り/書き込みサイズ）と
df.explain(True)の出力。 - 入力データのサイズと形式、およびすべての結合の両側のサイズ。
- クラスターの形状：エグゼキュータ数、コア数、メモリ。
- ジョブが新規作成かチューニングパスか - チューニングの場合は、コードに触れる前に実行計画から診断する。
ステップ2: 適切なセッションから始める
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = (
SparkSession.builder
.appName("etl")
.config("spark.sql.shuffle.partitions", "200")
.config("spark.sql.adaptive.enabled", "true")
.getOrCreate()
)
Adaptive Query Execution (AQE) を有効にします。これにより、実行時にシャッフルパーティションが統合され、結合戦略が切り替わります。次に spark.sql.shuffle.partitions を意図的に設定します：シャッフルパーティションあたり100〜200 MBを目標とし、パーティション数 ≈ シャッフルデータサイズ ÷ 128 MB、かつ合計エグゼキュータコアの少なくとも2〜3倍にすることで、どのコアもアイドル状態になりません。
ステップ3: DataFrame API内に留まり、Pythonを避ける
DataFrame操作はCatalystとTungstenを通過するため、クエリ最適化とオフヒープメモリが利用できます。DataFrameのプリミティブが存在しない場合にのみRDDにフォールバックします。Python UDFは各行をPythonワーカーにシリアライズして返すため、コード生成を阻害します。優先順位：
- 組み込みの
pyspark.sql.functions（例：F.regexp_extract、F.when）。 - Pythonが避けられない場合のPandas（ベクトル化）UDF。
- スカラーPython UDFは最後の手段としてのみ。
from pyspark.sql.functions import pandas_udf
@pandas_udf("double")
def normalize(s):
return (s - s.mean()) / s.std()
ステップ4: サイズに基づいて結合戦略を選択する
- 小さい方のテーブルがメモリに収まる場合はブロードキャストします。Sparkは
spark.sql.autoBroadcastJoinThreshold（デフォルト10 MB）未満を自動ブロードキャストします。エグゼキュータメモリが許せば、最大数百MBのテーブルを明示的にブロードキャストします。実用的な上限は約1 GBと見なしてください - ブロードキャストはテーブルをすべてのエグゼキュータにコピーするため、誤判断するとジョブがOOMになります。
from pyspark.sql.functions import broadcast
result = large.join(broadcast(small), "key")
- 結合前に列をフィルタリングおよび射影してシャッフルを減らします。削除されたすべての列と行は節約されたシャッフルバイトです。
- 2つの大きなテーブル → ソートマージ結合を受け入れますが、最初にスキューを確認します（ステップ5）。
ステップ5: スキューを診断して修正する
スキューの症状：1つまたは2つのタスクを除いてステージが完了している。最大タスク期間が中央値の約3〜5倍を超えている。Spark UIで1つのシャッフル読み取りパーティションが他のパーティションよりもはるかに大きい。AQEのスキュージョイン処理は、パーティションが中央値の5倍を超え、かつ256 MBを超える場合（デフォルト）にパーティションを分割します - 実行計画で実際に発動したことを確認します。AQEが役に立たない場合（例：スキューした集約）、ホットキーにソルトを適用します。大きい側のキーにランダムなサフィックス0..Nを追加し、小さい側をすべてのNサフィックスに展開し、結合してからソルトを取り除きます。N ≈ スキュー係数（中央値の20倍のキーにはN=20）を選択します。
ステップ6: パーティショニングと出力レイアウトを制御する
repartition(n, col)はデータのバランスをとるためにフルシャッフルを実行します。ワイド書き込みの前に使用します。coalesce(n)はシャッフルなしでパーティション数を減らします。小さな出力ファイルを避けるために使用します。- 出力ファイルのサイズは128 MB〜1 GBを目標にします。数千の小さなファイルはすべての下流リーダーに悪影響を及ぼします。
- 出力はカーディナリティの低い列（日付、地域 - 数十から数百の値、IDは不可）でのみパーティション分割します：
df.write.partitionBy("dt").mode("overwrite").parquet("/data/out")
ステップ7: 再利用されるものだけをキャッシュする
DataFrameがDAG内で複数回使用される場合にのみキャッシュし、マテリアライズして、使用が終わったらアンペアします：
df.cache()
df.count() # マテリアライズ
# ... dfを再利用 ...
df.unpersist()
一度しか使用しないDataFrameをキャッシュするとメモリを浪費し、重要なデータを追い出す可能性があります。
ステップ8: 実行計画で確認する
df.explain(True) を再実行し、意図した戦略が表示されていることを確認します（BroadcastHashJoin、プッシュダウンされたフィルター、削除された Exchange がないこと）。列形式（Parquet、ORC）を読み取り、述語プッシュダウンが機能するようにします。決して大きなデータを collect() してドライバに送らないでください。チェーンする際は、文字列の列参照よりも F.col 参照を優先します。
実例：悪い結合と良い結合
悪い例 - 2 TBのイベントテーブルと200 MBのディメンションを、全幅で、Python UDFで解析：
result = (events.join(dims, "store_id") # ソートマージ：2 TBすべてをシャッフル
.withColumn("region", parse_region_udf("meta")) # 行単位のPython
.filter(F.col("dt") == "2024-06-01")) # 結合後にフィルター
実行計画は両側に Exchange（2 TBシャッフル）を示し、UDFがコード生成をブロックします。
良い例 - 最初にフィルターと射影を行い、ディメンションをブロードキャストし、組み込み関数を使用：
result = (events
.filter(F.col("dt") == "2024-06-01") # まず約30 GBに絞る
.select("store_id", "amount", "meta")
.join(broadcast(dims.select("store_id", "region_code")), "store_id")
.withColumn("region", F.regexp_extract("meta", r"region=(\w+)", 1)))
実行計画は BroadcastHashJoin を示し、大きな側の Exchange はありません：約30 GBがスキャンされ、大きなテーブルのシャッフルはゼロ、Pythonの境界もありません。
成果物
修正されたジョブコードと、次の内容を含むチューニングノートを作成します：各結合の結合戦略とそのサイズ根拠、シャッフルパーティション設定とその計算根拠、発見されたスキュー（最大対中央値タスク比率）と適用された修正、キャッシュ判断、出力ファイルサイズの期待値、およびチューニングパスの場合は修正前後の explain の差分。
やってはいけないこと
- サイズを確認していないテーブルをブロードキャストしてはいけません。誤判断したブロードキャストはすべてのエグゼキュータを一度にOOMにします。
pyspark.sql.functionsを確認せずにPython UDFを作成してはいけません。ほとんどの文字列、日付、条件ロジックには組み込み関数があります。- 大きなDataFrameで
collect()またはtoPandas()を使用してはいけません。クラスターのデータを1つのドライバーに集約します。 - カーディナリティの高い列で出力をパーティション分割してはいけません。何百万ものディレクトリにそれぞれ1行ずつ格納されることになります。
- 遅いステージが1つのスキューしたタスクであるジョブにエグゼキュータを追加してはいけません。より多くのコアでも1つのパーティションを分割できません。
- 反射的にキャッシュしてはいけません。下流で再利用されるDataFrameのみをキャッシュし、アンペアします。
品質基準
完了したジョブは、次のすべてを満たした場合にのみ合格とします：すべての結合がその戦略とサイズ根拠を明示していること。spark.sql.shuffle.partitions がデータサイズから導出されており、習慣的な値のままになっていないこと。Spark UIで、シャッフルステージのどのタスクも中央値の約3倍を超える期間になっていないこと（またはスキューが説明され修正されていること）。組み込み関数が存在する場所にスカラーPython UDFが残っていないこと。出力ファイルが128 MB〜1 GBの範囲にあること。そして最終的な explain が意図した実行計画を確認していること。