MRJobで複数ステップのMap-Reduceを実行する

"def steps(self)" メソッドをoverrideして、各ステップにどのような処理を実施するか指定します。

MRJobのサンプルページに載ってる例だと↓な感じ。

def steps(self):
    return [self.mr(mapper=self.transform_input,
                    reducer=self.consolidate_1),
            self.mr(reducer_init=self.log_mapper_init,
                    reducer=self.consolidate_2)]

steps関数の戻り値として、配列で各ステップごとにどんな振る舞いをさせるかをmr関数の戻り値として記載するようになっています。複数要素の配列を返せば、複数のステップが実行されます。
mr関数に指定できる振る舞いは、以下のようです。

• mapper
• reducer
• combiner
• mapper_init
• mapper_final
• reducer_init
• reducer_final
• combiner_init
• combiner_final

ナイーブにMRJobを動かすと、１ステップごとに、outout出力されたファイルをS3から手元の環境にコピーするようになっています。その転送コストがかなりオーバーヘッドが高いので、S3上で出来る限り処理をしたい…という時に、複数ステップで実行すると嬉しいケースが多いです。