spark-optimization

Apache Spark の最適化

パーティショニング戦略、メモリ管理、シャッフル最適化、パフォーマンスチューニングなど、Apache Spark ジョブを最適化するための本番向けパターンです。

このスキルを使用する場面

• 遅い Spark ジョブの最適化
• メモリとエグゼキューター構成のチューニング
• 効率的なパーティショニング戦略の実装
• Spark のパフォーマンス問題のデバッグ
• 大規模データセット向けの Spark パイプラインのスケーリング
• シャッフルとデータスキューの削減

主要な概念

1. Spark の実行モデル

Driver Program
    ↓
Job (triggered by action)
    ↓
Stages (separated by shuffles)
    ↓
Tasks (one per partition)

2. 主要なパフォーマンス要因

要因	影響	解決策
シャッフル	ネットワーク I/O、ディスク I/O	ワイド変換を最小限に抑える
データスキュー	タスク実行時間の偏り	ソルティング、ブロードキャスト結合
シリアライズ	CPU オーバーヘッド	Kryo、列指向フォーマットを使用
メモリ	GC プレッシャー、スピル	エグゼキューターのメモリを調整
パーティション	並列度	パーティションを適切なサイズに

クイックスタート

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# Create optimized Spark session
spark = (SparkSession.builder
    .appName("OptimizedJob")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.shuffle.partitions", "200")
    .getOrCreate())

# Read with optimized settings
df = (spark.read
    .format("parquet")
    .option("mergeSchema", "false")
    .load("s3://bucket/data/"))

# Efficient transformations
result = (df
    .filter(F.col("date") >= "2024-01-01")
    .select("id", "amount", "category")
    .groupBy("category")
    .agg(F.sum("amount").alias("total")))

result.write.mode("overwrite").parquet("s3://bucket/output/")

詳細なパターンと実例

詳細なパターンのドキュメントは references/details.md にあります。上記のナビゲーション層では不十分な場合は、そのファイルをお読みください。

ベストプラクティス

すべきこと

• AQE を有効化する - 適応クエリ実行が多くの問題を処理します
• Parquet/Delta を使用する - 圧縮を伴う列指向フォーマット
• 小さなテーブルをブロードキャストする - 小さな結合ではシャッフルを避ける
• Spark UI を監視する - スキュー、スピル、GC を確認する
• パーティションを適切なサイズにする - パーティションあたり 128MB ～ 256MB

すべきでないこと

• 大量のデータを collect しない - データを分散させたままにする
• 不必要に UDF を使用しない - 組み込み関数を使用する
• 過剰にキャッシュしない - メモリは限られています
• データスキューを無視しない - それがジョブ時間の大部分を占めます
• 存在確認に .count() を使用しない - .take(1) または .isEmpty() を使用する