Sparkはすべてのコアをタスクに割り当てます

Question

私はより多くのコアの恩恵を受けることができるタスクを持っていますが、スタンドアロンスケジューラーはサブセットのみが使用可能な場合に起動します。むしろ、このタスクではすべてのクラスタコアを使用したいと思います。

リソースをタスクに割り当てる前にスケジューラにすべてを完了させる方法はありますか？言い換えれば、タスクを実行する前にすべてのパスを終了した場合、またはより多くのコアが使用可能になるまで待っていた場合、DAGはこのジョブに適しています。タスクが太っていることを示唆する方法でしょうか？私はそうではなく、糸を動かさないことを望んでいます。

簡潔に：それ以外のアイドル状態のクラスタでこのマップタスクを実行して、すべてのリソース/コアを持つようにする必要があります。これを行う方法はありますか？ハッキーな答えでさえ評価されるでしょう。

アイデア？

Answer 1

実行中のアプリケーションがない場合は、REST APIを使用してYARNを手動でチェックする必要があります。

GET http://<rm http address:port>/ws/v1/cluster/metrics 
{ 
    "clusterMetrics": 
    { 
    "appsSubmitted":0, 
    "appsCompleted":0, 
    "appsPending":0, 
    "appsRunning":0, 
    "appsFailed":0, 
    "appsKilled":0, 
    "reservedMB":0, 
    "availableMB":17408, 
    "allocatedMB":0, 
    "reservedVirtualCores":0, 
    "availableVirtualCores":7, 
    "allocatedVirtualCores":1, 
    "containersAllocated":0, 
    "containersReserved":0, 
    "containersPending":0, 
    "totalMB":17408, 
    "totalVirtualCores":8, 
    "totalNodes":1, 
    "lostNodes":0, 
    "unhealthyNodes":0, 
    "decommissionedNodes":0, 
    "rebootedNodes":0, 
    "activeNodes":1 
    } 
} 

保留中または実行中のアプリがない場合は、スクリプトを実行できます。 whileループ+スリープ状態にあって、両方が0になるのを待つシェルスクリプトを作成するだけです。

また、使用可能なメモリ/コアも検索できます。実際、私はあなたがいつも待っているというだけでなく、十分なリソースを保証するように、そのルートに行くでしょう。

Answer 2

動的リソース割り当てが、あなたが探しているものかもしれません。ワークロードに基づいてこのアプリケーションに登録されているエグゼキュータの数を増減します。

設定パラメータをSparkSessionに渡すことで有効にすることができます。以下のように：詳細はhttp://spark.apache.org/docs/latest/job-scheduling.html#dynamic-resource-allocation ：

val spark = SparkSession 
    .builder() 
    .appName("MyApp") 
    .config("spark.dynamicAllocation.enabled","true") 
    .config("spark.shuffle.service.enabled","true") 
    .getOrCreate() 

はこれを参照してください。