スタート・ガイド
データ・フローでSpark動的割当ての使用を開始する方法。
データ・フローで動的割当てを使用するために必要な前提条件と構成。
前提条件
- Spark 3.xを使用します。
- Spark動的割当てのメリットがあるのは、処理時間が10分を超えるジョブのみです。
- Spark動的割当ては構造化ストリーミングで使用できますが、バッチ・ジョブ用に最適化されています。詳細は、「Spark動的割当ておよびSpark構造化ストリーミング」を参照してください。
- Spark動的割当てを有効にすると、シャッフル・トラッキングが有効になります。
- 外部シャッフル・サービスはサポートされていません。
Spark動的割当ての構成
データ・フローでSpark動的割当てを構成する方法は3つあります。
コンソールの使用
アプリケーションの作成時に、「自動スケーリングの有効化」をクリックします。Spark構成プロパティに、デフォルト構成が移入されます。エグゼキュータの最小数は、spark.dynamicAllocation.minExecutorsプロパティの値と同一になります。最大エグゼキュータ数は、spark.dynamicAllocation.maxExecutorsプロパティの値と同一になります。spark.dynamicAllocation.executorIdleTimeoutおよびspark.dynamicAllocation.schedulerBacklogTimeoutプロパティには、デフォルトとは異なる値を設定できます。
APIの使用
データ・フロー・アプリケーションのspark.dynamicAllocation.enabledとspark.dynamicAllocation.shuffleTracking.enabledの両方をtrueに設定します。
動的割当てが有効になっているSparkアプリケーションでは、スケジュールを待機している保留中のタスクがある場合に、エグゼキュータの追加がリクエストされます。この状況は、送信されてまだ終了していないすべてのタスクを同時に処理するには、既存のエグゼキュータでは足りないことを示しています。
Sparkでは、エグゼキュータが繰返しリクエストされます。実際のリクエストは、spark.dynamicAllocation.schedulerBacklogTimeoutの秒数保留されているタスクがある場合にトリガーされます。保留中のタスクのキューが存続する場合、リクエストはそれ以降、spark.dynamicAllocation.sustainedSchedulerBacklogTimeout秒ごとに再度トリガーされます。各回でリクエストされるエグゼキュータの数は、前の回から指数関数的に増加します。たとえば、最初のラウンドで1つのエグゼキュータ、2つのエグゼキュータ、4つのエグゼキュータなどが、後のラウンドで追加されます。
spark.dynamicAllocation.executorIdleTimeout秒を超えたときにエグゼキュータが削除されます。 ほとんどの場合、この条件はリクエスト条件と相互に排他的で、保留中のタスクがまだスケジュールされている場合、エグゼキュータがアイドル状態になることはありません。
Spark-Submitの使用
--conf spark.dynamicAllocation.enabled=true
--conf spark.dynamicAllocation.minExecutors=1
--conf spark.dynamicAllocation.maxExecutors=4
--conf spark.dynamicAllocation.executorIdleTimeout=60
--conf spark.dynamicAllocation.shuffleTracking.enabled=true最小構成
- spark.dynamicAllocation.enabled:
true -
spark.dynamicAllocation.shuffleTracking.enabled:
true -
spark.dynamicAllocation.minExecutors:
1 -
spark.dynamicAllocation.maxExecutors:
4 -
spark.dynamicAllocation.executorIdleTimeout:
60 -
spark.dynamicAllocation.schedulerBacklogTimeout:
60 -
spark.dataflow.dynamicAllocation.quotaPolicy:
max
プロパティの説明
データ・フローで使用できるSpark動的割当てプロパティと使用可能な値の説明。
| プロパティ名 | デフォルト値 | 値の範囲 | 説明 | サポートされるSparkのバージョン |
|---|---|---|---|---|
| spark.dynamicAllocation.enabled | false | true | false | 動的リソース割当てを使用するかどうか。このプロパティは、ワークロードに基づいて、アプリケーションに登録されているエグゼキュータの数をスケール・アップおよびスケール・ダウンします。 | 3.x |
| spark.dynamicAllocation.shuffleTracking.enabled | true | true | エグゼキュータのシャッフル・ファイル・トラッキングが有効化されるため、外部のシャッフル・サービスを使用せずに動的割当てが可能になります。このオプションは、アクティブなジョブのシャッフル・データが保存されているエグゼキュータを有効なまま維持することを目的としています。 | 3.x |
| spark.dynamicAllocation.minExecutors | 1 | [1, maxExecutors] | 動的割当てが有効な場合のエグゼキュータ数の下限。 | 3.x |
| spark.dynamicAllocation.maxExecutors | 4 | [minExecutors, 1000] | 動的割当てが有効な場合のエグゼキュータ数の上限。 | 3.x |
| spark.dynamicAllocation.executorIdleTimeout | 60 | [60, 600] | 動的割当てが有効になっている場合、エグゼキュータがアイドル状態になってからそのエグゼキュータが削除されるまでの時間(秒単位)。 | 3.x |
| spark.dynamicAllocation.schedulerBacklogTimeout | 60 | [60, 600] | 動的割当てが有効で、保留中のタスクがある場合に、新しいエグゼキュータがリクエストされるまでの時間(秒単位) | 3.x |
| spark.dataflow.dynamicAllocation.quotaPolicy | min | min | max |
データ・フローでは実行の開始時に、
データ・フローでは実行の開始時に、 |
3.x |