ラベル emr の投稿を表示しています。 すべての投稿を表示
ラベル emr の投稿を表示しています。 すべての投稿を表示

2014年2月5日水曜日

EMRってなんじゃ?(ImpalaでCloudfrontの爆速ログ集計)




EMRがImpalaをサポートするようになりました。ImpalaはClouderaが提供するオープンソースのクエリエンジンで、Hiveより断然速いそうです。

例として、Cloudfrontのログを、タイムスタンプをJSTに直して1時間ごとのアクセス数の集計をしてみます。


ログバケットの確認


まずCloudFrontのログが以下のS3にたまっているとします。
s3://memorycraft-impala-input/cf/logs





EMRクラスタの起動


次に、EMRクラスタを起動します。
EMRのダッシュボードで「Create Cluster」をクリックし、新規クラスタ作成画面を表示します。

Cluster Configuration

Cluster nameに適当なクラスタ名を入力します。また、今回はEMRのログは出力しないのでLoggingのチェックはOFFなんかにしておきます。起動したインスタンスの名前をつける場合は、TagsのNameとしてインスタンス名をつけておきます。



Software Configuration

Hadoopのバージョンとアプリケーションのリストを設定します。
Applications to be installedのリストにはデフォルトでHiveとPigが追加されていますが、
AMI versionは現時点で最新の3.0,2を選ぶち、下のAdditional applicationsのドロップダウンでImpalaが選択できるようになります。
Impalaを選択し、「Configure and add」をクリックして追加します。



Hardware Configuration

ここでVPCやゾーンやインスタンス数を設定します。
ここでは、EC2-Classicで、ゾーンはA、インスタンス数はデフォルトにしてみます。



Security and Access

EC2 key pairで、hadoopインスタンスに接続するときのキーペア名を設定します。
ここが未定義だと、インスタンスに接続できないので注意が必要です。



Bootstrap Actions & Steps

ここでは、起動した後の挙動や起動パラメータなどを定義します。
今回は未設定のままでOKです。
最後に「Create cluster」ボタンをクリックして、クラスタを作成します。





EMRのインスタンスへの接続


起動完了するとクラスタ詳細画面で、Master public DNSに、マスタインスタンスのPublicDNSが表示されます。


そのPublicDNSに対して、先ほど選択した鍵を使ってhadoopユーザーでSSH接続します。
$ ssh -i ~/.ssh/keys/memorycraft.pem hadoop@ec2-54-199-41-72.ap-northeast-1.compute.amazonaws.com
The authenticity of host 'ec2-54-199-41-72.ap-northeast-1.compute.amazonaws.com (54.199.41.72)' can't be established.
RSA key fingerprint is 0f:ac:d5:8d:50:2b:7c:92:6a:ad:74:5e:f3:d1:52:05.
Are you sure you want to continue connecting (yes/no)? yes
Warning: Permanently added 'ec2-54-199-41-72.ap-northeast-1.compute.amazonaws.com,54.199.41.72' (RSA) to the list of known hosts.

       __|  __|_  )
       _|  (     /   Amazon Linux AMI
      ___|\___|___|

https://aws.amazon.com/amazon-linux-ami/2013.09-release-notes/
13 package(s) needed for security, out of 46 available
Run "sudo yum update" to apply all updates.
--------------------------------------------------------------------------------

Welcome to Amazon Elastic MapReduce running Hadoop and Amazon Linux.

Hadoop is installed in /home/hadoop. Log files are in /mnt/var/log/hadoop. Check
/mnt/var/log/hadoop/steps for diagnosing step failures.

The Hadoop UI can be accessed via the following commands:

  ResourceManager    lynx http://localhost:9026/
  NameNode           lynx http://localhost:9101/

--------------------------------------------------------------------------------
$
接続が完了しました。



S3データのコピー


現状、ImpalaでExternalテーブルにS3のパスを直接指定するとエラーになってしまいました。方法がわからなかったため、S3DistCpでHDFS上の/input/にコピーしてきます。
$ vim ./import.sh
#!/bin/bash

. /home/hadoop/impala/conf/impala.conf

hadoop jar /home/hadoop/share/hadoop/common/lib/EmrS3DistCp-1.0.jar -Dmapreduce.job.reduces=30 --src s3://memorycraft-impala-input/cf/logs/ --dest hdfs://$HADOOP_NAMENODE_HOST:$HADOOP_NAMENODE_PORT/input/ --outputCodec 'none'

$ chmod 755 ./import.sh
$ ./import.sh
14/02/02 20:12:41 INFO s3distcp.S3DistCp: Running with args: -Dmapreduce.job.reduces=30 --src s3://memorycraft-impala-input/cf/logs/ --dest hdfs://172.31.0.95:9000/input/ --outputCodec none
14/02/02 20:12:41 INFO s3distcp.S3DistCp: S3DistCp args: --src s3://memorycraft-impala-input/cf/logs/ --dest hdfs://172.31.0.95:9000/input/ --outputCodec none
14/02/02 20:12:45 INFO s3distcp.S3DistCp: Using output path 'hdfs:/tmp/8f7379de-1ebf-4f31-be50-bd17aa54f2d5/output'
14/02/02 20:12:46 INFO s3distcp.S3DistCp: GET http://169.254.169.254/latest/meta-data/placement/availability-zone result: ap-northeast-1a
14/02/02 20:12:46 INFO s3distcp.S3DistCp: Created AmazonS3Client with conf KeyId XXXXXXXXXXXXXX
〜(中略)〜
  Combine output records=0
  WRONG_LENGTH=0
  WRONG_MAP=0
  WRONG_REDUCE=0
 File Input Format Counters
  Bytes Read=3406
 File Output Format Counters
  Bytes Written=0
14/02/02 20:13:37 INFO s3distcp.S3DistCp: Try to recursively delete hdfs:/tmp/8f7379de-1ebf-4f31-be50-bd17aa54f2d5/tempspace

コピーの実行が終わりました。HDFSの内容を見ると、コピーされているのが分かります。
$ hadoop fs -ls /input/
Found 20 items
-rw-r--r--   1 hadoop supergroup       1728 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.21fcnUK5
-rw-r--r--   1 hadoop supergroup       1726 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.9jiEU8sU
-rw-r--r--   1 hadoop supergroup       1477 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.9pD0UZAy
-rw-r--r--   1 hadoop supergroup       4996 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.CECpH1q4
-rw-r--r--   1 hadoop supergroup       3636 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.DmeAD298
-rw-r--r--   1 hadoop supergroup        979 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.GhmYMPjI
-rw-r--r--   1 hadoop supergroup       1915 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.IvAE5n9h
-rw-r--r--   1 hadoop supergroup       1853 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.K3Crm40n
-rw-r--r--   1 hadoop supergroup       6120 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.KILmQ81g
-rw-r--r--   1 hadoop supergroup        969 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.KdRXIw4s
-rw-r--r--   1 hadoop supergroup       1731 2014-02-02 20:13 /input/E2KNLXPGJLJRRQ.2014-01-30-07.MIAnZGc4
.....



Impalaの実行


Impalaはimpala-shellというコマンドで専用のプロンプトを起動して使います。hiveのhiveコマンドと同じようなものです。
$ impala-shell
Starting Impala Shell without Kerberos authentication
Connected to ip-10-146-59-193.ap-northeast-1.compute.internal:21000
Server version: impalad version 1.2.1 RELEASE (build 8c1da7709727f3545974009a4bb677a0004024ec)
Welcome to the Impala shell. Press TAB twice to see a list of available commands.

Copyright (c) 2012 Cloudera, Inc. All rights reserved.

(Shell build version: Impala Shell v1.2.1 (8c1da77) built on Sun Dec  1 20:57:24 PST 2013)
[ip-10-146-59-193.ap-northeast-1.compute.internal:21000] >

まずは、入力テーブルをつくります。ファイルの場所はS3DistCpのコピー先に指定した/input/を指定します。
[ip-10-146-59-193.ap-northeast-1.compute.internal:21000] >
CREATE EXTERNAL TABLE IF NOT EXISTS input (
  cf_date STRING,
  cf_time STRING,
  x_edge_location STRING,
  sc_bytes INT,
  c_ip STRING,
  cs_method STRING,
  cs_host STRING,
  cs_uri_stem STRING,
  sc_status STRING,
  cs_referrer STRING, 
  cs_user_agent STRING,
  cs_uri_query STRING,
  cs_cookie STRING,
  x_edge_result_type STRING,
  x_edge_request_id STRING,
  x_host_header STRING,
  cs_protocol STRING,
  cs_bytes INT
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
LOCATION '/input/';

それでは、クエリを実行してみます。
[ip-10-146-59-193.ap-northeast-1.compute.internal:21000] >
SELECT
  w.jsttime,
  count(mydata)
FROM
  (
    SELECT
      SUBSTR(
        CAST(
          FROM_UTC_TIMESTAMP(
            FROM_UNIXTIME(
              UNIX_TIMESTAMP(CONCAT(cf_date, " ", cf_time), 'yyyy-MM-dd HH:mm:ss')
            ), 'JST'
          ) as STRING
        ), 1, 13
      ) AS jsttime,
      'AAA' AS mydata 
    FROM
      input
    WHERE
      cf_date NOT LIKE '#%'
  ) w
GROUP BY
  w.jsttime
;
Query: select w.jsttime, count(mydata) from (select substr(cast(from_utc_timestamp(from_unixtime(unix_timestamp(concat(cf_date, " ", cf_time), 'yyyy-MM-dd HH:mm:ss')), 'JST') as STRING), 1, 13) as jsttime, 'AAA' as mydata  from input where cf_date not like '#%') w group by w.jsttime
+---------------+---------------+
| jsttime       | count(mydata) |
+---------------+---------------+
| 2014-01-30 16 | 95            |
| 2014-01-30 17 | 33            |
+---------------+---------------+
Returned 2 row(s) in 0.88s
1秒かかりませんでした。



Hiveとの速度比較


これと同じ集計をHiveで行ってみます。
Total MapReduce jobs = 1
Launching Job 1 out of 1
Number of reduce tasks not specified. Estimated from input data size: 1
In order to change the average load for a reducer (in bytes):
  set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
  set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
  set mapred.reduce.tasks=<number>
Starting Job = job_1391362544864_0002, Tracking URL = http://10.132.128.116:9046/proxy/application_1391362544864_0002/
Kill Command = /home/hadoop/bin/hadoop job  -kill job_1391362544864_0002
Hadoop job information for Stage-1: number of mappers: 2; number of reducers: 1
2014-02-02 17:46:18,013 Stage-1 map = 0%,  reduce = 0%
2014-02-02 17:46:29,536 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:30,585 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:31,646 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:32,707 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:33,773 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:34,822 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:35,872 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:36,925 Stage-1 map = 100%,  reduce = 0%, Cumulative CPU 4.92 sec
2014-02-02 17:46:37,975 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 6.85 sec
2014-02-02 17:46:39,034 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 6.85 sec
2014-02-02 17:46:40,083 Stage-1 map = 100%,  reduce = 100%, Cumulative CPU 6.85 sec
MapReduce Total cumulative CPU time: 6 seconds 850 msec
Ended Job = job_1391362544864_0002
Counters:
MapReduce Jobs Launched:
Job 0: Map: 2  Reduce: 1   Cumulative CPU: 6.85 sec   HDFS Read: 58636 HDFS Write: 34 SUCCESS
Total MapReduce CPU Time Spent: 6 seconds 850 msec
OK
2014-01-30 16 95
2014-01-30 17 33
Time taken: 40.334 seconds, Fetched: 2 row(s)
40秒も掛かりました。いかにImpalaが速いかがわかります。



8億レコードでやってみた


上と全く同じ内容のクエリを使って、8億レコードのCloudFrontログ解析をImpala上で実行してみました。

データコピー(S3→HDFS)

$ . /home/hadoop/impala/conf/impala.conf
$
$ hadoop jar /home/hadoop/share/hadoop/common/lib/EmrS3DistCp-1.0.jar -Dmapreduce.job.reduces=30 --src s3://cloudfront-big-log/logs/cf/ --dest hdfs://$HADOOP_NAMENODE_HOST:$HADOOP_NAMENODE_PORT/input/ --outputCodec 'none'
14/02/03 05:21:10 INFO s3distcp.S3DistCp: Running with args: -Dmapreduce.job.reduces=30 --src s3://cloudfront-big-log/logs/cf/ --dest hdfs://172.31.0.95:9000/input/ --outputCodec none
14/02/03 05:21:10 INFO s3distcp.S3DistCp: S3DistCp args: --src s3://cloudfront-big-log/logs/cf/ --dest hdfs://172.31.0.95:9000/input/ --outputCodec none
14/02/03 05:21:14 INFO s3distcp.S3DistCp: Using output path 'hdfs:/tmp/9dc925c6-0f4e-4043-a19a-f21930f89b6b/output'
14/02/03 05:21:15 INFO s3distcp.S3DistCp: GET http://169.254.169.254/latest/meta-data/placement/availability-zone result: ap-northeast-1a
14/02/03 05:21:15 INFO s3distcp.S3DistCp: Created AmazonS3Client with conf KeyId XXXXXXXXXXXXXXXXXXXXXXXX
14/02/03 05:21:16 INFO s3distcp.S3DistCp: Skipping key 'logs/cf/' because it ends with '/'
14/02/03 05:21:16 INFO s3distcp.FileInfoListing: Opening new file: hdfs:/tmp/9dc925c6-0f4e-4043-a19a-f21930f89b6b/files/1
14/02/03 05:21:57 INFO s3distcp.S3DistCp: Created 1 files to copy 211761 files
14/02/03 05:21:57 INFO s3distcp.S3DistCp: Reducer number: 29
〜(中略)〜
  CPU time spent (ms)=16164160
  Physical memory (bytes) snapshot=18180935680
  Virtual memory (bytes) snapshot=53996851200
  Total committed heap usage (bytes)=14206107648
 Shuffle Errors
  BAD_ID=0
  CONNECTION=0
  IO_ERROR=0
  WRONG_LENGTH=0
  WRONG_MAP=0
  WRONG_REDUCE=0
 File Input Format Counters
  Bytes Read=39560891
 File Output Format Counters
  Bytes Written=0
14/02/03 06:30:50 INFO s3distcp.S3DistCp: Try to recursively delete hdfs:/tmp/9dc925c6-0f4e-4043-a19a-f21930f89b6b/tempspace


入力テーブル作成(input)

$ impala-shell
>
> CREATE EXTERNAL TABLE IF NOT EXISTS input (
>   cf_date STRING,
>   cf_time STRING,
>   x_edge_location STRING,
>   sc_bytes INT,
>   c_ip STRING,
>   cs_method STRING,
>   cs_host STRING,
>   cs_uri_stem STRING,
>   sc_status STRING,
>   cs_referrer STRING,
>   cs_user_agent STRING,
>   cs_uri_query STRING,
>   cs_cookie STRING,
>   x_edge_result_type STRING,
>   x_edge_request_id STRING,
>   x_host_header STRING,
>   cs_protocol STRING,
>   cs_bytes INT
> )
> ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
> LOCATION '/input/';
Query: create EXTERNAL TABLE IF NOT EXISTS input ( cf_date STRING, cf_time STRING, x_edge_location STRING, sc_bytes INT, c_ip STRING, cs_method STRING, cs_host STRING, cs_uri_stem STRING, sc_status STRING, cs_referrer STRING, cs_user_agent STRING, cs_uri_query STRING, cs_cookie STRING, x_edge_result_type STRING, x_edge_request_id STRING, x_host_header STRING, cs_protocol STRING, cs_bytes INT ) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION '/input/'

Returned 0 row(s) in 266.06s


入力件数

> select count(*) from input;
Query: select count(*) from input

+-----------+
| count(*)  |
+-----------+
| 820545673 |
+-----------+

Returned 1 row(s) in 642.03s


出力テーブル作成(output)

> CREATE EXTERNAL TABLE IF NOT EXISTS output_pv (
>   dth STRING,
>   cnt BIGINT
> )
> ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
> LOCATION '/output_pv/';
Query: create EXTERNAL TABLE IF NOT EXISTS output_pv ( dth STRING, cnt BIGINT ) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LOCATION '/output_pv/'

Returned 0 row(s) in 2.23s


解析(input→output)

> insert into output
> select w.jsttime, count(mydata) from
> (select substr(cast(from_utc_timestamp(from_unixtime(unix_timestamp(concat(cf_date, " ", cf_time), 'yyyy-MM-dd HH:mm:ss')), 'JST') as STRING), 1, 13) as jsttime, 'AAA' as mydata  from input where cf_date not like '#%') w
> group by w.jsttime
> ;
Query: insert into output select w.jsttime, count(mydata) from (select substr(cast(from_utc_timestamp(from_unixtime(unix_timestamp(concat(cf_date, " ", cf_time), 'yyyy-MM-dd HH:mm:ss')), 'JST') as STRING), 1, 13) as jsttime, 'AAA' as mydata  from input where cf_date not like '#%') w group by w.jsttime

Inserted 759 rows in 14940.46s



結果

データのコピー(S3→HDFS)約1時間
テーブル作成(input)約4分
テーブル作成(output)約2秒
解析(input→output)約4時間

想像していたより全然速いです。


ちなみに、特定のURIに絞ってクエリを実行したら15分ほどで終わってしまいました。
> insert into output_pv
> select w.jsttime, count(mydata) from
> (select substr(cast(from_utc_timestamp(from_unixtime(unix_timestamp(concat(cf_date, " ", cf_time), 'yyyy-MM-dd HH:mm:ss')), 'JST') as STRING), 1, 13) as jsttime, 'AAA' as mydata  from input where cf_date not like '#%' and (cs_uri_stem = '/' or cs_uri_stem = '/index.html')) w
> group by w.jsttime
> ;
Query: insert into output_pv select w.jsttime, count(mydata) from (select substr(cast(from_utc_timestamp(from_unixtime(unix_timestamp(concat(cf_date, " ", cf_time), 'yyyy-MM-dd HH:mm:ss')), 'JST') as STRING), 1, 13) as jsttime, 'AAA' as mydata  from input where cf_date not like '#%' and (cs_uri_stem = '/' or cs_uri_stem = '/index.html')) w group by w.jsttime

Inserted 755 rows in 852.34s

今回はHiveで8億レコードは試せませんでしたが、いままでHiveを使用していた感覚からするとImpalaは異常なほどの速さです。
これからEMRでクエリを使用する場合はImpala一択になりそうです。

以上です。



2013年4月17日水曜日

EMRってなんじゃ?(別アカウントのS3に入出力)

EMRをつかって、別のアカウントのS3バケットにアクセスしたい時があります。

ここでは例として、アカウントAのEMRからアカウントBのS3のログを集計して、アカウントBのS3バケットへ出力してみます。

まず、アカウントBのS3バケットのACLを設定します。
方法は以前の記事と同じようにSDKで設定します。
S3ってなんじゃ?(S3のログファイルを別アカウントでダウンロード


$src = 'memorycraft-log';//入力ログバケット
$target = 'memorycraft-archive';//出力先バケット
$owner_canonical_id = 'オーナー(アカウントB)の標準ユーザーID';
$other_canonical_id = '別アカウント(アカウントA)の標準ユーザーID';
$s3 = new AmazonS3(array('key'=>'オーナーのアクセスキー','secret'=>'オーナーのシークレットキー');
$s3->set_region(AmazonS3::REGION_APAC_NE1);

$res = $s3->set_bucket_acl($src, array(
  array('id' => AmazonS3::USERS_LOGGING,     'permission' => AmazonS3::GRANT_READ_ACP),
  array('id' => AmazonS3::USERS_LOGGING,     'permission' => AmazonS3::GRANT_WRITE),
  array('id' => $other_canonical_id,         'permission' => AmazonS3::GRANT_FULL_CONTROL),
  array('id' => $owner_canonical_id,         'permission' => AmazonS3::GRANT_FULL_CONTROL),
));

$res = $s3->set_bucket_acl($target, array(
  array('id' => $other_canonical_id,         'permission' => AmazonS3::GRANT_FULL_CONTROL),
  array('id' => $owner_canonical_id,         'permission' => AmazonS3::GRANT_FULL_CONTROL),
));



このままの状態でEMR(Hive)で出力すると出力されたファイルは以下のようになります。





パーミッションがありません。
これは出力するファイル自体のパーミッションの設定がおかしいためです。
これに権限を追加するにはHiveの場合は、HiveScriptの先頭に以下のコマンドを追加します。

set fs.s3.canned.acl=BucketOwnerFullControl;

再度実行すると、出力されたファイルにも権限が与えられているのがわかります。





この説明に関しては、最近日本語化されたEMRの公式ドキュメントにPigやカスタムJARの場合の対応法とともに詳しく記載されています。
http://docs.aws.amazon.com/ja_jp/ElasticMapReduce/latest/DeveloperGuide/emr-s3-acls.html

これで、アカウントをまたいだリソース集計が可能になります。
以上です。

2013年4月15日月曜日

EMRってなんじゃ?(HiveでS3のログをJSTに変換して1日分をまとめる)

S3のwebホスティングで、ログ出力の設定をしていた場合、ログファイルが大量に出力されます。
ログの記録時間は標準時で出力されていてわかりづらいです。

今回はEMRのHiveを利用して、日本時間の0時〜翌日の0時までのログを1ファイルにまとめてみたいと思います。


HiveScript



HiveScriptは以下のようにします。
まずフォーマット解析用のJARを読み込みます。

add jar /home/hadoop/hive/contrib/hive-contrib-0.8.1.jar;

入力用のテーブルを定義します。LOCATIONは引数から受け取ります。
ネックとなるのは、S3のログは基本的に半角スペース区切りですが、項目の値の中に半角スペースが入り込むため、「FIELDS TERMINATED BY」句ではなく、SERDEを使用して正規表現でログの1行を項目分けします。
今回は日付フィールドはそのまま年月日と時間部分に分けてしまいます。
CREATE EXTERNAL TABLE IF NOT EXISTS log (
 bucket_owner string,
 bucket_name string,
 log_time string,
 log_timezone string,
 remote_ip string,
 requester string,
 request_id string,
 operation string,
 key string,
 request_uri string,
 http_status string,
 error_code string,
 bytes_sent string,
 object_size string,
 total_time string,
 turn_around_time string,
 referer string,
 user_agent string,
 version_id string
) ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
  "input.regex" = "([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([^ \"]*|\"[^\"]*\") ([^ ]*) ([^ ]*) (-|[0-9]*) (-|[0-9]*) (-|[0-9]*) (-|[0-9]*) ([^ ]*) ([^ \"]*|\"[^\"]*\") ([^ ]*)",
      "output.format.string" = "%1$s %2$s %3$s %4$s %5$s %6$s %7$s %8$s %9$s %10$s %11$s %12$s %13$s %14$s %15$s %16$s %17$s %18$s %19$s"
)
LOCATION '${INPUT_BUCKET_LOCATION}';



出力用のテーブルを定義します。LOCATIONは引数から取得します。
パーティションはyyyy,mm,ddという単位で分割します。
CREATE EXTERNAL TABLE IF NOT EXISTS log_archive (
 bucket_owner string, 
 bucket_name string, 
 log_time string,  
 remote_ip string, 
 requester string, 
 request_id string, 
 operation string, 
 key string, 
 request_uri string, 
 http_status string, 
 error_code string, 
 bytes_sent string, 
 object_size string, 
 total_time string, 
 turn_around_time string, 
 referer string, 
 user_agent string, 
 version_id string
 )
 PARTITIONED BY (yyyy string, mm string, dd string)
 ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' STORED AS TEXTFILE LOCATION '${OUTPUT_BUCKET_LOCATION}'; 


集計設定をします。
DynamicPartitionを無効にし、出力をgz圧縮します。
set hive.exec.dynamic.partition= false;
set hive.exec.compress.output = true;


集計します。
対象日のYYYY、MM、DDを引数から受け取ります。
ポイントとなるのは日付の扱いで、日時分割した年月日の方のカラムを0:00:00としてタイムスタンプに変換しJST⇛UTC変換したもので当日分のログとしてフィルタしています。
INSERT INTO TABLE log_archive PARTITION (yyyy='${YYYY}', mm='${MM}', dd='${DD}')
SELECT
 bucket_owner,
 bucket_name,
 concat("[", from_utc_timestamp(from_unixtime(unix_timestamp(concat(log_time, " ", log_timezone), '[dd/MMM/yyyy:HH:mm:ss Z]')), 'JST')," +0900]") as jsttime,
 remote_ip,
 requester,
 request_id,
 operation,
 key,
 request_uri,
 http_status,
 error_code,
 bytes_sent,
 object_size,
 total_time,
 turn_around_time,
 referer,
 user_agent,
 version_id
 FROM
 log
 WHERE
 unix_timestamp(concat(log_time, " ", log_timezone), '[dd/MMM/yyyy:HH:mm:ss Z]') >= unix_timestamp(to_utc_timestamp(concat('${YYYY}-${MM}-${DD}', ' 00:00:00'), 'JST')) 
AND
 unix_timestamp(concat(log_time, " ", log_timezone), '[dd/MMM/yyyy:HH:mm:ss Z]') < unix_timestamp(to_utc_timestamp(concat(date_add('${YYYY}-${MM}-${DD}', 1),' 00:00:00'), 'JST'))
 SORT BY
 jsttime
;



実行


Hiveジョブフローの主な起動設定などは以下の通りです。
Extra ArgsでHiveスクリプトに渡す引数を設定しています。





実行すると以下のように出力されます。




中身を見ると以下のように1日分がまとまっています。
また、日時の部分はJSTに変換して出力されています。

b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:44 +0900],10.115.82.47,arn:aws:iam::821635308497:user/iam-user,52A74398D7C4DBFE,REST.GET.VERSIONING,-,"GET /myfirst-bucket?versioning HTTP/1.1",200,-,162,-,28,-,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:44 +0900],10.115.82.47,arn:aws:iam::821635308497:user/iam-user,7CB888CE4F00C988,REST.GET.BUCKET,-,"GET /myfirst-bucket?prefix=&max-keys=100&marker=&delimiter=/ HTTP/1.1",200,-,1254,-,1044,1043,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:44 +0900],10.15.128.6,arn:aws:iam::821635308497:user/iam-user,473043B904924EB1,REST.GET.LOCATION,-,"GET /myfirst-bucket?location HTTP/1.1",200,-,142,-,29,-,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:44 +0900],10.15.149.57,arn:aws:iam::821635308497:user/iam-user,E1A726EAAA9ED195,REST.GET.LOCATION,-,"GET /myfirst-bucket?location HTTP/1.1",200,-,142,-,47,-,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:49 +0900],10.115.82.47,arn:aws:iam::821635308497:user/iam-user,648FA2CF0669ED48,REST.GET.VERSIONING,-,"GET /myfirst-bucket?versioning HTTP/1.1",200,-,162,-,26,-,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:49 +0900],10.115.82.47,arn:aws:iam::821635308497:user/iam-user,4620CF20BB92DE10,REST.GET.BUCKET,-,"GET /myfirst-bucket?prefix=&max-keys=100&marker=&delimiter=/ HTTP/1.1",200,-,1254,-,601,600,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:51 +0900],10.115.82.47,arn:aws:iam::821635308497:user/iam-user,AD489CC457FB430B,REST.GET.ACL,-,"GET /myfirst-bucket?acl HTTP/1.1",200,-,1317,-,406,-,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:51 +0900],10.115.144.24,arn:aws:iam::821635308497:user/iam-user,700100DED6F992D5,REST.GET.NOTIFICATION,-,"GET /myfirst-bucket?notification HTTP/1.1",200,-,115,-,21,-,"-","S3Console/0.4",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 01:29:51 +0900],10.115.144.24,arn:aws:iam::821635308497:user/iam-user,EEF9831033863E2B,REST.GET.LOGGING_STATUS,-,"GET /myfirst-bucket?logging HTTP/1.1",200,-,932,-,414,-,"-","S3Console/0.4",-

〜〜中略〜〜

+0900],10.89.198.21,b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,139A543A03AB1E01,REST.GET.ACL,-,"GET /?acl=&x-amz-security-token=AQYGQXBwVGtubLMUm1%2BEi%2BJ69YRAEYru2QGPPayzdgIKktgcIhBULb3hBbeFaN3DjPydiCyvDuvQ7g4kucsy0nQsWnAMBR2zfK1R%2B7Y7%2BdVgxVeH2fVJ9FQLFTkg2kkCaVj%2Fj%2FEWi8r%2F1hShQ4prv8OLBfDFETmtdw%3D%3D HTTP/1.1",200,-,1317,-,14,-,"-","aws-sdk-java/unknown-version Linux/2.6.18-194.17.4.el5.acc4xen Java_HotSpot(TM)_Server_VM/20.12-b01",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 17:36:44 +0900],10.89.198.21,b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,5FA69CCF9D1F8ABF,REST.GET.BUCKET,-,"GET /?max-keys=0&x-amz-security-token=AQAGQXBwVGtu2GApXRs%2FDi9mf5qWz55MQRSyS9v%2FnmGvqPo7lsP%2BAkR1V57UhWpJlAn6tYGp2tOL%2BC6Ai1BrRVOWUdUp6JkO%2FMqep7zd4h%2B5xJP8HtcO6pEWEV6t%2BikUh0qYfG8TatCWKvh15j6qu3XExYR5fnveRw%3D%3D HTTP/1.1",200,-,237,-,19,19,"-","aws-sdk-java/unknown-version Linux/2.6.18-194.17.4.el5.acc4xen Java_HotSpot(TM)_Server_VM/20.12-b01",-
b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,myfirst-bucket,[2013-03-20 17:36:44 +0900],10.89.198.21,b00fb3b3fbeb37e2dc44f010cdbeff2c31bc467a2022f00401bf18abbf9e4543,EB88F164094C5ECB,REST.GET.ACL,-,"GET /?acl=&x-amz-security-token=AQEGQXBwVGtuw71vZ%2F%2BhXOKS3VoBVeYAGzc2Csc0R73DDmsCndxf9cMwWqyG9fGRYM%2F%2BJVpuKk4gdS%2Ftr64M7O2VA%2BmzWfPUz8eSwYA3zvEmdBvC8JJrDlvDImPfhqi8mXHF5weAqj2byecuDWLTqJ7AQZBtY3b8dQ%3D%3D HTTP/1.1",200,-,1317,-,26,-,"-","aws-sdk-java/unknown-version Linux/2.6.18-194.17.4.el5.acc4xen Java_HotSpot(TM)_Server_VM/20.12-b01",-


これで日本時間でのアクセス集計ができました。 以上です。

2013年3月23日土曜日

EMRってなんじゃ?(Bootstrap Actionで実行時メモリ使用量を設定する)

EMRを実行するときにメモリが足りなくてエラーになる場合があります。
EMRではクラスタのJavaのメモリ指定ができるのでやってみます。

EMRはジョブフローを起動するときに、Bootstrapアクションを指定出来ます。
AWSコンソールでジョブフローを起動する場合は、ジョブフロー作成のダイアログ中にBootstrapアクションを設定できます。

「BOOTSTRAP ACTIONS」のフェーズで「Configure your Bootstrap Actions」を選択すると以下のような画面が出てきます。



Action Typeにはいくつかあり、監視ツールの「Install Ganglia」や「Memory Intensive Configuration」などのプリセットとカスタムアクションがあります。
「Amazon S3 Location」欄にS3上のアドレスを記載すると、そこに置かれたシェルスクリプトなどがジョブの起動時に実行されるようです。


Memory Intensive Configuration


 ここでは、メモリ集中型設定というのをやってみます。 Action Typeから「Memory Intensive Configuration」という項目のを選びます。 「Amazon S3 Location」欄にS3アドレスが表示されます。



このアドレスに該当するスクリプトを取得すると以下のような内容になっています。
$ curl http://elasticmapreduce.s3.amazonaws.com/bootstrap-actions/configurations/latest/memory-intensive
#!/usr/bin/ruby

## Copyright 2010-2010 Amazon.com, Inc. or its affiliates. All Rights Reserved.
## Licensed under the Apache License, Version 2.0 (the "Liense"). You may not use this file except in compliance with the License.
## A copy of the License is located at http://aws.Amazon/apache2.0/
## or in the "license" file accompanying this file.
## This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
## See the License for the specific language governing permissions and limitations under the License.

require 'json'
require 'hpricot'
require 'tempfile'

CONFIG_HEADER = "<?xml version=\"1.0\"?>\n<?xml-stylesheet type=\"text/xsl\" href=\"configuration.xsl\"?>"

conf_fields = [
  { :field => "mapred.child.java.opts={VAR}",                   :roles => [:slave]          },
  { :field => "mapred.tasktracker.map.tasks.maximum={VAR}",     :roles => [:master, :slave] },
  { :field => "mapred.tasktracker.reduce.tasks.maximum={VAR}",  :roles => [:master, :slave] }
]

configs = {
  "m1.small"    => ["-Xmx512m",  "2",  "1"],
  "m1.large"    => ["-Xmx1024m", "3",  "1"],
  "m1.xlarge"   => ["-Xmx1024m", "8",  "3"],
  "c1.medium"   => ["-Xmx512m",  "2",  "1"],
  "c1.xlarge"   => ["-Xmx512m",  "7",  "2"],
  "m2.xlarge"   => ["-Xmx3072m", "3",  "1"],
  "m2.2xlarge"  => ["-Xmx4096m", "6",  "2"],
  "m2.4xlarge"  => ["-Xmx4096m", "14", "4"]
}

heap_fields = [
  { :field => "HADOOP_NAMENODE_HEAPSIZE={VAR}",    :roles => [:master, :slave] },
  { :field => "HADOOP_JOBTRACKER_HEAPSIZE={VAR}",  :roles => [:master, :slave] },
  { :field => "HADOOP_TASKTRACKER_HEAPSIZE={VAR}", :roles => [:master, :slave] },
  { :field => "HADOOP_DATANODE_HEAPSIZE={VAR}",    :roles => [:master, :slave] }
]

heaps = {
  "m1.small"    => ["512",  "512",  "256",  "128"],
  "m1.large"    => ["1024", "3072", "512",  "512"],
  "m1.xlarge"   => ["3072", "9216", "512",  "512"],
  "c1.medium"   => ["512",  "768",  "256",  "128"],
  "c1.xlarge"   => ["1024", "2048", "512",  "512"],
  "m2.xlarge"   => ["2048", "4096", "512",  "512"],
  "m2.2xlarge"  => ["2048", "8192", "1024", "1024"],
  "m2.4xlarge"  => ["8192", "8192", "1024", "1024"]
}

def parse_config_file(config_file_path)
  ret = []
  if File.exist?(config_file_path) then
    doc = open(config_file_path) { |f| Hpricot(f) }
    (doc/"configuration"/"property").each do |property|
      val = {:name => (property/"name").inner_html, :value => (property/"value").inner_html }
      if (property/"final").inner_html != "" then
        val[:final] =  (property/"final").inner_html
      end
      ret << val
    end
  else
    puts "#{config_file_path} does not exist, assuming empty configuration"
  end
  return ret
end

def dump_config_file(file_name, config)
  open(file_name, 'w') do |f|
    f.puts CONFIG_HEADER
    f.puts '<configuration>'
    for entry in config
      f.print "  <property><name>#{entry[:name]}</name><value>#{entry[:value]}</value>"
      if entry[:final] then
        f.print "<final>#{entry[:final]}</final>"
      end
      f.puts '</property>'
    end
    f.puts '</configuration>'
  end
end

def merge_config(default, overwrite)
  for entry in overwrite
    cells = default.select { |x| x[:name] == entry[:name]}
    if cells.size == 0 then
      puts "'#{entry[:name]}': default does not have key, appending value '#{entry[:value]}'"
      default << entry
    elsif cells.size == 1 then
      puts "'#{entry[:name]}': new value '#{entry[:value]}' overwriting '#{cells[0][:value]}'"
      cells[0].replace(entry)
    else
      raise "'#{entry[:name]}': default has #{cells.size} keys"
    end
  end
end

def do_overwrites(conf_list, heap_list)
  file = "/home/hadoop/conf/mapred-site.xml"
  default = parse_config_file(file)
  for arg in conf_list
    puts "Processing default file #{file} with overwrite #{arg}"
    key   = arg.split('=', 2)[0]
    value = arg.split('=', 2)[1]
    overwrite = [{:name => key, :value => value }]
    merge_config(default,overwrite)
  end
  dump_config_file(file + ".new", default)
  if File.exist?(file) then
    File.rename(file, file + ".old")
  end
  File.rename(file + ".new", file)
  puts "Saved #{file} with overwrites. Original saved to #{file}.old"
  file = "/home/hadoop/conf/hadoop-user-env.sh"
  if File.exist?(file) then
    File.delete(file)
  end
  open(file, 'w') do |f|
    f.puts "#!/bin/bash"
    for arg in heap_list
      f.puts arg
    end
  end
end

class JsonInfoFile

  INFO_DIR = "/mnt/var/lib/info/"

  def initialize(file_type)
    @json = JSON.parse(File.read(File.join(INFO_DIR, file_type + ".json")))
  end

  def [](json_path)
    json = @json
    begin
      path = json_path.split('.')
      visited = []
      for item in path
        if !json.kind_of? Hash then
          raise "#{visited.join('.')} not of type object, got '#{json.inspect}' from #{@json.inspect}"
        end
        visited << item
        json = json[item]
      end
      if json == nil then
        raise "#{visited.join('.')} does not exist"
      end
      return json
    rescue
      puts "Unable to process path '#{json_path}', #{$!}"
      exit -1
    end
  end
end

def warn(msg)
  STDERR.puts "#{Time.now.utc} WARN " + msg
end

def substitute_in(row, fields, instance_role)
  if row.size != fields.size then
    raise RuntimeError, "Incompatible row and field list row=#{row}, fields=#{fields}"
  end
  result = []
  for index in 0 ... row.size do
    if fields[index][:roles].include?(instance_role) then
      result << fields[index][:field].sub('{VAR}', row[index])
    end
  end
  return result
end

HPC_INSTANCE_TYPES = [ "cc1.4xlarge", "cg1.4xlarge" ]

jobflow_info = JsonInfoFile.new("job-flow")
instance_info = JsonInfoFile.new("instance")
if instance_info['isMaster'].to_s == 'true'  then
  instance_type = jobflow_info["masterInstanceType"]
  instance_role = :master
else
  instance_group_id = instance_info['instanceGroupId']
  instance_groups = jobflow_info['instanceGroups']
  index = instance_groups.index { |g| g['instanceGroupId'] == instance_group_id }
  instance_group = instance_groups[index]
  instance_type = instance_group['instanceType']
  instance_role = :slave
end

if HPC_INSTANCE_TYPES.include?(instance_type) then
  warn "This bootstrap action is not supported for the HPC instances (cc1.4xlarge and cg1.4xlarge)"
else
  conf_list = substitute_in(configs[instance_type], conf_fields, instance_role)
  heap_list = substitute_in(heaps[instance_type], heap_fields, instance_role)
  do_overwrites(conf_list, heap_list)
end

RUBYスクリプトのようです。
このスクリプトではインスタンスサイズに応じてネームノード、JobTracker、TaskTracker、データノードに対して、メモリ集中型のヒープサイズをそれぞれの環境変数に設定するように
/home/hadoop/conf/hadoop-user-env.sh
を上書きしています。

そして、hadoop-user-env.shは
/home/hadoop/conf/hadoop-env.sh
から以下のように読み込まれ、ジョブ実行時に実行される仕組みのようです。

#!/bin/bash

export HADOOP_DATANODE_HEAPSIZE="96"
export HADOOP_NAMENODE_HEAPSIZE="192"
export HADOOP_JOBTRACKER_HEAPSIZE="576"
export HADOOP_TASKTRACKER_HEAPSIZE="192"
export HADOOP_OPTS="$HADOOP_OPTS -server"
if [ -e /home/hadoop/conf/hadoop-user-env.sh ] ; then
  . /home/hadoop/conf/hadoop-user-env.sh
fi


これがメモリ集中型の場合
  • HADOOP_JOBTRACKER_HEAPSIZE:512
  • HADOOP_NAMENODE_HEAPSIZE:512
  • HADOOP_TASKTRACKER_HEAPSIZE:256
  • HADOOP_DATANODE_HEAPSIZE:128

のようになるようです。
ネームノードにより多くのメモリが割り当てられるようです。
これで実行すると、新しいメモリ割り当ての設定でジョブが実行されます。



Custom Action



プリセット以外にも、「Custom Action」という自由にスクリプトなどをアップできるオプションもあります。
「Amazon S3 Location」欄に自分でアップしたスクリプトファイルのS3アドレスを入力します。
ここでは、例として
s3://memorycraft-emr/script/heap.sh として以下のファイルアップしておきます。
前述のMemory Intensive Configurationではインスタンスごとに決め打ちのヒープサイズが自動で適用されましたが、このように自分で各ヒープサイズを指定するように hadoop-user-env.shを作るようなスクリプトを書くこともできます。



そして、以下のようにS3アドレスを入力します。


あとは通常どおり実行します。



インスタンスサイズは上げたくないけどOutOfMemoryを回避したいという場合には、まずこのようにBootstrapActionで調整してみるといいと思います。

また、Bootstrap設定はSDKなどのAPI経由のジョブフロー起動時にも行うことが可能です。
  .... 略 .....
   'AmiVersion'       => 'latest',
   'LogUri'           => "$LOG_BUCKET_LOCATION",
   'BootstrapActions' => array(
    array(
     'Name'                  => 'Custom Action',
     'ScriptBootstrapAction' => array(
      'Path' => 's3://memorycraft-emr/script/heap.sh',
     ),
    ),
   ),
   'Steps' => array(
    new \CFStepConfig(array(
  .... 略 .....


メモリ設定以外にも処理前に自由に処理を挿入できるので、カスタム処理を行ったり、実行環境を最適化できるので、いろいろ便利そうです。


以上です。


2013年3月11日月曜日

EMRってなんじゃ?(EMRの詳細なログをみる)

EMRをつかっていて、ステップの途中でエラーが出たときに、デバッグ用のログ出力先を設定すると実行ログがAWSコンソール上で見ることができます。

例として、S3のログバケットをまとめるHiveScriptを動かした時のエラーを見てみます。


エラーが発生しています。
LogURIを指定していると、このようにコンソールからLogFileとして確認することができます。
エラーが起こった3番目のステップのそれぞれのリンクをみてみると以下のようになっています。


controller
2013-03-11T11:31:41.005Z INFO Fetching jar file.
2013-03-11T11:31:48.146Z INFO Working dir /mnt/var/lib/hadoop/steps/3
2013-03-11T11:31:48.146Z INFO Executing /usr/lib/jvm/java-6-sun/bin/java -cp /home/hadoop/conf:/usr/lib/jvm/java-6-sun/lib/tools.jar:/home/hadoop:/home/hadoop/hadoop-core.jar:/home/hadoop/hadoop-tools.jar:/home/hadoop/hadoop-core-0.20.205.jar:/home/hadoop/hadoop-tools-0.20.205.jar:/home/hadoop/lib/*:/home/hadoop/lib/jetty-ext/* -Xmx1000m -Dhadoop.log.dir=/mnt/var/log/hadoop/steps/3 -Dhadoop.log.file=syslog -Dhadoop.home.dir=/home/hadoop -Dhadoop.id.str=hadoop -Dhadoop.root.logger=INFO,DRFA -Djava.io.tmpdir=/mnt/var/lib/hadoop/steps/3/tmp -Djava.library.path=/home/hadoop/native/Linux-i386-32 org.apache.hadoop.util.RunJar /mnt/var/lib/hadoop/steps/3/script-runner.jar s3://ap-northeast-1.elasticmapreduce/libs/hive/hive-script --run-hive-script --base-path s3://ap-northeast-1.elasticmapreduce/libs/hive/ --hive-versions latest --args -f s3n://memorycraft-archive/script/s3.hql -d INPUT_BUCKET_LOCATION=s3://memorycraft-log/logs/ -d OUTPUT_BUCKET_LOCATION=s3://memorycraft-archive/rslt/ -d YYYY=2013 -d MM=03 -d DD=04
2013-03-11T11:34:20.383Z INFO Execution ended with ret val 255
2013-03-11T11:34:20.384Z WARN Step failed with bad retval
2013-03-11T11:34:26.454Z INFO Step created jobs: job_201303111128_0001


stderr
Logging initialized using configuration in file:/home/hadoop/.versions/hive-0.8.1/conf/hive-log4j.properties
Hive history file=/mnt/var/lib/hive_081/tmp/history/hive_job_log_hadoop_201303111132_1431275973.txt
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/home/hadoop/lib/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/home/hadoop/.versions/hive-0.8.1/lib/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
OK
Time taken: 13.475 seconds
OK
Time taken: 0.765 seconds
Total MapReduce jobs = 1
Launching Job 1 out of 1
Number of reduce tasks determined at compile time: 1
In order to change the average load for a reducer (in bytes):
  set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
  set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
  set mapred.reduce.tasks=<number>
Starting Job = job_201303111128_0001, Tracking URL = http://ip-10-121-17-157.ap-northeast-1.compute.internal:9100/jobdetails.jsp?jobid=job_201303111128_0001
Kill Command = /home/hadoop/bin/hadoop job  -Dmapred.job.tracker=10.121.17.157:9001 -kill job_201303111128_0001
Hadoop job information for Stage-1: number of mappers: 1; number of reducers: 1
2013-03-11 11:33:11,849 Stage-1 map = 0%,  reduce = 0%
2013-03-11 11:34:12,380 Stage-1 map = 0%,  reduce = 0%
2013-03-11 11:34:19,431 Stage-1 map = 100%,  reduce = 100%
Ended Job = job_201303111128_0001 with errors
Error during job, obtaining debugging information...
Examining task ID: task_201303111128_0001_m_000002 (and more) from job job_201303111128_0001
Exception in thread "Thread-47" java.lang.RuntimeException: Error while reading from task log url
 at org.apache.hadoop.hive.ql.exec.errors.TaskLogProcessor.getErrors(TaskLogProcessor.java:130)
 at org.apache.hadoop.hive.ql.exec.JobDebugger.showJobFailDebugInfo(JobDebugger.java:211)
 at org.apache.hadoop.hive.ql.exec.JobDebugger.run(JobDebugger.java:81)
 at java.lang.Thread.run(Thread.java:662)
Caused by: java.io.IOException: Server returned HTTP response code: 400 for URL: http://10.122.24.232:9103/tasklog?taskid=attempt_201303111128_0001_m_000000_1&start=-8193
 at sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1436)
 at java.net.URL.openStream(URL.java:1010)
 at org.apache.hadoop.hive.ql.exec.errors.TaskLogProcessor.getErrors(TaskLogProcessor.java:120)
 ... 3 more
Counters:
FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.MapRedTask
MapReduce Jobs Launched: 
Job 0: Map: 1  Reduce: 1   HDFS Read: 0 HDFS Write: 0 FAIL
Total MapReduce CPU Time Spent: 0 msec
Command exiting with ret '255'


stdout
Downloading 's3://ap-northeast-1.elasticmapreduce/libs/hive/hive-script' to '/mnt/var/lib/hadoop/steps/3/.'
2013-03-11 11:31:52 GMT - INFO Running: /home/hadoop/.versions/hive-0.8.1/bin/hive '-f' 's3n://memorycraft-archive/script/s3.hql' '-d' 'INPUT_BUCKET_LOCATION=s3://memorycraft-log/logs/' '-d' 'OUTPUT_BUCKET_LOCATION=s3://memorycraft-archive/rslt/' '-d' 'YYYY=2013' '-d' 'MM=03' '-d' 'DD=04'
2013-03-11 11:34:20 GMT - ERROR Error executing cmd: /home/hadoop/.versions/hive-0.8.1/bin/hive '-f' 's3n://memorycraft-archive/script/s3.hql' '-d' 'INPUT_BUCKET_LOCATION=s3://memorycraft-log/logs/' '-d' 'OUTPUT_BUCKET_LOCATION=s3://memorycraft-archive/rslt/' '-d' 'YYYY=2013' '-d' 'MM=03' '-d' 'DD=04'


syslog
2013-03-11 11:31:52,225 INFO org.apache.hadoop.fs.s3native.NativeS3FileSystem (main): Opening '/libs/hive/hive-script' for reading


一番詳しく出ているのが、stderrのようですが、これだけではよくわかりません。
実際は、AWSコンソール上で確認できるEMRログはこれで全てではなく、詳細なタスクのログがEMRログバケットに出力されています。


追記:
TaskとTask Attemptsに関しては、右側のView Jobsから掘っていくことで、EMRのDebugダイアログでも確認でしましたが、それ以外のログはやはりS3バケットでしかみれなそうです。


EMRログバケットを見てみます。


ジョブフローIDごとにフォルダがあり、その中のstepsというフォルダが、AWSコンソールで表示されるログが入っています。


steps/3 以下に先ほどのログファイルが存在しています。
これ以上くわしい実行タスクのログを見るには、task-attemptsというフォルダに詳細なログが出力されています。



このtask-attempts以下の各サブフォルダにあるsyslogというファイルに詳細なログが出力されています。


task-attempts/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/syslog
2013-03-11 11:33:13,112 INFO org.apache.hadoop.util.NativeCodeLoader (main): Loaded the native-hadoop library
2013-03-11 11:33:13,268 INFO org.apache.hadoop.mapred.TaskRunner (main): Creating symlink: /mnt/var/lib/hadoop/mapred/taskTracker/distcache/-5343292120509370400_-2048022790_1496912286/10.121.17.157/mnt/var/lib/hive_081/tmp/scratch/hive_2013-03-11_11-32-20_837_75945303502865249/-mr-10001/3d096d97-37e9-47c7-a184-8152010d592a <- /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/work/HIVE_PLAN3d096d97-37e9-47c7-a184-8152010d592a
2013-03-11 11:33:13,354 INFO org.apache.hadoop.filecache.TrackerDistributedCacheManager (main): Creating symlink: /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/jars/job.jar <- /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/work/job.jar
2013-03-11 11:33:13,362 INFO org.apache.hadoop.filecache.TrackerDistributedCacheManager (main): Creating symlink: /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/jars/.job.jar.crc <- /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/work/.job.jar.crc
2013-03-11 11:33:13,370 INFO org.apache.hadoop.filecache.TrackerDistributedCacheManager (main): Creating symlink: /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/jars/META-INF <- /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/work/META-INF
2013-03-11 11:33:13,378 INFO org.apache.hadoop.filecache.TrackerDistributedCacheManager (main): Creating symlink: /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/jars/org <- /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/work/org
2013-03-11 11:33:13,386 INFO org.apache.hadoop.filecache.TrackerDistributedCacheManager (main): Creating symlink: /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/jars/hive-exec-log4j.properties <- /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/work/hive-exec-log4j.properties
2013-03-11 11:33:13,430 INFO org.apache.hadoop.filecache.TrackerDistributedCacheManager (main): Creating symlink: /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/jars/javaewah <- /mnt/var/lib/hadoop/mapred/taskTracker/hadoop/jobcache/job_201303111128_0001/attempt_201303111128_0001_m_000000_0/work/javaewah
2013-03-11 11:33:13,838 WARN org.apache.hadoop.metrics2.impl.MetricsSystemImpl (main): Source name ugi already exists!
2013-03-11 11:33:14,037 INFO org.apache.hadoop.mapred.MapTask (main): Host name: ip-10-122-12-88.ap-northeast-1.compute.internal
2013-03-11 11:33:14,088 INFO org.apache.hadoop.util.ProcessTree (main): setsid exited with exit code 0
2013-03-11 11:33:14,106 INFO org.apache.hadoop.mapred.Task (main):  Using ResourceCalculatorPlugin : org.apache.hadoop.util.LinuxResourceCalculatorPlugin@1dfd868
2013-03-11 11:33:14,342 INFO com.hadoop.compression.lzo.GPLNativeCodeLoader (main): Loaded native gpl library
2013-03-11 11:33:14,358 WARN com.hadoop.compression.lzo.LzoCodec (main): Could not find build properties file with revision hash
2013-03-11 11:33:14,358 INFO com.hadoop.compression.lzo.LzoCodec (main): Successfully loaded & initialized native-lzo library [hadoop-lzo rev UNKNOWN]
2013-03-11 11:33:14,367 WARN org.apache.hadoop.io.compress.snappy.LoadSnappy (main): Snappy native library is available
2013-03-11 11:33:14,367 INFO org.apache.hadoop.io.compress.snappy.LoadSnappy (main): Snappy native library loaded
2013-03-11 11:33:14,452 WARN org.apache.hadoop.hive.conf.HiveConf (main): hive-site.xml not found on CLASSPATH
2013-03-11 11:33:16,876 INFO org.apache.hadoop.fs.s3native.NativeS3FileSystem (main): Opening 's3://memorycraft-log/logs/2013-03-04-09-16-51-45565F6169B3E2F2' for reading
2013-03-11 11:33:17,238 INFO org.apache.hadoop.io.retry.RetryInvocationHandler (main): Exception while invoking retrievePair of class org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore. Not retrying.Status Code: 403, AWS Service: Amazon S3, AWS Request ID: 2981A88871482E0B, AWS Error Code: InvalidObjectState, AWS Error Message: The operation is not valid for the object's storage class, S3 Extended Request ID: 7yHCAsfLHZKbf+5ojWMS9q2B41bBwizoqW4zldSdfEpLoLrKvI9X1WI0brHGsC49
 at com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:548)
 at com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:288)
 at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:170)
 at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:2619)
 at com.amazonaws.services.s3.AmazonS3Client.getObject(AmazonS3Client.java:809)
 at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrievePair(Jets3tNativeFileSystemStore.java:266)
 at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrievePair(Jets3tNativeFileSystemStore.java:252)
 at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
 at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
 at java.lang.reflect.Method.invoke(Method.java:597)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:82)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:59)
 at org.apache.hadoop.fs.s3native.$Proxy7.retrievePair(Unknown Source)
 at org.apache.hadoop.fs.s3native.NativeS3FileSystem.open(NativeS3FileSystem.java:1002)
 at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:420)
 at org.apache.hadoop.mapred.LineRecordReader.<init>(LineRecordReader.java:78)
 at org.apache.hadoop.mapred.TextInputFormat.getRecordReader(TextInputFormat.java:51)
 at org.apache.hadoop.hive.ql.io.CombineHiveRecordReader.<init>(CombineHiveRecordReader.java:65)
 at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
 at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:39)
 at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:27)
 at java.lang.reflect.Constructor.newInstance(Constructor.java:513)
 at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileRecordReader.initNextRecordReader(HadoopShimsSecure.java:331)
 at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileRecordReader.<init>(HadoopShimsSecure.java:292)
 at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileInputFormatShim.getRecordReader(HadoopShimsSecure.java:406)
 at org.apache.hadoop.hive.ql.io.CombineHiveInputFormat.getRecordReader(CombineHiveInputFormat.java:549)
 at org.apache.hadoop.mapred.MapTask$TrackedRecordReader.<init>(MapTask.java:199)
 at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:423)
 at org.apache.hadoop.mapred.MapTask.run(MapTask.java:377)
 at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
 at java.security.AccessController.doPrivileged(Native Method)
 at javax.security.auth.Subject.doAs(Subject.java:396)
 at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1059)
 at org.apache.hadoop.mapred.Child.main(Child.java:249)

2013-03-11 11:33:17,315 INFO org.apache.hadoop.mapred.TaskLogsTruncater (main): Initializing logs' truncater with mapRetainSize=-1 and reduceRetainSize=-1
2013-03-11 11:33:17,452 INFO org.apache.hadoop.io.nativeio.NativeIO (main): Initialized cache for UID to User mapping with a cache timeout of 14400 seconds.
2013-03-11 11:33:17,453 INFO org.apache.hadoop.io.nativeio.NativeIO (main): Got UserName hadoop for UID 106 from the native implementation
2013-03-11 11:33:17,456 WARN org.apache.hadoop.mapred.Child (main): Error running child
java.io.IOException: java.lang.reflect.InvocationTargetException
 at org.apache.hadoop.hive.io.HiveIOExceptionHandlerChain.handleRecordReaderCreationException(HiveIOExceptionHandlerChain.java:97)
 at org.apache.hadoop.hive.io.HiveIOExceptionHandlerUtil.handleRecordReaderCreationException(HiveIOExceptionHandlerUtil.java:57)
 at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileRecordReader.initNextRecordReader(HadoopShimsSecure.java:345)
 at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileRecordReader.<init>(HadoopShimsSecure.java:292)
 at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileInputFormatShim.getRecordReader(HadoopShimsSecure.java:406)
 at org.apache.hadoop.hive.ql.io.CombineHiveInputFormat.getRecordReader(CombineHiveInputFormat.java:549)
 at org.apache.hadoop.mapred.MapTask$TrackedRecordReader.<init>(MapTask.java:199)
 at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:423)
 at org.apache.hadoop.mapred.MapTask.run(MapTask.java:377)
 at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
 at java.security.AccessController.doPrivileged(Native Method)
 at javax.security.auth.Subject.doAs(Subject.java:396)
 at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1059)
 at org.apache.hadoop.mapred.Child.main(Child.java:249)
Caused by: java.lang.reflect.InvocationTargetException
 at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
 at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:39)
 at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:27)
 at java.lang.reflect.Constructor.newInstance(Constructor.java:513)
 at org.apache.hadoop.hive.shims.HadoopShimsSecure$CombineFileRecordReader.initNextRecordReader(HadoopShimsSecure.java:331)
 ... 11 more
Caused by: Status Code: 403, AWS Service: Amazon S3, AWS Request ID: 2981A88871482E0B, AWS Error Code: InvalidObjectState, AWS Error Message: The operation is not valid for the object's storage class, S3 Extended Request ID: 7yHCAsfLHZKbf+5ojWMS9q2B41bBwizoqW4zldSdfEpLoLrKvI9X1WI0brHGsC49
 at com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:548)
 at com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:288)
 at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:170)
 at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:2619)
 at com.amazonaws.services.s3.AmazonS3Client.getObject(AmazonS3Client.java:809)
 at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrievePair(Jets3tNativeFileSystemStore.java:266)
 at org.apache.hadoop.fs.s3native.Jets3tNativeFileSystemStore.retrievePair(Jets3tNativeFileSystemStore.java:252)
 at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
 at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
 at java.lang.reflect.Method.invoke(Method.java:597)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:82)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:59)
 at org.apache.hadoop.fs.s3native.$Proxy7.retrievePair(Unknown Source)
 at org.apache.hadoop.fs.s3native.NativeS3FileSystem.open(NativeS3FileSystem.java:1002)
 at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:420)
 at org.apache.hadoop.mapred.LineRecordReader.<init>(LineRecordReader.java:78)
 at org.apache.hadoop.mapred.TextInputFormat.getRecordReader(TextInputFormat.java:51)
 at org.apache.hadoop.hive.ql.io.CombineHiveRecordReader.<init>(CombineHiveRecordReader.java:65)
 ... 16 more
2013-03-11 11:33:17,500 INFO org.apache.hadoop.mapred.Task (main): Runnning cleanup for the task


内容をみると、
Error Message: The operation is not valid for the object's storage class, S3 Extended Request ID
となっており、ストレージクラスが良くないと言われています。

EMRの入力ログバケットを見ると、Glaceirになっているクラスがありました。
EMRはGlaceirファイルは処理できないので、エラーになっていたようです。

EMRでワケの分からないエラーが出たらEMRログのS3バケットを見るようにしよう。
以上です。



2013年2月22日金曜日

EMRってなんじゃ?(HiveScriptの場合)

前回の記事ではSSHでEMRの中でhiveコンソールをつかって手動でクエリを打ちましたが、通常のプログラムなどからではHiveScriptで起動と同時に実行することが多いと思うので、細かいことですが前回との違いを紹介します。

HiveScriptの場合はスクリプトファイルの場所と、オプションで入力ファイルのS3ロケーションと出力ファイルのS3ロケーションなどを設定します。
前回やったとおり、入出力のロケーションについてはHiveScript中にもEXTERNAL TABLEとして記載できるのでここではHiveScriptのみの設定でOKです。

ただしHiveScriptの場合、スクリプトのS3ロケーションの部分を前回と内容を少しかえる必要があります。
前回は入力のバケットファイルをファイルパスまで指定しましたが、今回はバケットフォルダで指定し、その下の全てのファイルが入力となります。



hive.q
CREATE EXTERNAL TABLE IF NOT EXISTS csv_origin (yyyy string, mm string, dd string, title string, body string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' LOCATION 's3://csv-origin/input/';

CREATE EXTERNAL TABLE IF NOT EXISTS csv_archive (dd string, title string, body string) PARTITIONED BY (yyyy string, mm string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' STORED AS TEXTFILE LOCATION 's3://csv-archive/rslt/';                                                              

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
FROM csv_origin co INSERT OVERWRITE TABLE csv_archive PARTITION (yyyy, mm)
SELECT
 co.dd,
 co.title,
 co.body,
 co.yyyy,
 co.mm
DISTRIBUTE BY co.yyyy, co.mm;



スクリプト用のバケットにスクリプトファイルを配置します。



EMRでJob Flowを作成し、Hive Programを選択します。
いまアップしたスクリプトファイルの場所を設定します。

Script Location: s3://hivescript-test/hive.q



あとはそのまま実行します。
しばらくするとジョブが終了し、出力を確認できます。




03,title32,body32
16,title33,body33



以上です。

2013年1月28日月曜日

EMRってなんじゃ?(HiveのDynamicPartitionでファイルを分割してS3に出力)

S3にある数年分のデータが1つのファイルに含まれていて、EMRで年や月単位で分割したいという要望が一部であったので、やってみました。

例として、以下のようなCSVがあるとします。

test.csv
2012,01,12,title1,body1
2012,01,22,title2,body2
2012,02,02,title3,body3
2012,03,01,title4,body4
2012,03,11,title5,body5
2012,03,18,title6,body6
2012,04,04,title7,body7
2012,05,02,title8,body8
2012,05,05,title9,body9
2012,06,13,title10,body10
2012,06,14,title11,body11
2012,06,29,title12,body12
2012,07,03,title13,body13
2012,07,05,title14,body14
2012,07,14,title15,body15
2012,07,30,title16,body16
2012,08,12,title17,body17
2012,09,23,title18,body18
2012,10,10,title19,body19
2012,11,23,title20,body20
2012,12,22,title21,body21
2013,01,04,title22,body22
2013,01,05,title23,body23
2013,01,06,title24,body24
2013,01,17,title25,body25
2013,01,18,title26,body26
2013,01,23,title27,body27
2013,02,03,title28,body28
2013,02,04,title29,body29
2013,02,05,title30,body30
2013,02,06,title31,body31
2013,03,03,title32,body32
2013,03,16,title33,body33
2013,04,03,title34,body34
2013,04,04,title35,body35
2013,04,09,title36,body36
2013,04,12,title37,body37
2013,04,14,title38,body38
2013,04,17,title39,body39


まず、入力用のバケットと出力用のバケットを用意します。

入力
 csv-origin(上記のtest.csvをアップしておきます。)
出力
 csv-archive (rsltというフォルダを作成しておきます。)





EMRでジョブフローを作成します。今回はインタラクティブジョブフローを使います。
これはEMRクラスタにSSHで入って、hiveコンソールで直接クエリを実行する方式です。

まず、EMRで「Create New Job Flow」をクリックします。
ダイアログが立ち上がるのでJob FlowにHive Programを選択します。




今回HiveScriptは使用せずインタラクティブなので、「Start as Interactive Hive Session」を選択肢ます。




インスタンス数やサイズなどはとりあえずそのままでOKです。




SSHでログインするときのキーを選択して




以降はそのまま「Continue」を続けます。




すると、以下のようにジョブフローが起ち上がり、マスターノードのエンドポイントが記されます。




このエンドポイントに対して、指定したキーを使ってhadoopユーザーでSSHログインします。
ssh -i ./memorycraft.pem hadoop@ec2-175-41-212-129.ap-northeast-1.compute.amazonaws.com

hiveコマンドを打つとhiveコンソールに接続されます。
$ hive
Logging initialized using configuration in file:/home/hadoop/.versions/hive-0.8.1/conf/hive-log4j.properties
Hive history file=/mnt/var/lib/hive_081/tmp/history/hive_job_log_hadoop_201301280945_987980527.txt
hive>


ここでまず入力用バケットを表すテーブルを作成します。
  • LOCATION句で入力用S3バケットのtest.csvのパスを指定します。
hive> CREATE EXTERNAL TABLE IF NOT EXISTS csv_origin (yyyy string, mm string, dd string, title string, body string)
      > ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n'
      > LOCATION 's3://csv-origin/test.csv';



次に、出力用のバケットを表すテーブルを作成します。
  • 年ごと月ごとにフォルダ分けされるようにyyyyとmmでPARTITIONED BYを指定します。
  • テーブルの項目にはyyyy, mm以外の項目を指定します。
  • 出力ファイルがテキストファイルになるようSTORED AS TEXTFILEを指定します。
  • LOCATION句で出力用S3バケットのパスを指定します。
hive> CREATE EXTERNAL TABLE IF NOT EXISTS csv_archive (dd string, title string, body string) PARTITIONED BY (yyyy string, mm string)
      > ROW FORMAT DELIMITED
      > FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n'
      > STORED AS TEXTFILE
      > LOCATION 's3://csv-archive/rslt/';



あとはINSERT ~ SELECTなのですが、普通にやると、PARTITION(yyyy='2012', mm='02')のようにパーティションを決め打ちで指定しないといけなくなるので、DynamicPartitionというものを使います。
(通常のパーティションは別の記事で改めて書いてみたいと思います。)

DynamicPartitionを有効にするには、hive.exec.dynamic.partitionとhive.exec.dynamic.partition.modeというパラメータをセットしつつ、以下の様な構文を使用します。
hive> set hive.exec.dynamic.partition=true;
hive> set hive.exec.dynamic.partition.mode=nonstrict;
hive> FROM csv_origin co
      > INSERT OVERWRITE TABLE csv_archive PARTITION (yyyy, mm)
      > SELECT
      >  co.dd,
      >  co.title,
      >  co.body,
      >  co.yyyy,
      >  co.mm
      > DISTRIBUTE BY co.yyyy, co.mm;


このクエリにより、Map/Reduceが行われ、完了すると出力用のバケットに以下のようにフォルダが作成され、それぞれのフォルダに該当データが集約されたファイルが出力されます。



たとえば、yyyy=2013/mm=03というフォルダに出力されたファイルには以下の様に2013/03のデータだけが記述されています。

03,titile32,body32
16,titile33,body33

無事に年月単位でデータが分割されるようになりました。
今回のサンプルは数がすくないですが、大量にバラバラなデータが入ったファイルを規則的に分割する場合はこういった方法が有効かもしれません。

以上です。


2012年12月26日水曜日

DataPipelineってなんじゃ?


Data PipelineがUSリージョンで使えるようになっていました。
Data Pipelineはデータの移動と加工を定期的なスケジュールで処理できるためのワークフローを定義できるサービスだそうです。

前の記事では、ApacheのログをS3に流してEMRでマージと集計を行いました。
そのときはEMRのコマンドラインで対象日時をコマンド引数に渡す形にするところまでで終わりました。
これを日次で行うとすると、そのコマンド呼出をどこかの端末のcronに設定するような形になると思います。

Data Pipelineを使うと、そういったcronなどの実行コントロールをEC2やローカル端末などのOS内部ではなく、AWSのサービス上に持つことができます。





それでは前回行ったことをData Pipelineで同じようにやってみます。


設定


AWSコンソールでリージョンを「N.Virginia」にします。
ServicesのDeployment&ManagementにData Pipelineがあるので、クリックします。




Data Pipelineの画面が開き、Create Pipelineというボタンがあるのでクリックします。
するとData Pipelineの初期入力画面が表示されます。
ここではPipelineの名前と、スケジューリングのタイプを選択します。




Select Schedule Typeでは、Time SeriesととCronという2つのスタイルが選べます。
TimeSeriesは設定したスケジュールの終わりに処理が実行され、cronは設定したスケジュールの最初に実行されるようです。

ここではcronスタイルを選択します。
また、DataPipeline自体や、DataPipelineの処理をうけもつEC2などにIAMを紐付けたい場合はRoleで設定します。

「Create New Pipeline」をクリックすると、新しいPipelineが作成され、Pipelineコンソールが開きます。




Pipelineの一覧画面が表示されいて、いま作ったPipelineが一つ登録されています。
細かい項目を設定するため、「View Pipeline」をクリックします。

するとPipelineの設定画面が表示されます。
基本的には左ペインで要素を追加して、右ペインで詳細を設定していき、処理をつなげていく流れです。
基本的には、ヘッダの「Templates」ボタンを押すと、いくつかの典型的なプリセットが用意されているので、それを元に設定していけばいいのですが、今回は1からセットしてみます。




要素には主に以下のような種類があります。

Activity:処理を表す要素です。データコピー、EMR、HIVE、シェルコマンドが選べます。
DataNodes:処理の入力と出力をあらわすデータの要素です。DynamoDB、S3、MySQLが選べます。
Schedule:実行タイミングの要素です。実行間隔と処理期間を設定します。
Resources:処理を実行する環境要素です。EMRとEC2が選べます。
Precondition:処理のを行うかどうかの前提条件の要素です。DynamoDBやS3の存在チェックやシェル、SQLなどで自前のチェックなどが選べます。
Others:上記以外の設定項目を適宜設定できます。上記の要素の設定項目が依存している設定が半自動的に追加されます。


それではまず最初の要素を追加していきます。


DataNodes


ヘッダからactivityかdata nodeを追加できるようになっています。
ここでは、ログの入力元と出力先のdata nodeを追加するので、「Add data node」を3回クリックします。
すると3つのdata nodeが配置され、右側のペインにそれぞれの項目が現れます。




このうちの最初の1つは前回fluentdがログを流し込む入力元のバケットになります。

Nameにlogと入力し、TypeでS3DataNodeを選択します。
Scheduleに「Create new: Schedule」があるので、それを選択します。
するとSchedulesに新規にSchedule要素が追加されます。
また、Directory Pathにfluentdが流しこむ先のS3フォルダパスを記載します。
パスはYYYY-MM-ddで保存されるので、動的になりますが、Data Pipelineでは#{expressions}の形式で変数を割り当てられるようにExpressionという仕組みがあるので、それを利用して、
s3://memorycraft-us-log/logs/#{@scheduledStartTime.format('YYYY-MM-dd')}/ 
と設定します。
#{@scheduledStartTime.format('YYYY-MM-dd')} の部分がExpressionにあたり、処理予定開始日時をYYYY-MM-ddの形にフォーマットした文字列が代入されます。

また、Data Formatを指定します。create new: Data Formatを選択すると、Others用紙に新規データフォーマットの項目が作成されます。

これで1つdata nodeの設定ができました。現在以下のようになっています。

Name:log
Type:S3DataNode
Schedule:DefaultSchedule1
DirectoryPath:s3://memorycraft-us-log/logs/#{@scheduledStartTime.format('YYYY-MM-dd')}/
DataFormat:DefaultDataFormat1


同じように、2つの出力先も以下のように設定します。

Name:log-archive
Type:S3DataNode
Schedule:DefaultSchedule1
DirectoryPath:s3://memorycraft-us-log-archive/archive/#{@scheduledStartTime.format('YYYY-MM-dd')}/
DataFormat:create new: DataFormat

Name:log-stats
Type:S3DataNode
Schedule:DefaultSchedule1
DirectoryPath:s3://memorycraft-us-log-archive/stats/#{@scheduledStartTime.format('YYYY-MM-dd')}/
DataFormat:create new: DataFormat

ここまでできたら、次は先程追加したScheduleとDataFormatを設定します。



Schedule


data nodeの設定で追加されたScheduleを見てみます。Schedulesアコーディオンを開くと、
先ほど追加されたDefaultSchedule1が一つ登録されています。

これを以下のように設定します。

Name:schedule
Type:Schedule
StartDateTime:2012-12-24 00:00:00
Period:1 Days

StartDateTimeは処理期間、Periodは実行間隔です。
cronというよりはanacronのような仕組みのようで、DataPipelineが稼働し始めると、StartDateTimeが過去日付の場合は、現在に追いつくまでPeriod分のデータの処理を繰り返し処理してくれるようです。




DataFormat


Othersアコーディオンには、DefaultDataFormat1、2、3の3つのDataFormatが出来ています。それぞれDataNodesで設定したlog, log-archive, log-statsで新規追加したDataFormatです。これらはS3DataNodeからHiveを通すときに使用される変換用のマッピングです。

まず、log data nodeに割り当てられているDefaultDataFormat1を設定します。
前回の記事同様、fluentのフォーマットは、日時、fluentタグ、json文字列なので、3つデータカラムを登録します。
データカラムはColumnで設定し、カラム名<半角スペース>データ型の形で記載します。
データ型にはhiveで利用できる型の中から選びます。

ここでは、以下のように設定します。

Name:format-log
Type:TSV
Column:dt STRING
Column:tag STRING
Column:json STRING

同様に、2つの出力先も以下のように設定します。

Name:format-archive
Type:TSV
Column:dt STRING
Column:host STRING
Column:user STRING
Column:method STRING
Column:path STRING
Column:code STRING
Column:size BIGINT
Column:referer STRING
Column:agent STRING

Name:format-stats
Type:TSV
Column:path STRING
Column:cnt BIGINT

ここまででDataNodeに関連する設定はひとまず終了です。
おそらくですが、S3DataNodeとDataFormatを設定することで、実行時にHiveのCREATE EXTERNAL TABLEクエリが自動的に発行されるようです。



Activity


次に、処理要素であるActivityを追加します。ヘッダから「Add activity」ボタンをクリックすると、右ペインのActivitiesに新しいactivityが追加されます。
ここで、まず以下のように設定します。

Name:archive
Type:HiveActivity
Schedule:schedule
Input:log
Output:log-archive

Input、Outputを設定することで、左ペインでactivityとdata nodeが矢印で結ばれます。

続いて、Runs Onでcreate new: Emr Clusterを選択します。
すると、ResourcesにDefaultEmrClusterが追加されます。

続いて、Hive Scriptを設定します。ここにはHQLを入れるのですが、クエリ中にもExpressionを使用します。ここでは、以下のように入力します。

INSERT OVERWRITE TABLE ${output1} SELECT dt,host,user,method,path,code,size,referer,agent FROM ${input1} LATERAL VIEW json_tuple(${input1}.json, 'host', 'user', 'method', 'path', 'code', 'size', 'referer', 'agent') j AS host,user,method, path, code, size, referer, agent WHERE substr(dt, 0, 10) = '#{@scheduledStartTime.format('YYYY-MM-dd')}' ORDER BY dt;

${output1}はactivityにアタッチされたOutput要素の1番目、${input1}はactivityにアタッチされたInput要素の1番目のテーブルがそれぞれ割り当てられるようです。また、#{@scheduledStartTime.format('YYYY-MM-dd')}では、日時が2012-12-24などが動的に入ります。

また、OnSuccess,OnFailなどで設定追加したAction要素にSNSArnを設定しておくと、処理の結果をメールで受け取ることもできるようですが、ここでは割愛します。

同じように、集計用のactivityも追加して、以下のように設定します。

Name:stats
Type:HiveActivity
Schedule:schedule
Input:log
Output:log-stats
HiveScript:INSERT OVERWRITE TABLE ${output1} SELECT path,COUNT(dt) FROM ${input1} LATERAL VIEW json_tuple(${input1}.json, 'host', 'user', 'method', 'path', 'code', 'size', 'referer', 'agent') j AS host,user,method, path, code, size, referer, agent WHERE substr(dt, 0, 10) = '#{@scheduledStartTime.format('YYYY-MM-dd')}' GROUP BY path ORDER BY path;
RunsOn:DefaultEmrCluster1
LogUrl:s3://memorycraft-us-hive-log/pipeline/stats/




Resources


これは、処理を実際に行うリソースです。
HiveActivityを選んだ場合は、リソースはEMRになります。
ここでは、さきほどのactivityで追加されたDefaultEmrCluster1を設定します。

Name:emr-log
Type:EmrCluster
TerminateAfter:2 hours
EmrLogUrl:s3://memorycraft-us-hive-log/emr/
KeyPair:myfirstcloud-use-key
MasterInstanceType:m1.small
EnableDebugging:true
LogUri:s3://memorycraft-us-hive-log/task/
Schedule:schedule
CoreInstanceType:m1.small
CoreInstanceCount:2
InstallHive:latest
ActionOnTaskFailure:continue
ActionOnResourceFailure:retryAll

多くは、実際にEMRのジョブフローを設定するときに見かける項目ですが、
ここでのポイントはEnableDebuggingとLogUri,EmrLogUriです。
これを設定しておくと、EMRの画面でどんなエラーが起きているかがわかるので、慣れないうちは設定しておくと便利です。


これですべての設定が完了しました。

内容をまとめると以下のようになります。

【Activities】

 Name:archive
 Type:HiveActivity
 Schedule:schedule
 Input:log
 Output:log-archive
 HiveScript:INSERT OVERWRITE TABLE ${output1} SELECT dt,host,user,method,path,code,size,referer,agent FROM ${input1} LATERAL VIEW json_tuple(${input1}.json, 'host', 'user', 'method', 'path', 'code', 'size', 'referer', 'agent') j AS host,user,method, path, code, size, referer, agent WHERE substr(dt, 0, 10) = '#{@scheduledStartTime.format('YYYY-MM-dd')}' ORDER BY dt;
 RunsOn:emr-archive
 LogUrl:s3://memorycraft-us-hive-log/pipeline/archive/


 Name:stats
 Type:HiveActivity
 Schedule:schedule
 Input:log
 Output:log-stats
 HiveScript:INSERT OVERWRITE TABLE ${output1} SELECT path,COUNT(dt) FROM ${input1} LATERAL VIEW json_tuple(${input1}.json, 'host', 'user', 'method', 'path', 'code', 'size', 'referer', 'agent') j AS host,user,method, path, code, size, referer, agent WHERE substr(dt, 0, 10) = '#{@scheduledStartTime.format('YYYY-MM-dd')}' GROUP BY path ORDER BY path;
 RunsOn:emr-archive
 LogUrl:s3://memorycraft-us-hive-log/pipeline/stats/


【DataNodes】

 Name:log
 Type:S3DataNode
 Schedule:schedule
 DirectoryPath:s3://memorycraft-us-log/logs/#{@scheduledStartTime.format('YYYY-MM-dd')}/
 DataFormat:format-log
 Name:log-archive
 Type:S3DataNode
 Schedule:schedule
 DirectoryPath:s3://memorycraft-us-log-archive/archive/#{@scheduledStartTime.format('YYYY-MM-dd')}/
 DataFormat:format-archive
 Name:log-stats
 Type:S3DataNode
 Schedule:schedule
 DirectoryPath:s3://memorycraft-us-log-archive/stats/#{@scheduledStartTime.format('YYYY-MM-dd')}/
 DataFormat:format-stats


【Schedules】
 Name:schedule
 Type:Schedule
 StartDateTime:2012-12-24 00:00:00
 Period:1 Days


【Resources】
 Name:emr-log
 Type:EmrCluster
 TerminateAfter:2
 EmrLogUrl:s3://memorycraft-us-hive-log/emr/
 KeyPair:myfirstcloud-use-key
 MasterInstanceType:m1.small
 EnableDebugging:true
 LogUri:s3://memorycraft-us-hive-log/task/
 Schedule:schedule
 CoreInstanceType:m1.small
 CoreInstanceCount:2
 InstallHive:latest
 ActionOnTaskFailure:continue
 ActionOnResourceFailure:retryAll


【Others】
 Name:format-log
 Type:TSV
 Column:st STRING
 Column:tag STRING
 Column:json STRING


 Name:format-archive
 Type:TSV
 Column:dt STRING
 Column:host STRING
 Column:user STRING
 Column:method STRING
 Column:path STRING
 Column:code STRING
 Column:size BIGINT
 Column:referer STRING
 Column:agent STRING

 Name:format-stats
 Type:TSV
 Column:path STRING
 Column:cnt BIGINT

 ここまでの設定を保存します。ヘッダーの「Save pipeline」をクリックします。エラーがあるとこの時点で、右ペインのErrorsにエラーが表示されるのでそれにしたがって修正します。
エラーがなく、保存に成功すると以下のようなSuccessダイアログが出ます。




ここで、右上の「Back to List of Pipeline」のリンクをクリックします。
一覧ではPipelineが「PENDING」の状態になっています。
まだ設定を保存しただけで、このままでは実行されません。

次にアクティベートという処理を行います。
設定した内容のほとんどをFixし、実行もしくは実行待機状態にするための作業です。


もう一度「View pipeline」をクリックして、詳細画面に入ります。
ヘッダーに「Activate」ボタンがあるので、クリックします。
するとアクティベートに成功した旨のダイアログが現れます。




再度一覧に戻ると、ステータスが「SCHEDULED」になっています。
もし設定したScheduleが過去日付であれば、この時点で処理が開始されます。
未来日時であれば該当日時になると処理が開始されます。


処理の観察



この時点で、「View pipeline」を見ると、処理の一覧が表示されます。data nodeを用意したり、EMRフローを実行したりなどが表示されます。




個人的にはEMRのジョブフローのログを見たほうがわかりやすいので、EMRの方を見てみます。
すると、過去日付を設定したので、即座にジョブフローが立ち上がっているのがわかります。
また、ジョブフロー中のInstall TaskRunnerの内容で、何日のスケジュールを実行しているのかが判断できそうです。




そして、ジョブフローを選択し、上部の「Debug」ボタンをクリックします。
ジョブフローのデバッグダイアログでは、処理が完了し、出力の終わったログを見ることができます。







結果



処理が終わったようなので、S3を見てみます。




おお!ファイルが出力されているようです。
内容を見てみると、、、、


"2012-12-24T14:16:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:16:22+00:00""10.118.137.132""-""GET""/assets/img/logo.png""404""329""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:16:23+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:25+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:26+00:00""10.118.137.132""-""GET""/assets/img/logo.png""404""329""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:27+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:28+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:32+00:00""10.118.137.132""-""GET""/img/logo.png""404""322""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:32+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:34+00:00""10.118.137.132""-""GET""/assets/img/loading.gif""404""332""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:34+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:34+00:00""10.118.137.132""-""GET""/img/logo.png""404""322""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:35+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:36+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:37+00:00""10.118.137.132""-""GET""/img/loading.gif""404""325""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:37+00:00""10.118.137.132""-""GET""/favicon.ico""404""321""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
"2012-12-24T14:16:38+00:00""10.118.137.132""-""GET""/assets/css/bootstrap.css""404""334""-""Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_2) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.101 Safari/537.11"
~~ 略 ~~
"2012-12-24T14:46:53+00:00""::1""-""OPTIONS""*""200""-""-""Apache/2.2.23 (Amazon) (internal dummy connection)"
"2012-12-24T14:47:01+00:00""::1""-""OPTIONS""*""200""-""-""Apache/2.2.23 (Amazon) (internal dummy connection)"
"2012-12-24T14:47:03+00:00""::1""-""OPTIONS""*""200""-""-""Apache/2.2.23 (Amazon) (internal dummy connection)"
"2012-12-24T14:47:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:47:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:47:34+00:00""::1""-""OPTIONS""*""200""-""-""Apache/2.2.23 (Amazon) (internal dummy connection)"
"2012-12-24T14:47:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:47:45+00:00""::1""-""OPTIONS""*""200""-""-""Apache/2.2.23 (Amazon) (internal dummy connection)"
"2012-12-24T14:47:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:48:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:48:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:48:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:48:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:49:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:49:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:49:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:49:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:50:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:50:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:50:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:50:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:51:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:51:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:51:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:51:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:52:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:52:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:52:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:52:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:53:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:53:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:53:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:53:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:54:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:54:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:54:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:54:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:55:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:55:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:55:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:55:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:56:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:56:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:56:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:56:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:57:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:57:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:57:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:57:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:58:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:58:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:58:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:58:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:59:13+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:59:22+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:59:43+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"
"2012-12-24T14:59:52+00:00""10.118.137.132""-""GET""/health.txt""200""5""-""ELB-HealthChecker/1.0"


"*""16"
"/""19"
"/assets/css/bootstrap.css""5"
"/assets/img/loading.gif""3"
"/assets/img/logo.png""10"
"/assets/js/bootstrap.js""4"
"/assets/js/client.js""3"
"/assets/js/jquery.min.js""5"
"/favicon.ico""149"
"/health.txt""165"
"/icons/apache_pb2.gif""19"
"/img/loading.gif""1"
"/img/log.png""2"
"/img/logo.png""3"
"/signin""6"
"/signup""6"

おお、うまく出力されているようです!
ここまでで、ログの移行と処理がすべてAWS上で行えるようになりました。
このログにはELBからのヘルスチェックも含まれていますが、HiveQLのWHERE句一発で簡単に除外できますね。

今回Scheduleの調査が中途半端でした。
たとえばStartDateTime が 2012-12-25 15:30:00で、Periodが1dayの場合、次回が2012-12-26 15:30:00になるのか、2012-12-26 00:00:00になるのか、StartDateTimeが毎回の実行日時に影響するのかは追って調査したいと思います。
どちらにしてもTokyoリージョンに来た場合にはJSTに対応してほしいですね。。

今回はHiveActivityだったため、ShellCommandやEC2リソースなどもあるので、いろいろ試せそうです。
結局、DataPipelineというのは、データ同士や処理プロセスを疎結合にすることで、データを抽象的に扱うためのツールなのかもしれません。

それにしてもAWSは留まることを知りませんねー。
以上です。