sparkのリソース割り当てポリシーおよびジョブスケジューリングポリシー

Sparkには、計算間でリソースをスケジュールするためのツールがいくつかあります.Sparkが実行するクラスタマネージャ(cluster manager)(主にStandalone、Mesos、YARNの3種類を含む)は、静的パーティション方式と動的リソース割り当て方式の2つのアプリケーション間スケジューリングに便利です.次に、各Sparkアプリケーションの内部では、Sparkは各SparkContextのリソースをスケジューリングするための公平なスケジューラを含む.
1.アプリケーション間スケジューリング
1.1静的パーティション方式によるクラスタリソースの割り当て
すべてのクラスタマネージャで使用できる最も簡単なオプションは、リソースの静的パーティションです.この方法は、各Sparkアプリケーションが最大利用可能なリソースの総量を設定し、そのアプリケーションがライフサイクル全体でこのリソースを占めることを意味する.これはStandaloneおよびYARNモードおよび粗粒度Mesosモードで使用される方法である.
1.2動的資源配分方式
Sparkは、ワークロードに応じてアプリケーションが使用するリソースを動的に調整するメカニズムを提供します.これは、アプリケーションがリソースが使用されなくなったときにクラスタにリソースを返し、必要に応じて再び使用を要求できることを意味します.この機能は、Sparkクラスタ内のリソースを複数のアプリケーションが共有している場合に特に役立ちます.この機能はデフォルトでは無効になっており、すべての粗さクラスタマネージャ、すなわち独立モード、YARNモード、およびMesos粗さモードで使用できます.構成と導入方法について説明します.このプロパティを使用するには、2つの前提条件があります.まず、あなたのアプリケーションはsparkを設定する必要があります.dynamicAllocation.enabledはtrueです.次に、各ノードでexternal shuffleサービスを起動し、spark.shuffle.service.enabledをtrueに設定します.external shuffleサービスの目的は、executorを削除するときに、executorが出力するshuffleファイルを保持できることです.external shuffleサービスを有効にする方法は、クラスタマネージャごとに異なります.

Sparkが独立して配置するクラスタでは、workerが起動する前にsparkを設定する必要があります.shuffle.service.enabledはtrueでいいです.

Mesos粗粒度モードでは、各ノードで$SPARK_を実行する必要があります.HOME/sbin/start-mesos-shuffle-service.sh sparkを設定.shuffle.service.enabledはtrueでいいです.

YARNモードでは、次の手順に従って各NodeManagerで起動する必要があります.ここで

動的リソース割り当ての具体的な実装:全体的にSparkは、エフェクタ(executor)が空いているときにオフにし、後で使用するときに申請する必要があります.固定的な方法はないため、1つのエフェクタが後続ですぐにタスクを実行するかどうか、または新しい割り当てのエフェクタが実際に空いているかどうかを予測することができるので、エフェクタを申請するか削除するかを決定するための試行的な方法が必要です.

executorを要求するポリシー:動的割り当てが有効なSparkアプリケーションがタスクのスケジューリングを待つ必要がある場合、追加のexecutorsを申請します.この場合、既存のexecutorsは、すべての未完了のタスクを同時に実行するのに十分ではないことを意味します.Sparkは順番にエフェクタを申請します.実際のリソース申請は、タスクがspark.dynamicAllocation.schedulerBacklogTimeout秒保留された後に初めてトリガーされ、その後、待機キューに保留中のタスクがある場合は、spark.dynamicAllocation.sustainedSchedulerBacklogTimeout秒が経過するたびにリソース申請がトリガーされます.また,各申請のアクチュエータ個数は指数的に増加した.例えば、1つのSparkアプリケーションは、1回目に1つのエフェクタを申請することができ、後続の1回目の申請個数は2つ、4つ、8つであることができる.....インデックス・レベルの成長戦略を採用する理由は2つあります.1つ目は、少数の実行者だけで十分であることを防ぐために、アプリケーションが最初に慎重に実行器を要求する必要があります.第二に、Sparkアプリケーションが複数のエフェクタを申請する必要がある場合、必要なコンピューティングリソースがタイムリーに増加することを確保することができる.

executorを削除するポリシー:エフェクタを削除するポリシーはずっと簡単です.Sparkアプリケーションは、あるエフェクタの空きがspark.dynamicAllocation.executorIdleTimeout秒を超えた後に削除されます.多くの場合、エフェクタの削除条件と申請条件は反発します.つまり、エフェクタは、実行タスクの保留中である場合、空きを取るべきではありません.非動的割当てモードでは、エフェクタの終了原因として、実行に失敗したか、関連するSparkアプリケーションが終了した可能性があります.いずれの理由でも、アクチュエータのすべての状態は不要になり、破棄できます.しかし動的割当ての場合,アクチュエータはSparkアプリケーションの実行中に除去される可能性がある.この場合,Sparkアプリケーションがアクチュエータが格納している状態にアクセスしようとすると,このデータの一部を再計算する必要がある.したがってSparkは,アクチュエータを優雅に閉じながらその状態データを保持する機構が必要である.この問題を解決するには、Spark 1.2で導入されたexternal shuffleサービスが必要です.このサービスは、各ノードでSparkアプリケーションやエフェクタに依存しない独立したプロセスを開始し、executorを優雅に閉じることができます.

2.イントラスケジューリングの適用
指定されたSparkアプリケーション内(同じSparkContextインスタンスに対応)では、複数のスレッドが同時にSparkジョブ(job)をコミットすることがあります.Sparkスケジューラは完全なスレッドで安全であり、Sparkアプリケーションが複数のリクエスト(例えば、異なるユーザからのクエリー)を同時に処理することをサポートします.
2.1 FIFOスケジューリングポリシー
デフォルトでは、Sparkアプリケーション内でFIFOスケジューリングポリシーが使用されます.各ジョブは、mapフェーズおよびreduceフェーズなどの複数のフェーズに分けられ、最初のジョブは、起動後にすべての利用可能なリソースを優先的に取得し、2番目のジョブは再申請し、3番目のジョブは.......前のジョブがクラスタリソースを満たしていない場合、後続のジョブはすぐに実行を開始できます.そうしないと、後にコミットされたジョブに明らかな遅延が発生します.
2.2フェアスケジューリングポリシー
しかしSpark 0.8からSparkは各作業間の公平なスケジューリングもサポートできる.公平なスケジューリングではSparkは各ジョブにポーリングでリソースを割り当てるので,すべてのジョブで得られるリソースはほぼ平均的に割り当てられる.これは、大きなジョブが実行されていても、小さなジョブの再コミットは、前のジョブの終了を待つのではなく、すぐに計算リソースを得ることができ、遅延時間を大幅に削減することを意味する.このモードは、マルチユーザ構成に特に適している.公平スケジューラを有効にするには、SparkContextのsparkを設定するだけです.scheduler.modeプロパティはFAIRです.

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.set("spark.scheduler.mode", "FAIR")
val sc = new SparkContext(conf)

公平なスケジューリングポリシーが有効になっている場合、デフォルトでは各リソースプール間でクラスタ全体のリソースが分割されますが、リソースプール内ではデフォルトでは、ジョブはFIFO順序で実行されます.たとえば、各ユーザーにリソースプールを作成した場合、各ユーザー間でクラスタ全体のリソースが共有されていることを意味しますが、各ユーザーが自分でコミットしたジョブは、後でコミットしたジョブが前のジョブのリソースを優先することなく、順番に実行されます.デフォルトでは、新しくコミットされたジョブはデフォルトのリソースプールに入りますが、ジョブがどのリソースプールに対応するかは、ジョブをコミットするスレッドでSparkContextを使用できます.setLocalProperty設定spark.scheduler.poolプロパティ.サンプルコードは次のとおりです.

sc.setLocalProperty("spark.scheduler.pool", "pool1")

ローカルプロパティが設定されると、このリソースプールは、スレッドがコミットしたすべてのジョブ(すなわち、RDD.save/count/collectなどのaction演算子をスレッドに呼び出す)で使用されます.同様に、リソースプールの設定をクリアする必要がある場合は、対応するスレッドで次のコードを呼び出すだけです.

sc.setLocalProperty("spark.scheduler.pool", null)

もちろん、これらは変更できます.たとえば、フェアスケジューラは、ジョブをリソースプール(pool)にグループ化し、各リソースプールに異なるオプション(ウェイトなど)を構成することもできます.これにより、重要なジョブに「高優先度」リソースプールを作成したり、各ユーザーのジョブをグループ化したりすることができます.これにより、各ジョブがクラスタリソースを均等に分割するのではなく、各ユーザーがクラスタリソースを平均的に共有することができます.次に、リソースプールのプロパティはXMLファイルであり、conf/fairschedulerに基づいてもよい.xml.templateを修正し、SparkConfのspark.scheduler.allocation.fileプロパティファイルパスを指定するには、次の手順に従います.

conf.set("spark.scheduler.allocation.file", "/path/to/file")

リソースプールXMLプロファイルのフォーマットは次のとおりです.各プールには1つの要素が対応し、各リソースプールには独立した構成があります.

  
    FAIR
    1
    2
  
  
    FIFO
    2
    3
  

リソースプールのプロパティは、プロファイルで指定する必要があります.各リソースプールでは、次の3つのプロパティがサポートされています.

schedulingMode:リソースプール内のジョブがどのようにスケジュールされるかを制御します.FIFOまたはFAIRです.

weight:制御リソースプールが他のリソースプールに対してリソースに割り当てられる割合.デフォルトのすべてのリソースプールのweightは1です.あるリソースプールのweightを2に設定すると、そのリソースプールのリソースは他のプールの2倍になります.1000などのweightを高く設定すると、リソースプール間のスケジューリングの優先度を実現できます.すなわち、weight=1000のリソースプールは、対応するジョブをすぐに開始することができます.

minShare:全体weightに加えて、リソースプールごとに最小リソース割り当て値(CPU個数)を指定できます.管理者がこの設定を必要とする場合があります.フェアスケジューラは、アクティブなすべてのリソースプールの最小リソース割り当て値を優先的に満たし、各プールのweightに基づいて残りのリソースを割り当てます.したがって、minShareプロパティは、各リソースプールが少なくとも一定量のクラスタリソースを得ることを保証することができる.minShareのデフォルト値は0です.プロファイルで構成されていないリソースプールでは、デフォルトの構成が使用されます(schedulingMode:FIFO、weight:1、minShare:0).