mcpField
S
spark-optimization

spark-optimization

github.com/full-stack-skillsデータ・データベース147 回インストール4 週間前に更新
リポジトリ

spark-optimization について

Apache Spark ジョブのチューニングとトラブルシューティングに役立つ実践パターンをまとめたスキルです。性能、メモリ、パーティション、shuffle、データ偏りに取り組む技術者向けです。

主な機能

  1. 01パーティションサイズの戦略
  2. 02メモリと executor の調整
  3. 03shuffle とデータ偏りの軽減
  4. 04Spark 実行モデルの概要
  5. 05references/details.md の実践例

活用シーン

  1. 01遅い Spark ジョブの原因調査
  2. 02executor のメモリと設定の調整
  3. 03大規模データセット向けパイプラインの拡張
  4. 04shuffle とデータ偏りの低減

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/full-stack-skills/devops-skills --skill spark-optimization

インストール後、Claude Code で/spark-optimizationを入力して読み込まれたか確認します。

同梱ファイル

SKILL.md と一緒にインストールされる参考資料やスクリプト。

LICENSE.txt
agents/
openai.yaml
examples/
advanced-usage.md
basic-usage.md
integration.md
references/
details.md

詳細

出典の声明：wshobson/agents を基にしたスキルの内容です
（MIT License、Copyright (c) 2024 Seth Hobson。ライセンス全文は同じディレクトリの LICENSE.txt をご覧ください）。
使用、変更、再配布にあたっては、この MIT ライセンスの条項に従い、著作権表示を保持する必要があります。

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() を使用します

コメント

Markdown 対応