はじめに
小売業者の商品ファイルが整理され、アクセス可能になりました。これで最初の実験の疑問、「どの商品ラベルが最も多く登場するか？」に答えられます。数行なら手で数えられますが、増え続けるデータには、繰り返し実行できる計算が必要です。
MapReduce はバッチ計算を構成する方法です。Map は入力の記録をキーと値のペアに変換し、Shuffle は同じキーの値を集め、Reduce はキーごとの値をまとめて結果にします。WordCount では、キーは商品ラベルで、1回の出現につき値1を与えます。単語集計はデータの流れを見える形にするための小さな例であり、完全な売上分析システムではありません。
用意された学習用ツールで各段階を調べた後、同じ入力に対して、コンパイル済みの実際の Hadoop WordCount プログラムを実行します。学習用ツールは1つのプロセスでアルゴリズムの段階を表示するもので、分散実行の記録ではありません。Hadoop はローカルモードで実行し、計算に集中します。次の実験では、同じアルゴリズムを YARN に移します。Java や Python のプログラミングは必要ありません。
商品ラベルをキーと値のペアにする
このステップでは、入力を確認し、商品の各出現が Map を通る様子をたどります。
ローカルの作業ディレクトリから始めます。
cd /home/labex/project
入力の6行を読みます。
cat /home/labex/project/data/product-words.txt
期待されるテキストは次のとおりです。
notebook pen
notebook folder
pen pen
folder notebook
notebook
folder pen
ここで一度止まり、商品を手で数えてみてください。notebook は4つの行に1回ずつ登場します。pen は、3行目の2回も含め、合計4回です。folder は3回です。プログラムも同じ合計を出すはずです。
キーと値のペアは、ラベルと値を対応させたものです。このアルゴリズムでは、Map は出現するたびに (product, 1) を出力します。pen が2回あれば、入力行につき1ペアではなく、別々の2ペアを作ります。
用意された実行ファイル wordcount-stages は、段階の名前とローカルの入力ファイルを受け取ります。map の表示を使います。
/home/labex/project/tools/wordcount-stages map /home/labex/project/data/product-words.txt
出力は次のように始まります。
notebook 1
pen 1
notebook 1
folder 1
キーと値の間はタブです。商品の出現1つにつき1つ、合計11ペアがあります。このツールはアルゴリズムを見える形にするもので、Hadoop サービスの起動やタスクのスケジュールは行いません。この段階の出力を保存します。
/home/labex/project/tools/wordcount-stages map /home/labex/project/data/product-words.txt > /home/labex/project/output/map-pairs.txt
入力の表現が変わりました。行全体が、商品別にまとめられる個々の加算分になりました。
加算分をグループ化して集計する
このステップでは、ペアが Shuffle と Reduce を通る様子をたどり、出力を手作業の予想と結び付けます。
Map は個別の加算分を出力します。ある商品の合計を求めるには、その商品への加算分を集める必要があります。Shuffle と Sort は、Reducer が特定の商品の値を受け取れるよう、キーごとにペアを整理します。分散ジョブではタスク間のデータ転送を伴う場合がありますが、学習用ツールはグループ化した値を表示するだけです。
グループ化した表示を調べます。
/home/labex/project/tools/wordcount-stages shuffle /home/labex/project/data/product-words.txt
期待されるグループ化後の値は次のとおりです。
folder 1,1,1
notebook 1,1,1,1
pen 1,1,1,1
キーは並べ替えられています。カンマで区切られた各1は、新しい記録ではなく、元の出現を表したままです。グループ化の結果を保存します。
/home/labex/project/tools/wordcount-stages shuffle /home/labex/project/data/product-words.txt > /home/labex/project/output/shuffled-values.txt
Reduce はキーごとの値をまとめます。ここでは整数を足します。別のアプリケーションでは異なるまとめ方もでき、MapReduce は集計だけに限られません。
Reduce の表示を確認します。
/home/labex/project/tools/wordcount-stages reduce /home/labex/project/data/product-words.txt
期待される結果は次のとおりです。
folder 3
notebook 4
pen 4
ステップ1で手で数えた結果と比べてください。次に実際の Hadoop アプリケーションと比較できるよう、結果を保存します。
/home/labex/project/tools/wordcount-stages reduce /home/labex/project/data/product-words.txt > /home/labex/project/output/teaching-counts.txt
元の出現が別々の行にあっても、結果は各キーのすべての出現によって決まります。だからこそ、Map と Reduce の間にグループ化の段階があるのです。
実際の Hadoop WordCount プログラムを実行する
このステップでは、コンパイル済みの Hadoop サンプルで、同じ集計をローカルモードで実行します。
JAR は Java のプログラムコードをまとめたものです。Hadoop とサンプル JAR はインストール済みなので、コードをコンパイルする必要はありません。hadoop jar は指定した JAR を実行し、wordcount はその WordCount サンプルを選びます。以下の2つの -D 設定は、この実行でローカルの実行フレームワークとローカルファイルシステムを選択します。最後の2つの引数は、入力ファイルと新しい出力ディレクトリです。
hadoop jar /opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.5.0.jar wordcount -Dmapreduce.framework.name=local -Dfs.defaultFS=file:/// /home/labex/project/data/product-words.txt /home/labex/project/output/local-wordcount
プロンプトが戻るまで待ちます。Hadoop は進行状況と診断メッセージを出力しますが、パス、識別子、所要時間は変わります。これは YARN アプリケーションではなく、ローカルランナーを使った実際の Hadoop プログラムです。この実験では HDFS や YARN のサービスを起動せず、入力と出力は両方ともローカルです。
ローカルの出力ディレクトリを一覧表示します。
ls /home/labex/project/output/local-wordcount
_SUCCESS マーカーと、part-... の結果ファイルがあるはずです。_SUCCESS は出力が正常に完了したことを示し、結果ファイルには確認するデータが入っています。ローカルのシェルの part-* パターンで結果ファイルを選びます。比較を安定させるため、連結した内容を sort に渡します。
cat /home/labex/project/output/local-wordcount/part-* | sort > /home/labex/project/output/hadoop-counts.txt
学習用ツールの集計と比較します。
diff /home/labex/project/output/teaching-counts.txt /home/labex/project/output/hadoop-counts.txt
差分はないはずです。説明で学んだアルゴリズムと、実際のプログラムの出力を結び付けられました。この実行の出力ディレクトリがすでにある場合は、完成したファイルを確認してください。むやみに削除したり、再投入すれば上書きされると考えたりしてはいけません。
後で重要になる違いも押さえましょう。HDFS のブロックはバイト列を保存し、入力スプリットは入力の作業単位を定義し、タスクは作業を実行します。これらは関係していますが、常に一対一に対応するわけではありません。結果ファイルは処理タスクの出力であり、元の保存ブロックのコピーではありません。
入力の変更を1つ予想して試す
このステップでは、作業用コピーに商品を1回追加し、どの結果が変わるか確かめます。
読み取り専用の元の入力は変更しません。ローカルの cat と > で、新しい書き込み可能な作業用コピーを作ります。両方のパスはローカルで、変更するのは出力側のファイルです。
cat /home/labex/project/data/product-words.txt > /home/labex/project/output/changed-product-words.txt
notebook の出現を1つ追記した場合の新しい集計を予想してください。その合計だけが4から5に増えるはずです。printf は指定したテキストを表示し、\n は改行を意味します。>> は既存の内容を置き換えず、出力を末尾に追加します。
printf 'notebook\n' >> /home/labex/project/output/changed-product-words.txt
変更した入力を確認します。
cat /home/labex/project/output/changed-product-words.txt
末尾に新しい1行があるはずです。入力と出力のパスだけを変え、同じ Hadoop プログラムを実行します。
hadoop jar /opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.5.0.jar wordcount -Dmapreduce.framework.name=local -Dfs.defaultFS=file:/// /home/labex/project/output/changed-product-words.txt /home/labex/project/output/changed-wordcount
実際の結果を並べ替えて読みます。
cat /home/labex/project/output/changed-wordcount/part-* | sort
期待される結果は次のとおりです。
folder 3
notebook 5
pen 4
新しい Map のペアは notebook グループに入るので、その Reduce の合計だけが変わります。これで、コマンドが成功したことを認識するだけでなく、結果を説明できます。次の実験では、アルゴリズムと集計ルールをそのままに、HDFS が入力を提供し、YARN が実行リソースを管理する構成に進みます。
まとめ
商品の出現を Map のペアに変え、Shuffle で値をグループ化し、Reduce でまとめる流れをたどりました。実際の Hadoop WordCount は手計算と学習用ツールに一致し、入力に出現を1つ加えると、予想どおりに結果が変わりました。
アルゴリズムだけに集中するため、ローカルランナーを使いました。次はアルゴリズムをそのままに、実際の YARN アプリケーションが HDFS の入力を読み、HDFS に結果を出力する仕組みを学びます。
