mrjob で手軽に elastic map reduce

http://pythonhosted.org/mrjob/

mrjobは、AWSのElastic MapReduceサービスを手軽に使用するためのPython実装です。
EMRのクライアントライブラリとして有名な"Amazon Elastic MapReduce Ruby Client( http://aws.amazon.com/developertools/2264 ) "と比べると出来ることは少ない印象ですが、その分シンプルに使うことができます。


◯インストール

pip install mrjob

◯jobの作成
mrjobに含まれている"MRJob" というクラスを継承することで、手軽にmapper / reducerのjob作成する事ができます。以下はmrjobのサイトにも記載されている、word countを行うサンプルのソース。

#mrjob_sample.py
from mrjob.job import MRJob


class MRWordCounter(MRJob):

    def mapper(self, key, line):
        for word in line.split():
            yield word, 1

    def reducer(self, word, occurrences):
        yield word, sum(occurrences)


if __name__ == '__main__':
    MRWordCounter.run()

◯ローカル環境で実行
ローカル環境で作成したjobの動作確認ができます。

python mrjob_sample.py test.txt > result.txt

test.txtの中身が以下のようなものだった場合

A job consists of an MRJob subclass, one or more mapper, combiner, or reducer methods, a call at the bottom to invoke mrj    ob’s logic, and all associated dependencies.
For jobs that consist only of one mapper, combiner, and reducer, in that order, you can simply override one or more of th    e mapper(), combiner(), and reducer() methods:

MapReduceの実行結果が出力されたresult.txtの中身は以下のようになります。

"A" 1
"For"   1
"MRJob" 1
"a" 1
"all"   1
"an"    1
"and"   3
"associated"    1
"at"    1
"bottom"    1
(snip.)

◯EMRの設定
EMRの環境でjobを実行するには、設定ファイルを作成する必要があります。
この中でAWSのcertificate情報や、起動するインスタンスの種類などを指定します。
以下は最低限動作させるための一例です。詳しくは公式サイトのドキュメントを参考ください。
http://pythonhosted.org/mrjob/guides/quickstart.html#configuration

#emr.conf
runners:
    emr:
        aws_access_key_id: [access_key_id]
        aws_region: [region]
        aws_secret_access_key: [secret_access_key]
        ec2_instance_type: m1.small
        num_ec2_instances: 3
        cmdenv:
        s3_log_uri: s3://[bucket_name]/emr/logs/

◯EMR環境で実行
emr環境で実行する場合は" -r emr" オプションを指定します。また、解析対象のデータのs3のパスを指定します。

python mrjob_sample.py -r emr --conf-path ./emr.conf s3://[bucket_name]/[file_path] > result.txt

◯バグ？
実は、pip installでmrjobをダウンロードした場合、上記の手順では正常にEMR上で動作させる事ができませんでした。
ローカル実行は可能ですが、EMR上で動作させる際に"returned an invalid certificate"というエラーメッセージが表示し、emrのエンドポイントへの接続が正しく行われませんでした。

同様のエラーが以下issueに報告されていますが、現在は未だ修正が行われていないようです。
https://github.com/Yelp/mrjob/issues/621

上記issueにも記載されているように、mrjobのemr.py中に記載されているEMRのendpoint URLを

    #'ap-northeast-1': 'elasticmapreduce.ap-northeast-1.amazonaws.com',
    'ap-northeast-1': 'ap-northeast-1.elasticmapreduce.amazonaws.com',

という形で書き換えてあげると、うまく動作するようになります。



mrjobはシンプルとはいえ、色々な記述方法、設定方法があり、各種管理コマンドも用意されているようなので、以降も気が向いたら記事を書いていこうと思います。