今回は、halookでHiveがどのように実行されるかを見ていきます。なお、前回の連載までと異なり、今回は、halookをCDH3ではなくCDH4に適用した結果を記載しています。
Hiveとは？
Hiveは、HDFS上のデータをHiveQLと呼ばれるSQLライクなクエリ言語で処理できるようにしたもので、CDHにも含まれています。HiveQLで記述した命令が、自動でMapReduceに変換されるため、ユーザ自身がMapReduceジョブプログラムを記述する必要がなく、データ解析が簡便に行えます。
Hiveを使うためには、まずデータを、Hiveテーブルとして保存する必要があります。カンマ区切りや、タブ区切りで保存されたデータに対して、カラム名や型などのテーブル定義を示すことによって、Hiveテーブルを作成できます。HiveQLの構文はSQLと若干異なるところがあるものの、多くのSQL構文をサポートしているため、テーブルを作ってしまえば、RDBMSと同様の感覚で操作することができます。
また、Hiveの実行には、同じくCDHに含まれている、Hueを使うのが便利です。Hueは、Hadoop用のWeb UIで、HDFSのファイル操作など、ブラウザ上で様々な処理を実行することができます。Hueを使うと、HDFS上へのデータのアップロード、Hiveテーブルの作成、HiveQLの実行、実行結果や実行履歴の確認まで、すべてブラウザ上で行うことができます。
図1 HiveQLを実行する、Hueのクエリエディタの画面
HiveQLの実行例
ためしにHiveQLを実行してみましょう。
SELECT * FROM member m JOIN item i ON m.id = i.member_id
上記は、memberテーブルとitemテーブルを、memberテーブルのidカラムとitemテーブルのmember_idカラムでJOINする例です。これが、内部で自動的にMapReduceジョブに変換されます。
実行後、halookで変換されたMapReduceジョブを視覚化したところ以下のようになりました。
図2 JOIN実行時のMapReduceジョブ
図中の緑のラインが、時系列で見た際のMap処理の実行期間を表し、青のラインがReduce処理を表します。これにより、HiveQLが2つのMap処理と1つのReduce処理に変換して実行されたことと、Map/Reduce処理がserver0、server1、server2の3台で分散して実行されていることがわかります。また、2つのMap処理の実行時間には2倍程度の開きがあることがわかります。なお、一般的に、Mapの数やReduceの数は、データ量や、テーブルに設定されたパーティションの有無などで変わります。
次に、先ほどのクエリを少し変えて実行してみます。
SELECT
m.id, m.name, i.item
FROM
member m JOIN item i
ON
m.id = i.member_id
ORDER BY
m.id
大きな違いは、最後にORDER BY句を入れたことです。このORDER BY句の影響で、実行されるMapReduce Jobが2つに増えました。
図3 ORDER BY句を加えたJOIN実行時のMapReduceジョブ Stage-1
図4 ORDER BY句を加えたJOIN実行時のMapReduceジョブ Stage-2
このように、MapReduceジョブ1回で実行できない内容のクエリの場合、Stage-1、Stage-2、Stage-3・・・という形で、複数のMapReduceジョブが順番に実行されます。
同じ処理でもHiveQLの書き方で実行時間が変わることがある
SQLと同様、HiveQLにおいても、記述上のテクニックが存在します。同じ処理を行うHiveQLであっても、書き方によって、変換されるMapReduceジョブが変わり、実行時間に差が出ることがあります。
以下にその例を挙げます。
クエリ1
SELECT count(DISTINCT col_0) FROM table0001
クエリ2
SELECT
count(*)
FROM (
SELECT
DISTINCT
col_0
FROM
table0001
) t
クエリ1と2は、どちらも、table0001テーブルの、col_0カラムの重複を除いた要素数を数えるものですが、書き方が異なります。
実行した結果を、halookのジョブ一覧画面で確認したところ、以下のようになりました。
図5 クエリ1と2の実行結果
これは、MapReduceジョブの実行期間を、時系列に表示したビューです。クエリ1のほうは1つのMapReduceジョブで実行されていますが、クエリ2のほうは、副問い合わせを使って書かれているため、MapReduceジョブが2つに分かれています。ですが、2つのジョブに分かれて実行されているのにもかかわらずそれらを合わせても、クエリ2の実行時間の方が短くなっていることがわかります。クエリ1の実行時間は9分59秒、クエリ2の実行時間は8分41秒です。
なぜこのようなことになったのか、それぞれのジョブを見てみるとよくわかります。
図6 クエリ1のMapReduceジョブ
図7 クエリ2のMapReduceジョブ Stage-1
図8 クエリ2のMapReduceジョブ Stage-2
まず図6を見ることで、クエリ1のMapReduceジョブでは、Reduceタスクが1つしか動いていないことがわかります。これが処理時間を遅らせていた原因です。クエリ1のような書き方だと、重複を除いた個数のカウント処理が、1つのReducerでしか実行されないのです。
一方、図7を見ると、複数のReduceタスクが並列して走っていることが分かります。3本だけ長いReduceタスクがありますが、これは、Mapタスクが全て完了する前から先行して一部のReducerが開始しているためで、Reduceタスクの実質的な処理時間は、短い矢印の長さとなります。図8に示した、クエリ2のStage-2のMapReduceタスクは、非常に短い時間で完了する単純なタスクです。
クエリ1のMapReduceタスクは、halookのジョブ閲覧画面で、「Bubble Chart」タブを表示することでも問題点が視覚化されます。
図9 クエリ1のMapReduceジョブの、タスク実行時間のばらつき
このグラフは、横軸がタスクの開始時間で、縦軸が各タスクの実行時間となっています。緑の点で示した、Mapタスクの実行時間は、一定程度に収まっているのですが、青い点で示したただ1つのReduceタスクに、飛びぬけて長い時間がかかっていることがわかるでしょう。このReduceタスクが分散して処理されるようにすることで、処理時間が短縮されます。
まとめ
Hiveは、RDBMSのSQL文を実行するのと同じ感覚でHDFS上のデータを処理できる便利な機能です。ですが、裏でMapReduceジョブが動いていることに変わりはありません。
今回は例として、同じ処理を行う2つのクエリ実行を試しました。3ノードで実行してもある程度の差が出ましたが、対象となるデータ量と、ノード数が増えると、クエリ1とクエリ2の実行時間の差はさらに顕著になります。大規模な計算になればなるほど、MapReduceジョブが分散して実行されているかどうかがより重要になるのです。クエリの実行が遅いかな、と感じたら、どんなMapReduceジョブが実際に実行されているのか、確認してみるとよいでしょう。
