LabEx
登録

YARNのアーキテクチャと開発

LinuxBeginner
オンラインで実践に進む

はじめに

Hadoop 2.0以降では、新しいリソース管理方式であるYARNが導入されました。これにより、クラスターのリソース利用率が向上し、リソースを一元管理してデータを共有できるようになりました。この実験では、Hadoopの疑似分散クラスターを構築した状態から、YARNフレームワークのアーキテクチャ、動作原理、設定、開発、監視の方法を学びます。

この実験では、Javaプログラミングの基礎知識が必要です。

開発手順では、完成したJavaソースファイルを使用します。各ファイルの内容をすべてエディターに貼り付け、クライアントによるアプリケーションの送信方法と、ApplicationMasterによるタスクコンテナの要求・起動方法を確認してください。

YARNのアーキテクチャとコンポーネント

このステップでは、YARNのアーキテクチャと各コンポーネントの役割を確認します。

Hadoop 0.23でMapReduce 2.0（MRv2）の一部として導入されたYARNは、Hadoopクラスターにおけるリソース管理とジョブスケジューリングを大きく変えました。

  • JobTrackerの分割: MRv2では、JobTrackerの機能を個別のデーモンに分割しています。リソース管理はResourceManagerが担当し、ジョブのスケジューリングと監視はApplicationMasterが担当します。
  • グローバルなResourceManager: アプリケーションごとに対応するApplicationMasterがあります。アプリケーションはMapReduceジョブの場合もあれば、ジョブを表すDAGの場合もあります。
  • データ計算フレームワーク: ResourceManager、Slave、NodeManagerで構成されるフレームワークで、ResourceManagerがすべてのアプリケーションのリソースを管理します。
  • ResourceManagerのコンポーネント: Schedulerは容量やキューなどの制約に基づいてリソースを割り当てます。ApplicationsManagerはジョブの送信とApplicationMasterの実行を管理します。
  • リソースの割り当て: リソース要件は、メモリ、CPU、ディスク、ネットワークなどの要素を含むリソースコンテナを使って定義します。
  • NodeManagerの役割: NodeManagerはコンテナのリソース使用状況を監視し、ResourceManagerとSchedulerに報告します。
  • ApplicationMasterのタスク: ApplicationMasterはSchedulerとリソースコンテナについて交渉し、状態を追跡して進行状況を監視します。

次の図に、各コンポーネントの関係を示します。

YARN architecture components diagram

YARNは以前のバージョンとのAPI互換性を保っているため、実行中のMapReduceタスクをシームレスに移行できます。Hadoopクラスターで効率的にリソースを管理し、ジョブをスケジューリングするには、YARNのアーキテクチャと各コンポーネントの役割を理解することが重要です。

Hadoopデーモンの起動

このステップでは、YARNアプリケーションの実行に必要なHadoopデーモンを起動します。

関連する設定パラメーターやYARNアプリケーションの開発方法を学ぶ前に、必要なときに利用できるようHadoopデーモンを起動します。

まず、デスクトップ上のXfceターミナルをダブルクリックして開き、次のコマンドを入力してhadoopユーザーに切り替えます。

su - hadoop

ヒント: hadoopユーザーのパスワードは「hadoop」です。

ユーザーの切り替えが完了したら、HDFSやYARNフレームワークなど、Hadoop関連のデーモンを起動できます。

ターミナルで次のコマンドを実行して、デーモンを起動してください。

/home/hadoop/hadoop/sbin/start-dfs.sh
/home/hadoop/hadoop/sbin/start-yarn.sh

起動後、必要に応じてjpsコマンドを実行し、関連するデーモンが動作しているか確認できます。

hadoop:~$ jps
3378 NodeManager
3028 SecondaryNameNode
3717 Jps
2791 DataNode
2648 NameNode
3240 ResourceManager

設定ファイルの準備

このステップでは、Hadoopの主要な設定ファイルの1つであるyarn-site.xmlを確認し、このファイルでYARNクラスターにどのような設定を指定できるかを学びます。

設定ファイルを誤って変更しないように、Hadoopの設定ファイルを別のディレクトリにコピーしてから開きます。

まず、ターミナルで次のコマンドを実行し、設定ファイル用のディレクトリを作成します。

mkdir /home/hadoop/hadoop_conf

次に、YARNの主要な設定ファイルyarn-site.xmlをインストールディレクトリから新しく作成したディレクトリにコピーします。

ターミナルで次のコマンドを実行してください。

cp /home/hadoop/hadoop/etc/hadoop/yarn-site.xml /home/hadoop/hadoop_conf/yarn-site.xml

続いて、vimエディターでファイルを開き、内容を確認します。

vim /home/hadoop/hadoop_conf/yarn-site.xml

設定ファイルの仕組み

このステップでは、実行中のHadoopクラスターで使われている設定パラメーターを確認します。

YARNフレームワークには、ResourceManagerとNodeManagerという2つの重要な役割があります。そのため、ファイル内の各設定項目は、この2つのコンポーネントに関する設定です。

このファイルには多くの設定項目を指定できますが、デフォルトでは、カスタム設定項目は含まれていません。たとえば、現在開いているファイルには、以前に疑似分散Hadoopクラスターを設定したときに指定したaux-services属性だけが含まれています。次の図に示すとおりです。

hadoop:~$ cat /home/hadoop/hadoop/etc/hadoop/mapred-site.xml

...
<configuration>
    <property>
        <name>mapreduce.framework.name</name>
        <value>yarn</value>
    </property>
</configuration>

この設定項目は、NodeManager上で実行する必要がある依存サービスを指定します。指定した設定値はmapreduce_shuffleで、MapReduceプログラムをYARN上で実行するために必要なデフォルト値を示します。

ファイルに記述されていない設定項目は機能しないのでしょうか。そうではありません。設定パラメーターがファイルに明示的に指定されていない場合、HadoopのYARNフレームワークは内部ファイルに保存されたデフォルト値を読み込みます。yarn-site.xmlファイルに明示的に指定した設定項目はすべてデフォルト値を上書きします。この仕組みにより、Hadoopシステムをさまざまな利用環境に適応させることができます。

ResourceManagerの設定項目

Hadoopクラスターで効率的にリソースを管理し、ジョブを実行するには、yarn-site.xmlファイルのResourceManager関連設定を理解し、適切に設定することが重要です。主な設定項目は次のとおりです。

  • yarn.resourcemanager.address: アプリケーションの送信や強制終了に使用する、クライアント向けのアドレスです。デフォルトのポートは8032です。
  • yarn.resourcemanager.scheduler.address: リソースの要求や解放に使用する、ApplicationMaster向けのアドレスです。デフォルトのポートは8030です。
  • yarn.resourcemanager.resource-tracker.address: ハートビートの送信やタスクの取得に使用する、NodeManager向けのアドレスです。デフォルトのポートは8031です。
  • yarn.resourcemanager.admin.address: 管理コマンドに使用する、管理者向けのアドレスです。デフォルトのポートは8033です。
  • yarn.resourcemanager.webapp.address: クラスター情報を確認するWeb UIのアドレスです。デフォルトのポートは8088です。
  • yarn.resourcemanager.scheduler.class: スケジューラーのメインクラス名を指定します（例: FIFO、CapacityScheduler、FairScheduler）。
  • スレッドの設定:
    • yarn.resourcemanager.resource-tracker.client.thread-count
    • yarn.resourcemanager.scheduler.client.thread-count
  • リソースの割り当て:
    • yarn.scheduler.minimum-allocation-mb
    • yarn.scheduler.maximum-allocation-mb
    • yarn.scheduler.minimum-allocation-vcores
    • yarn.scheduler.maximum-allocation-vcores
  • NodeManagerの管理:
    • yarn.resourcemanager.nodes.exclude-path
    • yarn.resourcemanager.nodes.include-path
  • ハートビートの設定:
    • yarn.resourcemanager.nodemanagers.heartbeat-interval-ms

これらのパラメーターを設定すると、Hadoopクラスター内のResourceManagerの動作、リソース割り当て、スレッド処理、NodeManagerの管理、ハートビート間隔を細かく調整できます。各設定項目を理解しておくと、問題の防止につながり、クラスターを安定して運用できます。

NodeManagerの設定項目

Hadoopクラスター内のリソースとタスクを効率的に管理するには、yarn-site.xmlファイルのNodeManager関連設定が重要です。主な設定項目は次のとおりです。

  • yarn.nodemanager.resource.memory-mb: NodeManagerが利用できる物理メモリの合計を指定します。この値はYARNの実行中に変化しません。
  • yarn.nodemanager.vmem-pmem-ratio: 仮想メモリと物理メモリの割り当て比率を設定します。デフォルト値は2.1です。
  • yarn.nodemanager.resource.cpu-vcores: NodeManagerが利用できる仮想CPUの合計数を指定します。デフォルト値は8です。
  • yarn.nodemanager.local-dirs: NodeManager上で中間結果を保存するパスです。複数のディレクトリを設定できます。
  • yarn.nodemanager.log-dirs: NodeManagerのログディレクトリのパスです。複数のディレクトリを設定できます。
  • yarn.nodemanager.log.retain-seconds: NodeManagerのログを保持する最大時間です。デフォルトは10800秒（3時間）です。

これらのパラメーターを設定すると、HadoopクラスターでNodeManagerがリソースを効率よく利用できるよう、リソース割り当て、メモリ管理、ディレクトリパス、ログの保持期間を細かく調整できます。各設定項目を理解しておくと、クラスターを安定して運用し、タスクを効率的に実行できます。

設定項目の確認とデフォルト値の参照

YARNやその他の一般的なHadoopコンポーネントで利用できるすべての設定項目を確認するには、Apache Hadoopが提供するデフォルト設定ファイルを参照してください。デフォルト設定ファイルは次のリンクから確認できます。

これらのデフォルト設定を確認すると、各設定項目の詳細な説明と用途が分かります。各パラメーターがHadoopのアーキテクチャ設計でどのような役割を果たすかを理解するのに役立ちます。

設定内容の確認が終わったら、vimエディターを閉じて、Hadoopの設定確認を終了します。

プロジェクトのディレクトリとファイルの作成

このステップでは、アプリケーションのソースファイルを作成します。クライアントがApplicationMasterを送信し、ApplicationMasterがタスクコンテナを1つ要求して挨拶文を出力する、最小構成のYARNアプリケーションを完成させます。

まず、プロジェクト用のディレクトリを作成します。ターミナルで次のコマンドを実行してください。

mkdir /home/hadoop/yarn_app

次に、プロジェクト内にソースコードファイルを2つ作成します。

最初のファイルはClient.javaです。ターミナルでtouchコマンドを実行して作成します。

touch /home/hadoop/yarn_app/Client.java

続いて、ApplicationMaster.javaファイルを作成します。

touch /home/hadoop/yarn_app/ApplicationMaster.java

hadoop:~$ tree /home/hadoop/yarn_app/
/home/hadoop/yarn_app/
├── ApplicationMaster.java
└── Client.java

0 directories, 2 files

クライアントコードの作成

このステップでは、アプリケーションを送信するクライアントを完成させます。hadoopユーザーのまま、次のコマンドでソースファイルを開きます。

vim /home/hadoop/yarn_app/Client.java

package宣言とimport文を含め、ファイル全体を以下のコードに置き換えます。Vimで:set pasteと入力してEnterキーを押してから、iを押して挿入モードにします。ファイル全体を貼り付け、すべての行が表示されるまで待ちます。その後、Escを押し、:wqと入力して保存し、終了します。

package com.labex.yarn.app;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.ConverterUtils;

public class Client {
    public static void main(String[] args) throws Exception {
        if (args.length != 1) {
            throw new IllegalArgumentException("Usage: Client /absolute/path/to/yarn-app.jar");
        }
        YarnConfiguration conf = new YarnConfiguration();
        YarnClient client = YarnClient.createYarnClient();
        client.init(conf);
        client.start();
        try {
            YarnClientApplication application = client.createApplication();
            ApplicationSubmissionContext context = application.getApplicationSubmissionContext();
            ApplicationId id = context.getApplicationId();
            FileSystem fs = FileSystem.get(conf);
            Path destination = new Path(fs.getHomeDirectory(), "yarn-app/" + id + "/app.jar");
            fs.mkdirs(destination.getParent());
            fs.copyFromLocalFile(new Path(args[0]), destination);
            FileStatus status = fs.getFileStatus(destination);
            LocalResource jar = LocalResource.newInstance(
                ConverterUtils.getYarnUrlFromPath(fs.makeQualified(destination)),
                LocalResourceType.FILE, LocalResourceVisibility.APPLICATION,
                status.getLen(), status.getModificationTime());
            Map<String, LocalResource> resources = new HashMap<>();
            resources.put("app.jar", jar);
            Map<String, String> environment = new HashMap<>();
            // The single-node lab uses the same Hadoop installation in every container.
            environment.put("CLASSPATH", "./app.jar:" + System.getProperty("java.class.path"));
            String command = ApplicationConstants.Environment.JAVA_HOME.$$()
                + "/bin/java -Xmx128m com.labex.yarn.app.ApplicationMaster"
                + " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout"
                + " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr";
            ContainerLaunchContext launch = ContainerLaunchContext.newInstance(
                resources, environment, Collections.singletonList(command), null, null, null);
            context.setApplicationName("LabEx YARN Hello");
            context.setQueue("default");
            context.setResource(Resource.newInstance(256, 1));
            context.setAMContainerSpec(launch);
            client.submitApplication(context);
            System.out.println("Application ID: " + id);
            long deadline = System.currentTimeMillis() + 180000;
            while (System.currentTimeMillis() < deadline) {
                ApplicationReport report = client.getApplicationReport(id);
                YarnApplicationState state = report.getYarnApplicationState();
                if (state == YarnApplicationState.FINISHED
                    || state == YarnApplicationState.FAILED
                    || state == YarnApplicationState.KILLED) {
                    System.out.println("Final status: " + report.getFinalApplicationStatus());
                    if (report.getFinalApplicationStatus() != FinalApplicationStatus.SUCCEEDED) {
                        throw new IllegalStateException(report.getDiagnostics());
                    }
                    return;
                }
                Thread.sleep(1000);
            }
            client.killApplication(id);
            throw new IllegalStateException("Application timed out after three minutes");
        } finally {
            client.stop();
        }
    }
}

YarnClientはResourceManagerに接続し、アプリケーションIDを取得します。クライアントはJARをHDFSにコピーし、ローカルリソースapp.jarとして宣言します。これにより、YARNはそのJARをApplicationMasterの作業ディレクトリに配置できます。起動コンテキストにはクラスパスとJavaコマンドを指定し、送信コンテキストにはキューとコンテナのリソースを指定します。

クラスパスには、この単一ノードの実験環境で各コンテナが共有する、インストール済みのHadoopライブラリを使用します。MavenやGradleによるダウンロードは必要ありません。この例は、実験環境のKerberosを使用しないクラスターを対象としています。

送信後、クライアントはアプリケーションレポートをポーリングし、終了状態になるまで待ってから最終ステータスを確認します。killApplicationは、アプリケーションが3分のタイムアウトを超えた場合にのみ使用します。最後のステップで、両方のクラスをコンパイルして実行します。

ApplicationMasterコードの作成

このステップでは、ApplicationMasterを完成させます。ResourceManagerに登録し、コンテナを1つ要求して、NodeManager経由で挨拶文を出力するタスクを起動し、その結果を報告します。

hadoopユーザーのまま、次のコマンドでファイルを開きます。

vim /home/hadoop/yarn_app/ApplicationMaster.java

ファイル全体を以下のコードに置き換えます。:set pasteと入力してEnterキーを押してから、iを押します。ファイル全体を貼り付け、すべての行が表示されるまで待ちます。その後、Escを押し、:wqと入力して保存し、終了します。

package com.labex.yarn.app;

import java.util.Collections;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.NMClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;

public class ApplicationMaster {
    public static void main(String[] args) throws Exception {
        YarnConfiguration conf = new YarnConfiguration();
        AMRMClient<AMRMClient.ContainerRequest> rm = AMRMClient.createAMRMClient();
        NMClient nm = NMClient.createNMClient();
        rm.init(conf);
        nm.init(conf);
        rm.start();
        nm.start();
        try {
            rm.registerApplicationMaster("", 0, "");
            AMRMClient.ContainerRequest request = new AMRMClient.ContainerRequest(
                Resource.newInstance(256, 1), null, null, Priority.newInstance(0));
            rm.addContainerRequest(request);
            boolean launched = false;
            long deadline = System.currentTimeMillis() + 120000;
            while (System.currentTimeMillis() < deadline) {
                // Each allocate call also sends a heartbeat to the ResourceManager.
                AllocateResponse response = rm.allocate(launched ? 0.5f : 0.0f);
                for (Container container : response.getAllocatedContainers()) {
                    if (launched) {
                        rm.releaseAssignedContainer(container.getId());
                        continue;
                    }
                    rm.removeContainerRequest(request);
                    String command = "/bin/echo Hello-from-YARN"
                        + " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout"
                        + " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr";
                    ContainerLaunchContext launch = ContainerLaunchContext.newInstance(
                        Collections.emptyMap(), Collections.emptyMap(),
                        Collections.singletonList(command), null, null, null);
                    nm.startContainer(container, launch);
                    launched = true;
                }
                for (ContainerStatus status : response.getCompletedContainersStatuses()) {
                    if (status.getExitStatus() != 0) {
                        throw new IllegalStateException(status.getDiagnostics());
                    }
                    rm.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED,
                        "Hello container completed", "");
                    return;
                }
                Thread.sleep(1000);
            }
            rm.unregisterApplicationMaster(FinalApplicationStatus.FAILED,
                "No successful container within two minutes", "");
            throw new IllegalStateException("Container timed out");
        } finally {
            nm.stop();
            rm.stop();
        }
    }
}

AMRMClientは登録とリソース要求を処理します。allocateを呼び出すたびにハートビートが送信され、新しく割り当てられたコンテナと完了したコンテナの状態が返されます。NMClientは割り当てられたコンテナ内でコマンドを実行します。この同期ポーリングループを使うことで、未定義のコールバッククラスやヘルパーメソッドを使わずに、例を完結させています。

タスクはコンテナログにHello-from-YARNを出力します。終了ステータスが0の場合、ApplicationMasterはSUCCEEDEDを指定して登録を解除します。その後、クライアントが最終ステータスを報告します。ResourceManagerは要求された256 MBを最小割り当て量まで切り上げることがあります。割り当てを待つ間、ループはハートビートを送り続けます。

アプリケーション起動の流れ

このステップでは、作成した2つのJavaクラスをコンパイルして実行し、ResourceManagerのWebインターフェースでアプリケーションを確認します。ターミナルでhadoopユーザーのまま作業を続けてください。

アプリケーションのコンパイルと起動

ソースディレクトリに移動します。

cd /home/hadoop/yarn_app

出力ディレクトリを作成します。

mkdir -p classes

VMにインストール済みのHadoopライブラリを使って、2つのソースファイルをコンパイルします。デフォルトのコンパイラーが新しいバージョンであっても、--release 8を指定すると、HadoopのJava 8ランタイムと互換性のあるクラスが生成されます。

javac --release 8 -cp "$(/home/hadoop/hadoop/bin/hadoop classpath --glob)" -d classes Client.java ApplicationMaster.java

非推奨APIに関する注意が表示されても問題ありません。コンパイルエラーがないことを確認してください。コンパイルしたクラスをパッケージ化します。

jar cf yarn-app.jar -C classes .

クライアントを実行し、出力を保存します。最後の引数には、クライアントがApplicationMaster用にHDFSへアップロードするJARを指定します。

set -o pipefail
/home/hadoop/hadoop/bin/yarn jar /home/hadoop/yarn_app/yarn-app.jar com.labex.yarn.app.Client /home/hadoop/yarn_app/yarn-app.jar | tee /home/hadoop/yarn_app/application.log

アプリケーションが完了するまで待ちます。Hadoopのログメッセージの中に、次の出力が表示されます。

Application ID: application_<timestamp>_<sequence>
Final status: SUCCEEDED

アプリケーションIDは実行するたびに変わります。挨拶文はタスクコンテナのログに書き込まれ、クライアントはアプリケーションIDと最終結果を出力します。この実行では、別のビルド済みMapReduceの例ではなく、自分で作成したクライアントとApplicationMasterが動作します。

アプリケーションの実行結果を確認する

デスクトップでFirefoxを開き、次のURLにアクセスします。

http://localhost:8088

アプリケーション一覧からLabEx YARN Helloを探します。ターミナルに表示されたアプリケーションIDを使って見つけてください。状態がFINISHED、最終ステータスがSUCCEEDEDになっていることを確認します。アプリケーションIDをクリックすると、詳細を確認できます。ResourceManagerはアプリケーションを追跡し、ApplicationMasterはタスクコンテナを管理し、NodeManagerはそのコンテナを実行します。

まとめ

Hadoopの疑似分散クラスターを構築した状態から、この実験ではYARNフレームワークのアーキテクチャ、動作原理、設定、開発、監視の方法を学びました。実験には多くのコードと設定ファイルが含まれているため、内容をよく確認してください。

関連 Linux コース
Linux のクイックスタート

Linux のクイックスタート

linux
ジュニアシステムアドミニストレーターになる

ジュニアシステムアドミニストレーターになる

linuxshell
初心者のための Linux

初心者のための Linux

linux