S3 上の大量データを EMR するときは S3DistCp を使うと捗る
CloudFront のアクセスログを S3 に出力するように設定していると、日に日に大量のログファイルが溜まっていきます。
中には 1 MB 程度のファイルもあれば、数百 bytes 程度の小さなものもあります。
何にせよ、とにかく大量にあります。
小さいファイルはまとめる
CloudFront のログのように小さいファイルが大量にあるようなケースは Hadoop が苦手とするパターンで、そのまま扱うと大抵めちゃんこ時間がかかります。
そういう場合は、S3DistCp という便利ツールを使って S3 上のログファイルをもうすこし大きいサイズに連結して、HDFS にコピーするとパフォーマンスがあがることがあります。
ちなみに、EMR のベストプラクティスによると、Gzip で圧縮されている場合は 1 ~ 2GB、LZO で圧縮されている場合は 2 ~ 4GB が最適なファイルサイズらしいです。
S3DistCp の使用例
S3DistCp を使うと S3 上のログデータをファイル名によってフィルタリングしたり、グルーピングしたり、圧縮フォーマットを変更したりできます。
以下 SSH でマスターノードにログインして、ホームディレクトリから実行する例をあげていきます。
また、S3 に上がっているログは CloudFront のログで Gzip 圧縮されている前提です。
圧縮なし & 1 時間ごとにグルーピング & フィルタリングなし
hadoop jar lib/emr-s3distcp-1.0.jar \ --src s3://logs.example.com/cf/ \ --dest hdfs://`hostname -f`:9000/data/ \ --groupBy '.*ABCDEFG1234567.([0-9]+-[0-9]+-[0-9]+-[0-9]+).*' \ --targetSize 2048 \ --outputCodec none
CloudFront のログは {distribution-ID}.{YYYY}-{MM}-{DD}-{HH}.{unique-ID}.gz みたいな規則で命名されています。
なので、--groupBy で上のようにグルーピングを指定することで、1 時間ごとに連結されたファイルが生成されます。
--targetSize は MB 単位で指定します。
--groupBy で連結されたファイルが --targetSize を超えた場合は、複数ファイルに分割されます。
--outputCodec は圧縮フォーマットを指定します。ここでは圧縮しない設定として none を指定しています。
各オプションの詳細、その他のオプションについてはこちら。
Gzip 圧縮 & 1 時間ごとにグルーピング & フィルタリングなし
hadoop jar lib/emr-s3distcp-1.0.jar \ --src s3://logs.example.com/cf/ \ --dest hdfs://`hostname -f`:9000/data/ \ --groupBy '.*ABCDEFG1234567.([0-9]+-[0-9]+-[0-9]+-[0-9]+).*(\.gz)' \ --targetSize 2048
--outputCodec を省略するとコピー元と同様の圧縮フォーマットでコピーしてくれます。
CloudFront のログは Gzip 圧縮されているので、生成されるファイルを Gzip で圧縮したい場合は --outputCodec を省略します。
ここで --outputCodec に gzip を指定すると、2 重圧縮されたファイルが生成されてしまうという悲しい状態になってしまいます。 (ドキュメントをちゃんと読んでいなかったせいでここでかなりハマりました...)
また、ここで --outputCodec を省略すると、生成されるファイルは Gzip ファイルにもかかわらず gz 拡張子をつけてくれません。
これでは Hive 等からロードする場合、Gzip ファイルとして認識してくれません。
したがって、--groupBy オプションの指定を上のようにすることで、生成されるファイル名の末尾に .gz とつくようにしています。
Gzip 圧縮 & 1 日ごとにグルーピング & フィルタリング指定なし
hadoop jar lib/emr-s3distcp-1.0.jar \ --src s3://logs.example.com/cf/ \ --dest hdfs://`hostname -f`:9000/data/ \ --groupBy '.*ABCDEFG1234567.([0-9]+-[0-9]+-[0-9]+)-[0-9]+.*(\.gz)' \ --targetSize 2048
--groupBy の指定をすこし変えて 1 日ごとにグルーピングするようにしています。
ファイルサイズが小さく、対象とする日付の範囲も広いような場合には 1 日単位でまとめちゃってもいいかもしれません。
Gzip 圧縮 & 1 日ごとにグルーピング & 日付指定 (2014-05-19 ~ 2014-05-20) でフィルタリング
hadoop jar lib/emr-s3distcp-1.0.jar \ --src s3://logs.example.com/cf/ \ --dest hdfs://`hostname -f`:9000/data/ \ --srcPattern '.*ABCDEFG1234567.2014-05-(19|20)-[0-9]+.*' \ --groupBy '.*ABCDEFG1234567.([0-9]+-[0-9]+-[0-9]+)-[0-9]+.*(\.gz)' \ --targetSize 2048
--srcPattern を指定することでマッチするファイル名だけにフィルタリングすることができます。
特定の日時の範囲に絞ることが許されるのであれば、こういうフィルタリングをしておくといいかもしれません。
LZO 圧縮について
通常、ソース側が非圧縮状態であれば --outputCodec で lzo と指定することで LZO で圧縮されるようになります。
ただ、ソース側が Gzip 圧縮されている状態で LZO 圧縮するように指定すると、二重圧縮されるようになってしまいます。
これは S3DistCp のバグのようで、次回の AMI のリリースにあわせて修正されていると嬉しいなぁという感じです。
ちなみに、2014/05/21 現時点での最新の AMI のバージョンは 3.1.0 で、S3DistCp のバージョンは 1.0 です。
