分散処理（Spark）

Spark は、大きなデータを小さな塊（パーティション）に分け、多くのマシン（エグゼキューター）で並列に処理するエンジンです。DataFrame や SQL で書くと、Spark が実行計画を作って分散させます。DWH の SQL では難しい大量のファイル処理や、Python での複雑な変換、機械学習の前処理によく使います。

Spark では、filter や join などの変換を書いただけでは何も計算されません。write や count などのアクションを呼んだときに、それまでの変換をまとめて見て、むだのない手順を考えてから実行します（遅延評価）。

同じ DataFrame を何度もアクションに使うと、そのたびに最初から計算し直します。何度も使う途中の結果は cache するか、いったん書き出します。

シャッフルは、GROUP BY・JOIN・DISTINCT・並べ替えのときに、同じキーのデータを同じマシンに集めるため、ネットワーク越しにデータを並べ替えて送り合う処理です。ディスクとネットワークを大量に使うので、Spark が遅いときの主な原因です。不要な列や行はシャッフルの前に落とし、小さいテーブルとの JOIN はブロードキャスト JOIN（小さいほうを全マシンに配る）にしてシャッフルをなくします。

シャッフルのあとのパーティション数は spark.sql.shuffle.partitions（既定は 200）で決まります。AQE が有効なら、実行中に小さなパーティションをまとめてくれます。

スキュー（データの偏り）は、特定のキーにデータが集中することです。たとえば user_id が NULL の行や、特定の大口の取引先にデータの大半が集まると、そのキーを担当する1つのタスクだけがいつまでも終わらず、全体がそれを待ちます。Spark の画面でタスクの所要時間に極端な差があれば、スキューを疑います。

分散処理の JOIN は、どちらの表がどれくらい大きいかで方式が変わります。Spark は表の大きさの見積もりから自動で選びますが、見積もりが外れると遅い方式になることがあります。

1つのタスクが扱うデータがメモリに入らないと、Spark は途中の結果をディスクに書き出して続けます（スピル）。動きはしますが遅くなり、それでも足りないとエグゼキューターがメモリ不足（OOM）で落ちます。

ファイルを読むときのパーティションの大きさは spark.sql.files.maxPartitionBytes（既定は 128MB）で決まります。設定を変える前に、読むデータを減らせないかを先に考えます。

書き出しのときは、出力のファイル数にも注意します。パーティションが多すぎると小さなファイルが大量にでき、次に読む処理が遅くなります。

• partitionBy("dt") で日付ごとのフォルダに書くと、各タスクが各日付にファイルを作るので、タスク数×日付の数だけファイルができることがある。書く前に日付で並べ直すなどして減らす。
• テーブル形式（Iceberg・Delta Lake）なら、書いたあとにまとめ直す機能も使える。
• 1ファイルの大きさは、数百 MB 前後を目安にする（スモールファイル問題）。