はじめに
Map・Shuffle・Reduce の流れをたどり、実際のローカル WordCount の結果を確認しました。今度は、小売業者が HDFS のファイルに同じ集計ルールを適用し、環境に用意されたリソース管理サービスを使いたいと考えています。業務上の疑問は同じまま、保存と実行の構成が変わります。
YARN はアプリケーションのリソースを管理します。ResourceManager はリソースの割り当てを調整し、NodeManager はワーカーマシンでの実行を管理します。投入された各アプリケーションには、作業を調整する ApplicationMaster があります。また、割り当てられたコンテナーで作業を実行します。コンテナーとは、プロセスを動かすためのリソースの割り当てです。YARN のコンテナーは、必ずしも Docker コンテナーではありません。
準備済みの WordCount JAR を投入し、実際の YARN アプリケーションを特定して、完了状態とログを調べ、HDFS の結果を確かめます。2回目は別の出力パスを使い、元の入力と最初の結果を残します。この実験でデータを準備し、サービスを起動するため、前の実験のファイルや Java のプログラミング知識は必要ありません。ターミナルを使ってください。
既習のアルゴリズムを YARN に投入する
このステップでは、HDFS に入力を準備し、設定済みの YARN フレームワークで同じ WordCount アルゴリズムを実行します。
ローカルの作業ディレクトリから始めます。
cd /home/labex/project
準備済みの入力を読みます。
cat /home/labex/project/data/product-words.txt
アルゴリズムの説明で使ったものと同じ6行です。期待される合計は引き続き folder=3、notebook=4、pen=4 です。HDFS の raw ディレクトリと、結果用の別の親ディレクトリを作ります。
hdfs dfs -mkdir -p /user/labex/yarn-batch/raw /user/labex/yarn-batch/results
入力をアップロードします。
hdfs dfs -put /home/labex/project/data/product-words.txt /user/labex/yarn-batch/raw/product-words.txt
すでに学んだように、-put はローカルの元ファイルを残します。HDFS の raw がジョブの入力です。過去の出力を新しい入力として読まないよう、results はその外に置きます。
サンプル JAR と wordcount の名前は、ローカル実行と同じです。今回はローカルモードの上書き設定を省きます。インストール済みの設定は mapreduce.framework.name=yarn と、既定の HDFS ファイルシステムを使います。そのため、以下の最後の2引数は両方とも HDFS パスです。ジョブ自身が出力ディレクトリ run-1 を作成するので、このディレクトリは事前に存在していてはいけません。
hadoop jar /opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.5.0.jar wordcount /user/labex/yarn-batch/raw /user/labex/yarn-batch/results/run-1
コマンドが終了するまで待ちます。投入されたアプリケーション ID、ジョブの進行状況、正常完了のメッセージが表示されます。アプリケーション ID、ジョブ ID、所要時間、進行状況の順番は変わります。短時間の RUNNING 状態を捉える必要はなく、完了済みのアプリケーションも調べられます。
コマンドと各役割を結び付けてみましょう。
クライアントがアプリケーションを投入 → ResourceManager がリソースを割り当てる
ApplicationMaster が MapReduce を調整 → NodeManager がコンテナーを実行する
タスクが HDFS 入力を読む → タスクが HDFS に結果を書く
引き続き、これは1台のマシンによる学習環境です。プロセスが同じマシンを共有していても、YARN の各役割は実際に動いています。次に、正常完了したことを示す直接の証拠を YARN に問い合わせます。
完了したアプリケーションを調べる
このステップでは、実際のアプリケーション ID を保持し、それを使って状態と集約されたログを問い合わせます。
YARN はアプリケーションを受け付けると、アプリケーション ID を割り当てます。HDFS パスや結果のファイル名とは別のものです。完了済みも含め、すべての状態のアプリケーションを一覧表示します。
yarn application -list -appStates ALL
この新しい環境で投入されたアプリケーションは1つです。種類は MAPREDUCE、状態は FINISHED、最終ステータスは SUCCEEDED のはずです。FINISHED だけでは成功を証明できず、最終ステータスも重要です。
例の値をコピーせず、実際のアプリケーション ID を保存します。次のパイプラインでは、grep -o が一致する部分だけを表示します。このパターンの [0-9]* は、アプリケーション ID 内の数字の並びに一致します。
yarn application -list -appStates ALL | grep -o 'application_[0-9]*_[0-9]*' > /home/labex/project/output/application-id.txt
取得した ID を読みます。
cat /home/labex/project/output/application-id.txt
application_ で始まる1行があるはずです。残りの部分はこの環境で割り当てられるので、ほかの学習者の ID を使わないでください。
保存した値をコマンドに入れるには、$(cat ...) で cat を実行し、その出力を代入します。二重引用符は、代入された値を1つの引数としてまとめます。実際のアプリケーションを問い合わせます。
yarn application -status "$(cat /home/labex/project/output/application-id.txt)"
State、Final-State、Application-Type を探します。それぞれ FINISHED、SUCCEEDED、MAPREDUCE が期待されます。レポートを保存します。
yarn application -status "$(cat /home/labex/project/output/application-id.txt)" > /home/labex/project/output/application-status.txt
YARN はコンテナーログも収集します。logs -applicationId は、選択したアプリケーションのログを読みます。読みやすい範囲を確認できるよう、ローカルに保存します。
yarn logs -applicationId "$(cat /home/labex/project/output/application-id.txt)" > /home/labex/project/output/application.log
ローカルの tail -n 20 で末尾の20行を表示します。
tail -n 20 /home/labex/project/output/application.log
正確なログメッセージは変わります。コンテナーの見出しとタスクのメッセージから、アプリケーションと YARN が実際に実行した作業を結び付けられます。ログは動作の説明に役立ちますが、安心できそうな1行のログより、成功状態と正しい業務上の出力を合わせたほうが強い証拠になります。次の実験では、同じ診断ツールを失敗時に使います。
入力に出力を混ぜずに照合して再実行する
このステップでは、結果データを確認してエクスポートし、2つ目の出力パスに再実行します。
最初の結果ディレクトリを一覧表示します。
hdfs dfs -ls /user/labex/yarn-batch/results/run-1
_SUCCESS マーカーは出力の完了を示します。集計は part-* ファイルに入っています。前と同じように、引用符付きの HDFS パターンを使い、記録を並べ替えて安定した表示にします。
hdfs dfs -cat '/user/labex/yarn-batch/results/run-1/part-*' | sort
期待される結果は次のとおりです。
folder 3
notebook 4
pen 4
読み戻しの方法で、並べ替えたローカルのエクスポートを保存します。
hdfs dfs -cat '/user/labex/yarn-batch/results/run-1/part-*' | sort > /home/labex/project/output/run-1-counts.txt
次の実行前に、同じ入力で再度実行すると集計が変わるか予想してください。変わらないはずです。新しい出力パスを使えば、最初の結果を保持し、出力がすでに存在するエラーも避けられます。
hadoop jar /opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.5.0.jar wordcount /user/labex/yarn-batch/raw /user/labex/yarn-batch/results/run-2
正常完了したら、2回目の集計を並べ替えて読み、保存します。
hdfs dfs -cat '/user/labex/yarn-batch/results/run-2/part-*' | sort > /home/labex/project/output/run-2-counts.txt
ローカルにエクスポートした2つを比較します。
diff /home/labex/project/output/run-1-counts.txt /home/labex/project/output/run-2-counts.txt
差分はないはずです。結果は raw 入力ディレクトリの外に並んでいるため、再実行しても前の出力が次の入力バッチに混ざりませんでした。アルゴリズムを変えずに、実際のアプリケーション、ログ、結果データ、再現性を確認できました。
まとめ
既習の集計アルゴリズムを、ローカル実行から HDFS と YARN に移しました。実際の MapReduce アプリケーションを特定し、正常完了とコンテナーログを調べ、2つの出力パスで正しい結果を確認しました。
アルゴリズムは同じまま、ストレージとリソース管理の構成が変わりました。次は、クライアント側の問題と投入済みアプリケーションの失敗を区別し、適切な最小限の修復を行う方法を学びます。
