amazon EMRを使ってみた

2013年05月15 23時57分

amazon EMRを使ってみた

amazon EMR(Elastic Map Reduce)を使ってみたメモ。普段使っているのはClouderaのCDH4。

質問は何でもどうぞ！

この記事を見て分からないところがありましたら、@ts_3156までお気軽にご質問ください。未経験者が迷うポイントを知りたいので、質問大歓迎です！(^^)

今回はhiveのみ使います

hadoopとその周りのツールには、pigとかmahoutとか色々ありますが、今回はhiveのみ使います。

EMRはjob flowという単位でhadoopクラスタを起動しますが、hiveの普通のCUIも使えます。mysqlコマンドでログインしてクエリを実行するのとほとんど同じです。

hiveを使うと、SQLっぽいクエリだけで色々な集計ができるようになります。普通の人がhadoopを使いたい時、おそらく9割方はhiveだけで解決できます。

カスタムMap/ReduceとかHadoop Streamingとか他にも色々ありますが、どれもかなり敷居が高いと思うので、まずhiveだけ使ってみることを強くおすすめします(^^)

EMRを使ってhadoopクラスタを起動する

EMRの起動自体はブラウザからクリックしていくだけで本当に簡単です。「Amazon Management Console」から「Amazon Elastic Mapreduce」を選んで後は流れにそってください。

下記の二箇所だけ、選択肢を間違えないでください。

「Hive Program」を選ぶこと。

「Start an Interactive Hive Session」を選ぶこと。

インスタンスの種類とhadoopクラスタの構成台数は、ちょっと試してみるだけなら適当で大丈夫です。

実運用の際は扱うデータの大きさでインスタンスの種類と構成台数を変更してください。むちゃくちゃ大雑把な目安としては、数百GB以上のデータを扱うなら「CPU16コア、メモリ20GBのインスタンスを数十台くらい」みたいな感じだとそれなりに快適になります。

ちなみにCDHをインストールするなら「Cloudera Managerを使って、hadoop、hive等をEC2の複数台のサーバーにインストールする方法」の手順通りにやればOKです。

hadoopクラスタにsshでログインする

普通のEC2インスタンスと同じようにログインすればいいのですが、直接hadoopクラスタ(のmasterノード)にログインするのではなく、別のEC2インスタンス経由でのログインをおすすめします。

何故かと言うと、hadoopクラスタには最近ならデフォルトで揃っている便利ツール(yumとかllコマンドとか)が全くないためです。

普通のEC2インスタンスからscreen(またはtmux)経由でログインするのが一番手軽だと思います。

# hadoopクラスタにログインするためのkey pairは事前にアップロードしておくこと。

# hadoopクラスタにログインするためのEC2インスタンス上で実行する。
chmod 600 key_pair.pem
ssh -i key_login.pem hadoop@[hadoopクラスタのpublic DNS名]

上記のsshコマンドでhadoopクラスタにログインできます。ユーザー名は「hadoop」で固定です。

hiveでテーブルを作成しデータをロードしてみる

ログインができたら、hiveのクエリを実行してみます。基本的にはmysqlと同じです。

cat 2013-04-01.csv
2013-04-01,pc,1
2013-04-01,pc,2
2013-04-01,pc,3
2013-04-01,pc,4
2013-04-01,pc,5
2013-04-01,pc,6
2013-04-01,pc,7
2013-04-01,pc,8

上記のcsvファイルを事前に準備しておいてください。このcsvデータをhiveテーブルにロードします。

# コマンドラインで実行してください
hive

# ここからは、hiveクライアント上で実行してください
# hive> って書いてあったらそれがhiveクライアントです

CREATE DATABASE log;

use log;

DROP TABLE IF EXISTS log.access_log;
CREATE TABLE log.access_log (
  dt STRING,
  device STRING,
  user_id STRING
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ","
STORED AS TEXTFILE;

LOAD DATA
LOCAL INPATH '/home/hadoop/2013-04-01.csv'
OVERWRITE INTO TABLE log.access_log
;

上記のクエリにより、csvデータがhiveテーブルにロードされました。

hiveクライアントの起動とかに10秒くらいかかるのは正常です。そういうものだと思っておいてください(^^)

hiveでデータを集計してみる

下記のクエリで集計できます。

SELECT
  dt,
  device,
  count(1),
  count(distinct user_id)
FROM
  log.access_log
WHERE
  dt = '2013-04-01'

GROUP BY
  dt,
  device
;

集計結果は下記の通りです。

MapReduce Total cumulative CPU time: 3 seconds 980 msec
Ended Job = job_201305151255_0002
Counters:
MapReduce Jobs Launched:
Job 0: Map: 1  Reduce: 1   Accumulative CPU: 3.98 sec   HDFS Read: 364 HDFS Write: 18 SUCCESS
Total MapReduce CPU Time Spent: 3 seconds 980 msec
OK
2013-04-01      pc      8       8
Time taken: 57.038 seconds

1分くらいかかっています。この結果から分かる通り、データが少なくてもhiveはむっちゃ遅いです。もしgrepでどうにかなる程度のデータなら、普通にgrep、もしくはmysqlで集計してしまうことをおすすめします。

ちなみにhive.exec.parallelはデフォルトでtrueのようです。

ちょっと高度なhiveクエリ

パーティションで区切ったり、viewを作ったりするクエリもサンプルとして載せておきます。

DROP TABLE IF EXISTS log.access_log;
CREATE TABLE log.access_log (
  device STRING,
  user_id STRING
)
PARTITIONED BY (dt STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ","
STORED AS TEXTFILE;


LOAD DATA
LOCAL INPATH '/aaa/bbb/2013-04-01.csv.gz'
OVERWRITE INTO TABLE log.access_log
PARTITION (dt='2013-04-01')

use log;
ALTER TABLE access_log DROP IF EXISTS PARTITION (dt='2013-04-01');

INSERT OVERWRITE TABLE log.access_log PARTITION(dt='2013-04-01')


use log;

CREATE VIEW IF NOT EXISTS access_log_pvuu
AS SELECT
  dt,
  device,
  count(1) pv,
  count(distinct user_id) uu
FROM
  log.access_log
WHERE
  dt = '2013-04-01'

GROUP BY
  dt,
  device
;