"Managed Service for Apache Spark" is the new name for the product formerly known as "Dataproc on Compute Engine" (cluster deployment) and "Google Cloud Serverless for Apache Spark" (serverless deployment).

Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

ワークロードの resource_allocation_properties を自動スケーリングする

このドキュメントでは、Managed Service for Apache Spark の自動スケーリングについて説明します。 Spark ワークロードを送信すると、Managed Service for Apache Spark は、エグゼキュータの数などのワークロードリソースを動的にスケーリングして、ワークロードを効率的に実行できます。Managed Service for Apache Spark の自動スケーリングはデフォルトの動作であり、 Spark 動的リソース割り当てを使用して、ワークロードをスケーリングするかどうか、またその方法とタイミングを決定します。

Managed Service for Apache Spark 自動スケーリング V2

Managed Service for Apache Spark 自動スケーリングバージョン 2（V2）では、デフォルトバージョン 1（V1）に機能と改善を追加して、Managed Service for Apache Spark ワークロードの管理、ワークロードのパフォーマンスの改善、費用の削減を行います。

非同期ノードのスケールダウン: 自動スケーリング V2 は、V1 の同期スケールダウンを非同期スケールダウンに置き換えます。非同期スケールダウンを使用すると、Managed Service for Apache Spark は、すべてのノードがシャッフル移行を完了するのを待たずに、ワークロードリソースをスケールダウンします。つまり、スケールダウンに時間がかかるロングテールノードがスケールアップをブロックすることはありません。
インテリジェントなスケールダウンノード選択: 自動スケーリング V2 は、V1's ランダムなノード選択を、最初にスケールダウンするのに最適なノードを特定するインテリジェントなアルゴリズムで置き換えます。このアルゴリズムでは、ノードのシャッフルデータサイズやアイドル時間などの要素が考慮されます。
構成可能な Spark の正常なデコミッションとシャッフル移行の動作: 自動スケーリング V2 では、標準の Spark プロパティを使用して、Spark の正常なデコミッションとシャッフル移行を構成できます。この機能は、カスタマイズされた Spark プロパティとの移行互換性を維持するのに役立ちます。

Managed Service for Apache Spark 自動スケーリングの機能

機能	Managed Service for Apache Spark 自動スケーリング V1	Managed Service for Apache Spark 自動スケーリング V2
ノードのスケールダウン	同期	非同期
スケールダウンのノード選択	ランダム	インテリジェント
Spark の正常なデコミッションとシャッフル移行	構成不可	構成可能

Spark の動的割り当てのプロパティ

次の表は、バッチワークロードを送信して自動スケーリングを制御する際に設定できる Spark の動的割り当てプロパティの一覧です（ Spark プロパティの設定方法をご覧ください）。

プロパティ	説明	デフォルト
`spark.dataproc.scaling.version`	Managed Service for Apache Spark 自動スケーリングのバージョン。バージョンバージョン`1`または`2`を指定します。	`1`
`spark.dynamicAllocation.enabled`	ワークロードに基づいてエグゼキュータの数をスケールアップまたはスケールダウンする動的リソース割り当てを使用するかどうか。値を `false` に設定すると、ワークロードの自動スケーリングが無効になります。デフォルト: `true`。	`true`
`spark.dynamicAllocation.initialExecutors`	ワークロードに割り当てられるエグゼキュータの初期数。ワークロードが開始されると、自動スケーリングによってアクティブなエグゼキュータの数が変化することがあります。最小値は `2` で、最大値は `2000` です。	`2`
`spark.dynamicAllocation.minExecutors`	ワークロードをスケールダウンするエグゼキュータの最小数。最小値は `2` です。	`2`
`spark.dynamicAllocation.maxExecutors`	ワークロードをスケールアップするエグゼキュータの最大数。最大値は `2000` です。	`1000`
`spark.dynamicAllocation.executorAllocationRatio`	Spark ワークロードのスケールアップをカスタマイズします。 `0`～`1` の値を受け付けます。`01`値 `1.0` は、最大スケールアップ機能を提供し、最大並列処理を実現するのに役立ちます。値 `0.5` では、スケールアップ機能と並列処理が最大値の半分に設定されます。	`0.3`
`spark.dynamicAllocation.diagnosis.enabled`	`true` の場合、実行中のエグゼキュータが `spark.dynamicAllocation.diagnosis.interval` で指定された期間に必要な最大エグゼキュータ数を超えると、診断情報がログに記録されます。診断には、アイドル状態のエグゼキュータ数とアイドル時間のパーセンタイル、アクティブタスクの分布、シャッフルデータサイズ、キャッシュされた RDD サイズを含むエグゼキュータの概要が含まれます。`spark.dynamicAllocation.diagnosis.logLevel` を使用して、出力ログレベルを制御します。	`false`
`spark.dynamicAllocation.profile`	`performance` または `cost` に設定すると、パフォーマンスまたは費用対効果に最適化された事前定義済みの構成セットが適用されます。ユーザー定義のプロパティは、プロファイルのデフォルトをオーバーライドします。詳細については、 Spark 動的割り当てプロファイルをご覧ください。	`none`
`spark.dynamicAllocation.shuffleTracking.dynamicTimeout.enabled`	`true` の場合、シャッフルデータを保持するエグゼキュータの動的タイムアウト計算が有効になります。静的な `spark.dynamicAllocation.shuffleTracking.timeout` を使用する代わりに、タイムアウトはエグゼキュータに保存されているシャッフルデータの量に基づいて計算されます。これにより、シャッフルが少ないエグゼキュータはより早く解放され、シャッフルが多いエグゼキュータはより長く存続します。	`false`
`spark.reducer.fetchMigratedShuffle.enabled`	`true` に設定すると、Spark 動的割り当てによってデコミッションされたエグゼキュータからのフェッチが失敗した後、Spark ドライバからシャッフル出力ロケーションをフェッチできます。これにより、廃止されたエグゼキュータからライブエグゼキュータへのシャッフルブロックの移行によって発生する `ExecutorDeadException` エラーが減り、 `FetchFailedException` エラーによって発生するステージの再試行が少なくなります（ ExecutorDeadException による FetchFailedException をご覧ください）。このプロパティは、Managed Service for Apache Spark Spark ランタイムバージョン `1.1.12` 以降と `2.0.20` 以降で使用できます。	`false`
`spark.scheduler.excludeShuffleSkewExecutors`	`true` の場合、シャッフルスキューエグゼキュータ（大量のシャッフルデータまたは完了したマップタスクの数が多いエグゼキュータ）でタスクのスケジュール設定を回避します。これにより、シャッフルスキューを軽減してパフォーマンスを向上させることができます。	`false`

Spark 動的割り当てプロファイル

spark.dynamicAllocation.profile プロパティを performance または cost に設定すると、パフォーマンスまたは費用対効果に最適化された事前定義済みの Spark 構成セットが適用されます。spark.dynamicAllocation.profile プロパティの設定に加えて Spark プロパティを設定すると、それらのプロパティのプロファイルのデフォルトがオーバーライドされます。

**performance**: このプロファイルは、次のデフォルト設定を適用して、実行時間を最小限に抑えるように最適化します。

spark.scheduler.excludeShuffleSkewExecutors: true
spark.dynamicAllocation.executorIdleTimeout: 300s
spark.dynamicAllocation.initialExecutors: 10

cost: このプロファイルは、次のデフォルト設定を適用して、リソース消費量を削減するように最適化します:

spark.dynamicAllocation.executorIdleTimeout: 120s
spark.dynamicAllocation.cachedExecutorIdleTimeout: 120s
spark.dynamicAllocation.shuffleTracking.dynamicTimeout.enabled: true
spark.dynamicAllocation.diagnosis.enabled: true

Spark 動的割り当ての指標

Spark バッチワークロードは、Spark 動的リソース割り当てに関連する次の指標を生成します（Spark 指標の詳細については、モニタリングとインストルメンテーションをご覧ください）。

指標	説明
`maximum-needed`	実行中のタスクと保留中のタスクをすべて達成するために現在の負荷で必要なエグゼキュータの最大数。
`running`	タスクを実行している稼働中のエグゼキュータ数。

Spark 動的割り当ての問題と解決策

ExecutorDeadException による FetchFailedException

原因: Spark 動的割り当てがエグゼキュータをスケールダウンすると、シャッフルファイルがライブエグゼキュータに移行します。ただし、エグゼキュータの Spark レデューサタスクは、レデューサタスクの開始時に Spark ドライバによって設定されたロケーションからシャッフル出力をフェッチするため、シャッフルファイルが移行されても、レデューサは引き続き廃止されたエグゼキュータからのシャッフル出力のフェッチを試行できます。これにより、ExecutorDeadException と FetchFailedException エラーが発生します。

解決策: Managed Service for Apache Spark バッチワークロードを実行するときに、 spark.reducer.fetchMigratedShuffle.enabled を true に設定して、シャッフルロケーションの再取得を有効にします（ Spark バッチワークロードのプロパティの設定をご覧ください）。このプロパティを有効にすると、廃止されたエグゼキュータからのフェッチに失敗した後、レデューサタスクはドライバからシャッフル出力のロケーションを再取得します。