YARNは数千ノードまでスケールすることが知られています。YARNのスケーラビリティはResourceManagerによって決まり、ノード数、アクティブなアプリケーション、アクティブなコンテナ、およびハートビートの頻度（ノードとアプリケーションの両方）に比例します。ハートビートを下げるとスケーラビリティは向上しますが、利用率は低下します（以前のHadoop 1.xの経験を参照）。このドキュメントでは、複数のYARNサブクラスタを統合することで、単一のYARNクラスタを数万ノードにスケールするための、フェデレーションベースのアプローチについて説明します。提案されているアプローチは、大規模な（10〜100kノード）クラスタを、それぞれ独自のYARN RMと計算ノードを持つ、サブクラスタと呼ばれる小さな単位に分割することです。フェデレーションシステムは、これらのサブクラスタをつなぎ合わせて、アプリケーションに対して1つの大きなYARNクラスタとして表示します。この統合環境で実行されるアプリケーションは、単一の巨大なYARNクラスタを認識し、統合クラスタの任意のノードでタスクをスケジュールできます。内部的には、フェデレーションシステムはサブクラスタのリソースマネージャとネゴシエートして、アプリケーションにリソースを提供します。目標は、個々のジョブがサブクラスタをシームレスに「スパン」できるようにすることです。
この設計は、各RMが担当するノード数を制限し、適切なポリシーによって、ほとんどのアプリケーションが単一のサブクラスタ内に収まるようにすることで、構造的にスケーラブルです。そのため、各RMが認識するアプリケーションの数も制限されます。これは、サブクラスタを追加するだけで（サブクラスタ間の調整はほとんど必要ないため）、ほぼ線形にスケールできることを意味します。このアーキテクチャは、各サブクラスタ内でスケジューリングの不変条件を非常に厳密に適用できます（単純にYARNから継承されます）。一方、サブクラスタ間の継続的なリバランスにより、これらのプロパティがグローバルレベルでも（それほど厳密ではなく）尊重されるようにします（たとえば、サブクラスタが多数のノードを失った場合、キューを他のサブクラスタに再マッピングして、障害が発生したサブクラスタで実行されているユーザーが不当に影響を受けないようにすることができます）。
Federationは、既存のYARNコードベースの「レイヤー」として設計されており、コアYARNメカニズムの変更は限定的です。
前提条件
OSS YARNは約数千ノードまでスケールすることが知られています。提案されたアーキテクチャは、そのような小さなYARNクラスタ（サブクラスタと呼ばれる）を、数万ノードで構成されるより大きな統合YARNクラスタに統合するという概念を活用しています。この統合環境で実行されるアプリケーションは、単一の大きなYARNクラスタを認識し、クラスタ内の任意のノードでタスクをスケジュールできます。内部的には、フェデレーションシステムはサブクラスタRMとネゴシエートして、アプリケーションにリソースを提供します。図1の論理アーキテクチャは、統合クラスタを構成する主要コンポーネントを示しています。これらのコンポーネントについては、以下で説明します。
サブクラスタは、最大数千ノードのYARNクラスタです。サブクラスタの正確なサイズは、デプロイ/メンテナンスの容易さ、ネットワークまたは可用性ゾーンとの整合性、一般的なベストプラクティスを考慮して決定されます。
サブクラスタYARN RMは、作業を維持する高可用性が有効になっている状態で実行されます。つまり、YARN RM、NMの障害を最小限の中断で許容できる必要があります。サブクラスタ全体が危険にさらされた場合、外部メカニズムによってジョブが別のサブクラスタで再送信されることが保証されます（これは最終的にフェデレーション設計に含まれる可能性があります）。
サブクラスタは、統合環境のスケーラビリティ単位でもあります。1つ以上のサブクラスタを追加することで、統合環境をスケールアウトできます。
注：設計上、各サブクラスタは完全に機能するYARN RMであり、フェデレーションへの貢献度は全体的な容量のごく一部に設定できます。つまり、サブクラスタはフェデレーションに対して「部分的な」コミットメントを行いながら、容量の一部を完全にローカルな方法で提供する機能を保持できます。
YARNアプリケーションは、いずれかのルータに送信されます。ルータは、ルーティングポリシー（ポリシーストアから取得）を適用し、ステートストアにサブクラスタURLを照会し、アプリケーション送信リクエストを適切なサブクラスタRMにリダイレクトします。ジョブが開始されるサブクラスタを「ホームサブクラスタ」と呼び、ジョブがスパンする他のすべてのサブクラスタを「セカンダリサブクラスタ」と呼びます。ルータは、ApplicationClientProtocolを外部に公開し、複数のRMの存在を透過的に隠します。これを達成するために、ルータはアプリケーションとそのホームサブクラスタ間のマッピングをステートストアにも永続化します。これにより、ルータはソフトステートになりながら、ユーザーリクエストを安価にサポートできます。これは、どのルータもこのアプリケーションからホームサブクラスタへのマッピングを回復し、リクエストをブロードキャストすることなく適切なRMに送信できるためです。パフォーマンスのキャッシュとセッションのスティッキネスが推奨される場合があります。フェデレーションの状態（アプリケーションとノードを含む）は、Web UIを介して公開されます。
AMRMProxyは、アプリケーションがサブクラスタ全体でスケールおよび実行できるようにするための重要なコンポーネントです。AMRMProxyはすべてのNMマシンで実行され、ApplicationMasterProtocolを実装することにより、AMのYARN RMへのプロキシとして機能します。アプリケーションは、サブクラスタRMと直接通信することはできません。システムによって、AMRMProxyエンドポイントにのみ接続するように強制されます。AMRMProxyエンドポイントは、（通信を動的にルーティング/分割/マージすることにより）複数のYARN RMへの透過的なアクセスを提供します。いつでも、ジョブは1つのホームサブクラスタと複数のセカンダリサブクラスタにまたがることができますが、AMRMProxyで動作するポリシーは、スケジューリングインフラストラクチャのオーバーヘッドを最小限に抑えるために、各ジョブのフットプリントを制限しようとします（スケーラビリティ/負荷のセクションで詳しく説明します）。ARMMProxyのインターセプターチェーンアーキテクチャを図に示します。
AMRMProxyの役割
グローバルポリシージェネレーター（GPG）は、フェデレーション全体を監視し、システムが常に適切に構成および調整されていることを保証します。重要な設計ポイントは、クラスタの可用性が常時稼働のGPGに依存しないことです。GPGはすべてのクラスタ操作とは別に継続的に動作し、独自の視点を提供します。これにより、グローバルな不変条件を適用し、負荷分散に影響を与え、メンテナンスを受けるサブクラスタのドレイニングをトリガーするなどが可能です。より正確には、GPGはユーザー容量割り当てとサブクラスタのマッピングを更新し、ルーター、AMRMProxy（および場合によってはRM）で実行されるポリシーをまれに変更します。
GPGが利用できない場合、クラスタ操作はGPGが最後にポリシーを公開した時点の状態のまま続行されます。長期間の利用不能は、バランス、最適なクラスタ使用率、グローバルな不変条件などの望ましい特性の一部が失われる可能性がありますが、計算とデータへのアクセスは損なわれません。
注：現在の実装では、GPGは手動調整プロセスであり、CLIを介して公開されています（YARN-3657）。
フェデレーションシステムのこの部分は、YARN-5597の今後の作業の一部です。
フェデレーション状態は、複数の個々のサブクラスタを1つの大きなフェデレーションクラスタに疎結合するために維持する必要がある追加の状態を定義します。これには、以下の情報が含まれます。
メンバーYARN RMは、状態ストアに継続的にハートビートを送信して、稼働状態を維持し、現在の容量/負荷情報を公開します。この情報は、グローバルポリシージェネレーター（GPG）が適切なポリシー決定を行うために使用されます。また、この情報は、ルーターが最適なホームサブクラスタを選択するためにも使用できます。このメカニズムにより、サブクラスタを追加または削除することで「クラスタフリート」を動的に拡張/縮小できます。これにより、各サブクラスタのメンテナンスも容易になります。これはYARN RMに追加する必要がある新しい機能ですが、個々のYARN RM HAと同様であるため、メカニズムはよく理解されています。
アプリケーションマスター（AM）が実行されるサブクラスタは、アプリケーションの「ホームサブクラスタ」と呼ばれます。AMはホームサブクラスタのリソースに限定されず、セカンダリサブクラスタと呼ばれる他のサブクラスタのリソースも要求できます。フェデレーション環境は定期的に構成および調整されるため、AMがサブクラスタに配置されると、ホームサブクラスタ上のほとんどのリソースを見つけることができます。特定の場合にのみ、他のサブクラスタのリソースを要求する必要があります。
フェデレーションポリシーストアは、論理的に分離されたストア（同じ物理コンポーネントによってサポートされている場合がありますが）であり、アプリケーションとリソース要求が異なるサブクラスタにどのようにルーティングされるかについての情報が含まれています。現在の実装では、ランダム/ハッシュ/ラウンドロビン/プライオリティから、サブクラスタの負荷と要求の局所性のニーズを考慮した、より高度なものまで、いくつかのポリシーを提供しています。
アプリケーションが送信されると、システムはアプリケーションを実行するのに最も適切なサブクラスタを決定します。これをアプリケーションのホームサブクラスタと呼びます。AMからRMへのすべての通信は、AMマシンでローカルに実行されているAMRMProxyを介してプロキシされます。AMRMProxyは、YARN RMと同じApplicationMasterServiceプロトコルエンドポイントを公開します。AMは、ストレージレイヤーによって公開されている局所性情報を使用してコンテナを要求できます。理想的なケースでは、アプリケーションは、アプリケーションに必要なすべてのリソースとデータが利用可能なサブクラスタに配置されますが、他のサブクラスタのノードにコンテナが必要な場合は、AMRMProxyがそれらのサブクラスタのRMと透過的にネゴシエートし、リソースをアプリケーションに提供することにより、アプリケーションはフェデレーション環境全体を1つの巨大なYARNクラスタとして表示できます。AMRMProxy、グローバルポリシージェネレーター（GPG）、およびルーターは連携して、これをシームレスに実現します。
この図は、以下のジョブ実行フローのシーケンス図を示しています。
YARN が Federation を使用するように設定するには、conf/yarn-site.xml で次のプロパティを設定します。
これらは、フェデレーションの各マシンの conf/yarn-site.xml に表示される一般的な構成です。
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.federation.enabled |
true |
フェデレーションが有効かどうか |
yarn.resourcemanager.cluster-id |
<一意のサブクラスタID> |
このRMの一意のサブクラスタ識別子（HAに使用されるものと同じ）。 |
現在、ZooKeeperとSQLベースの状態ストアの実装をサポートしています。
注：状態ストアの実装は、常に以下のいずれかで上書きする必要があります。
ZooKeeper：HadoopのZooKeeper設定を行う必要があります
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.federation.state-store.class |
org.apache.hadoop.yarn.server.federation.store.impl.ZookeeperFederationStateStore |
使用する状態ストアのタイプ。 |
hadoop.zk.address |
ホスト:ポート |
ZooKeeperアンサンブルのアドレス。 |
SQL：次のパラメータを設定する必要があります
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.federation.state-store.class |
org.apache.hadoop.yarn.server.federation.store.impl.SQLFederationStateStore |
使用する状態ストアのタイプ。 |
yarn.federation.state-store.sql.url |
jdbc:mysql://<ホスト>:<ポート>/FederationStateStore |
SQLFederationStateStoreの場合、状態が格納されるDBの名前。 |
yarn.federation.state-store.sql.jdbc-class |
com.mysql.jdbc.jdbc2.optional.MysqlDataSource |
SQLFederationStateStoreの場合、使用するjdbcクラス。 |
yarn.federation.state-store.sql.username |
<dbuser> |
SQLFederationStateStoreの場合、DB接続のユーザー名。 |
yarn.federation.state-store.sql.password |
<dbpass> |
SQLFederationStateStoreの場合、DB接続のパスワード。 |
MySQLとMicrosoft SQL Serverのスクリプトを提供しています.
MySQLの場合、MVNリポジトリから最新のjarバージョン5.xをダウンロードし、CLASSPATHに追加する必要があります。次に、データベースで次のSQLスクリプトを実行することにより、DBスキーマが作成されます。
同じディレクトリに、ストアドプロシージャ、テーブル、ユーザー、およびデータベースを削除するためのスクリプトを提供しています。
注： FederationStateStoreUser.sqlは、DBのデフォルトのユーザー/パスワードを定義しています。適切な強力なパスワードに設定することを強くお勧めします。
SQL Serverの場合、プロセスは似ていますが、jdbcドライバーは既に含まれています。SQL Serverスクリプトは、sbin/FederationStateStore/SQLServer/ にあります。
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.federation.failover.enabled |
true |
各サブクラスタ内でRMフェイルオーバーを考慮して再試行するかどうか。 |
yarn.federation.blacklist-subclusters |
<サブクラスタID> |
ブラックリストに登録されたサブクラスタのリスト。サブクラスタを無効にするのに役立ちます。 |
yarn.federation.policy-manager |
org.apache.hadoop.yarn.server.federation.policies.manager.WeightedLocalityPolicyManager |
ポリシーマネージャーの選択により、アプリケーションとResourceRequestがシステムをどのようにルーティングされるかが決まります。 |
yarn.federation.policy-manager-params |
<バイナリ> |
ポリシーを設定するペイロード。この例では、ルーターとamrmproxyポリシーの重みのセット。これは通常、プログラムで構成されたpolicymanagerをシリアル化するか、状態ストアに.jsonシリアル化形式を入力することによって生成されます。 |
yarn.federation.subcluster-resolver.class |
org.apache.hadoop.yarn.server.federation.resolver.DefaultSubClusterResolverImpl |
ノードが属するサブクラスタと、ラックが属するサブクラスタを解決するために使用されるクラス。 |
yarn.federation.machine-list |
<マシンリストファイルのパス> |
SubClusterResolver が使用するマシンリストファイルのパス。ファイルの各行は、サブクラスタとラック情報を持つノードです。以下は例です。node1, subcluster1, rack1 node2, subcluster2, rack1 node3, subcluster3, rack2 node4, subcluster3, rack2 |
これらは、各ResourceManagerの conf/yarn-site.xml に表示される追加の構成です。
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.resourcemanager.epoch |
<一意のエポック> |
エポックのシード値。これは、異なるRMによって生成されたコンテナIDの一意性を保証するために使用されます。そのため、サブクラスタ間で一意であり、エポックを増分する障害に対応できるように「十分に間隔を空けて」おく必要があります。1000単位の増分により、多数のサブクラスタが可能になり、衝突の可能性がほぼゼロになります（衝突は、コンテナが1つのRMの1000回の再起動の間も存続し、次のRMが再起動せず、アプリがより多くのコンテナを要求した場合にのみ発生します）。 |
オプション
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.federation.state-store.heartbeat-interval-secs |
60 |
RMが中央の状態ストアにフェデレーションへのメンバーシップを報告する頻度。 |
これらは、各ルーターの conf/yarn-site.xml に表示される追加の構成です。
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.router.bind-host |
0.0.0.0 |
ルーターをバインドするホストIP。サーバーが実際にバインドするアドレス。このオプションのアドレスが設定されている場合、RPCサーバーとwebappサーバーは、それぞれこのアドレスとyarn.router.*.addressで指定されたポートにバインドされます。これは、0.0.0.0 に設定することで、ルーターをすべてのインターフェースでリッスンさせる場合に最も役立ちます。 |
yarn.router.clientrm.interceptor-class.pipeline |
org.apache.hadoop.yarn.server.router.clientrm.FederationClientInterceptor |
クライアントとインターフェースをとる際に、ルーターで実行されるインターセプタークラスのカンマ区切りリスト。このパイプラインの最後のステップは、Federation Client Interceptorである必要があります。 |
オプション
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.router.hostname |
0.0.0.0 |
ルーターのホスト名。 |
yarn.router.clientrm.address |
0.0.0.0:8050 |
ルーターのクライアントアドレス。 |
yarn.router.webapp.address |
0.0.0.0:8089 |
ルーターのウェブアプリケーションアドレス。 |
yarn.router.admin.address |
0.0.0.0:8052 |
ルーターの管理アドレス。 |
yarn.router.webapp.https.address |
0.0.0.0:8091 |
ルーターのセキュアウェブアプリケーションアドレス。 |
yarn.router.submit.retry |
3 |
諦めるまでにルーターで再試行する回数。 |
yarn.federation.statestore.max-connections |
10 |
各ルーターがステートストアに対して行う並列接続の最大数。 |
yarn.federation.cache-ttl.secs |
60 |
ルーターは情報をキャッシュし、これはキャッシュが無効になるまでの時間です。 |
yarn.router.webapp.interceptor-class.pipeline |
org.apache.hadoop.yarn.server.router.webapp.FederationInterceptorREST |
RESTインターフェースを介してクライアントとインターフェースをとる際に、ルーターで実行されるインターセプタークラスのカンマ区切りリスト。このパイプラインの最後のステップは、Federation Interceptor RESTである必要があります。 |
これらは、各ノードマネージャーの conf/yarn-site.xml に記載する必要がある追加設定です。
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.nodemanager.amrmproxy.enabled |
true |
AMRMProxyが有効かどうか。 |
yarn.nodemanager.amrmproxy.interceptor-class.pipeline |
org.apache.hadoop.yarn.server.nodemanager.amrmproxy.FederationInterceptor |
amrmproxyで実行されるインターセプターのカンマ区切りリスト。フェデレーションの場合、パイプラインの最後のステップはFederationInterceptorである必要があります。 |
オプション
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.nodemanager.amrmproxy.ha.enable |
true |
複数のアプリケーション試行をサポートするためにAMRMProxy HAが有効かどうか。 |
yarn.federation.statestore.max-connections |
1 |
各AMRMProxyからステートストアへの並列接続の最大数。多くのAMRMProxyが多数のDB接続をすぐに使い果たしてしまう可能性があるため、この値は通常、ルーターの値よりも低くなります。 |
yarn.federation.cache-ttl.secs |
300 |
AMRMProxyキャッシュの有効期限。AMRMProxyの数が多く、集中型ステートストアへの負荷を制限したいので、通常はルーターよりも大きくなります。 |
Federationクラスタにジョブを送信するには、ジョブが送信されるクライアント用に個別の設定セットを作成する必要があります。これらの設定では、conf/yarn-site.xml に次の追加設定が必要です。
| プロパティ | 例 | 説明 |
|---|---|---|
yarn.resourcemanager.address |
<router_host>:8050 |
クライアントで起動されたジョブをルーターのクライアントRMポートにリダイレクトします。 |
yarn.resourcemanager.scheduler.address |
localhost:8049 |
ジョブをフェデレーションAMRMProxyポートにリダイレクトします. |
クラスタのYARNジョブは、上記で説明したクライアント設定から送信できます。フェデレーションを通じてジョブを起動するには、最初にこちらで説明されているように、フェデレーションに関与するすべてのクラスタを起動します。次に、次のコマンドを使用して、ルーターマシンでルーターを起動します。
$HADOOP_HOME/bin/yarn --daemon start router
ここで、$HADOOP_CONF_DIRが上記で説明したクライアント設定フォルダを指している状態で、通常どおりにジョブを実行します。上記で説明したクライアント設定フォルダの設定は、ジョブをルーターのクライアントRMポートに転送します。ルーターは、起動後にこのポートでリッスンしているはずです。クライアントからフェデレーションクラスタでPiジョブを実行する例を次に示します。
$HADOOP_HOME/bin/yarn jar hadoop-mapreduce-examples-3.0.0.jar pi 16 1000
このジョブはルーターに送信されます。ルーターは、上記で説明したように、GPGから生成されたポリシーを使用して、ジョブの送信先となるホームRMを選択します。
この特定のサンプルジョブからの出力は、次のようになります。
2017-07-13 16:29:25,055 INFO mapreduce.Job: Job job_1499988226739_0001 running in uber mode : false 2017-07-13 16:29:25,056 INFO mapreduce.Job: map 0% reduce 0% 2017-07-13 16:29:33,131 INFO mapreduce.Job: map 38% reduce 0% 2017-07-13 16:29:39,176 INFO mapreduce.Job: map 75% reduce 0% 2017-07-13 16:29:45,217 INFO mapreduce.Job: map 94% reduce 0% 2017-07-13 16:29:46,228 INFO mapreduce.Job: map 100% reduce 100% 2017-07-13 16:29:46,235 INFO mapreduce.Job: Job job_1499988226739_0001 completed successfully . . . Job Finished in 30.586 seconds Estimated value of Pi is 3.14250000......
ジョブの状態は、routerhost:8089 のRouter Web UIでも追跡できます。フェデレーションを使用するために、コードの変更や入力jarの再コンパイルは必要ありません。また、このジョブの出力は、フェデレーションなしで実行した場合とまったく同じです。また、フェデレーションのメリットを最大限に活用するには、複数のクラスタが必要になるほど多数のマッパーを使用してください。上記の例では、その数は16です。