BigData向けワークロード（Spark/Flink）に適したKubernetesカスタムスケジューラ Apache YuniKorn について

大事なバッチにはクラスタの半分のリソースを優先して割り当てる
部門Aのアドホックなワークロードにはクラスタリソースの10%までを割り当てる
1. ただし、バッチなど優先したいワークロードがなくヒマしているなら、10%を超えて使ってもOK
2. とはいえ、メインのバッチが始まったらワークロードをプリエンプティブする
3. さらに、部門Aのリソースのうち、半分はBさんが優先して使ってOK

そして、これまでHadoopエコシステムのストレージ層はHDFSでしたが、Hadoop基盤自体がパブリッククラウドで利用されるようになった結果、S3、ADSL、GCSといったオブジェクトストレージを中心に利用するようになりました（HDFSの場合は常時稼動が必要になり割高になるので）。

また、データを貯めていたストレージ層とワークロードを実行するコンピュート層が分離しやすい環境が整ってきた結果、コンピュート層を柔軟にスケール出来る環境としてコンテナの活用が期待されています。

そこで、ワークロードの中心的なSparkとFlinkは、YARN以外にもKubernetesでアプリを実行できるようになりました⁵。

一方、Kubernetesのリソーススケジューラは、上記のような柔軟なリソーススケジューリングには対応できないので、そこを補う形でBigDataやMLワークロードに特化したリソーススケジューラとしてYuniKornが出てきました。

ただ、Kubernetesとしてバッチのワークロードについて何もしていないわけではなく、 Batch Working Group として議論・サポートを進めています⁶。

どこが使ってるのか？

Hadoopといえば、ClouderaですがClouderaのパブリッククラウドサービスのCDP Public CloudのCloudera Data Engineering（CDE）というサービスにYuniKornが利用されています（AWSとAzureで使えるみたいです）。

また、Cloudera以外にもAlibaba、Apple、Lyft、Visa、Zillowで使っているようです。

そして、10月にあったYuniKorn MeetupではPinterestがスケジューラの性能評価について話していて、YARNからKubernetes（YuniKorn）への移行を検討しているようです⁷。

リリース状況

YuniKorn自体は、今年の5月（2022/5/6）にGAしています！

現状の最新は 2022/9/8 にリリースされた v1.1.0 となっています。
また、Roadmapによると、年明け1月には、v1.2.0がリリース予定のようです。

YuniKornの特徴

YuniKornの主な機能は以下の通りです。

アプリを考慮したスケジューリング

KubernetesデフォルトのPod単位のスケジューリングとは異なり、ユーザー、アプリ、キューを認識して、リソースの状況や順序などの要素を考慮しながらスケジューリングを行います。

Kubernetesでは、PodをSubmitする際にPodがNamespaceのクォータに収まるように強制し、収まらない場合には、Podは拒否されます。その為、クライアント側で再試行出来るよう複雑な実装が必要になります。

一方、YuniKornは、Kubernetesのクォータとは異なり、SubmitされたたPodは常に受け入れられます。

つまり、YuniKornはキューに入ったPodのリソースを消費したクォータとしてカウントせずに、そのPodをキューにいれます。YuniKornがPodをスケジュールしようとすると、スケジューリング時にPodが割り当てられたキューに設定されたクォータに収まるかどうかチェックします。その時にPodがクォータに収まらない場合、Podはスキップされてリソース消費にカウントされません。これは、Podのスケジュール試行に成功するまでは、YuniKornクォータシステムでPodはリソースを消費していないことを意味しています。

階層型のリソースキュー

クラスタリソースを階層型にキューで管理します。また、キューの最小と最大容量を設定出来ます。階層ごとにリソースを制御出来るので、マルチテナント利用の場合に柔軟な対応が可能になります。

そして、リソースキューの状況はYuniKornのUIでモニタリング出来ます（クラスタリソースの容量、利用率、およびすべてのアプリケーション情報を一元的に表示出来ます）。

出典： Get Started | Apache YuniKorn より

ジョブオーダーとキューイング

アプリは作業キューに格納して、どのアプリが最初にリソースを取得できるのかは、以下のポリシーに応じて決定します。

現状設定可能なポリシーは、以下の３つ。

FIFO ：そのまま。先入れ先出し。
Fair ：アプリケーションの使用量に応じて公平にリソースを配分。
StateAware ：実行中または受け入れ状態のアプリは１つに制限される。

StateAware が分かりにくいので補足します。
Sparkアプリを例にとると想像しやすいです。Sparkアプリは、アプリ全体の調整役となるDriverと、Driverが割り当てた細かいタスクを実行する複数のExecuterに分かれて実行していきます。

例えば、あるキュー（app-queue-1）にapp1→app2の順にSparkアプリをsubmitしたときは以下の様に実行されます。

さらに、キューの最大容量を設定した場合、ジョブやタスクはリソースキューに適切にキューイングされて、残りの容量が十分でない場合は、いくつかのリソースが開放されるまで待機させることが出来ます。

リソースの公平性

マルチテナント環境で、多くのユーザーがクラスタリソースを共有します。テナントがリソースを競合させ、飢餓状態（starving）を避けるために、アプリケーションに割り当てるリソースの重みや優先度を設定出来るようになっています。

リソース予約

YuniKornは未解決のリクエストに対して自動的に予約します。Podが割り当てられなかった場合、YuniKornは適切なNodeにPodを予約しようとして、（他のNodeを試す前に）予約されたNodeにPodを仮で割り当てます。

この仕組みにより、後からSubmitされたより小さくてこだわりのないPodによってこのPodが飢餓状態になる事を回避出来ます。

高いスループット

スループットは、スケジューラの性能を測る重要な基準です。スループットが悪いとアプリケーションはスケジュール待ちで時間を浪費し、サービスのSLAに影響します。クラスタが大きくなるほど高いスループットが要求されます。

Kubemarkを用いて、YuniKornとKubernetesのデフォルトのスケジューラに比べた結果、YuniKornが2倍から4倍のスループットを出したようです。

詳細は以下を参照ください。

yunikorn.apache.org

YuniKornのアーキテクチャ

YuniKornは、カスタムKubernetesスケジューラとしてデプロイ出来るようになっています。
これは異なるshim層を追加して、Apache Hadoop YARNやその他のシステムを含む異なるリソースマネージャの実装に対応する事が可能になっています。

しかし、独自実装を盛り込んだ結果、Kubernetesのアップデートに追従するのが大変でしたが、YuniKorn v1.0.0からKubernetes v1.19以降に追加されたScheduling Frameworkで導入された拡張ポイント（PreFilter、Filter、PostBindの3つ）を実装したプラグインとして利用できるようになりました⁸。

下図は、YuniKornのScheduler Pluginを利用しない場合のハイレベルなアーキテクチャです。

出典： Architecture | Apache YuniKorn より

コンポーネント

Scheduler Core

※図中のYuniKorn Coreの事

Scheduler Coreは、すべてのスケジューリングアルゴリズムをカプセル化して、下位のリソース管理プラットフォーム（YARN/Kubernetesなど）からリソースを収集し、コンテナの割り当て要求を担当します。

各リクエストに対して最適な場所はどこかを判断し、リソース管理プラットフォームにレスポンスの割り当てを下位のプラットフォームに依存しないようにすべての通信はScheduler Interfaceを利用します。

詳しくはScheduler Coreの設計を参照ください。

Scheduler Interface

※図中のYuniKorn Scheduler Interfaceの事

リソース管理プラットフォーム（YARN/Kubernetesなど）がgRPCやプログラミング言語バインディングのようなAPIを介して会話する抽象レイヤ

Kubernetes Shim

※図中のYuniKorn K8s-Shimの事

Kubernetes ShimはKubernetesとの対話を担当し、Kubernetesのクラスタリソースや、Scheduler Interface経由のリソースリクエストをScheduler Coreに送ります。そしてスケジューラが決定すると、Podを特定のNodeに結合する役割を担っています。ShimとScheduler Core間の通信はすべてScheduler Interfaceを介して行われます。

詳しくは、Kubernetes Shimの設計を参照ください。