Hadoop Shuffle Combiner を用いたデータ処理の最適化

はじめに

宇宙戦闘シミュレーションシステムにおけるデータ管理を担当する才能あるエンジニアであると想像してみてください。あなたの目標は、MapReduce プロセスにおいて Hadoop Shuffle Combiner テクニックを実装することにより、システムのパフォーマンスを最適化することです。Combiner を活用することで、シミュレーション中のデータ処理におけるネットワークトラフィックを削減し、全体的な効率を向上させることを目指しています。

Mapper を記述する

このステップでは、入力データを処理して中間のキー-値ペアを生成する Mapper クラスを記述します。

端末を開き、以下の手順に従って始めましょう。

ユーザをhadoopに変更し、次にhadoopユーザのホームディレクトリに切り替えます。

su - hadoop

Mapper クラス用の Java ファイルを作成します。

nano /home/hadoop/SpaceBattleMapper.java

次に、SpaceBattleMapper.javaファイルに以下のコードを追加します。

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import java.io.IOException;

public class SpaceBattleMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 入力行を単語に分割する
        String[] words = value.toString().split("\\s+");
        // 各単語に対してキー-値ペアを生成する
        for (String w : words) {
            word.set(w);
            context.write(word, one);
        }
    }
}

ヒント：右側のプロンプトボックスからコードをコピーし、Ctrl + Shift + Vで開いている nano エディタに貼り付けることができます。nano エディタからのプロンプトが表示されたときに、Ctrl + Oを押してファイルを保存し、Enterを押して確認します。最後に、Ctrl + Xを使ってエディタを終了します。

SpaceBattleMapperクラスは、Hadoop フレームワークのMapperクラスを拡張しています。これは、キー-値ペア形式の入力データを処理するために使用されます。ここで、キーは入力ファイル内の行のバイトオフセットを表すLongWritableで、値はテキストの行を表すTextオブジェクトです。

このクラスは 2 つのプライベートフィールドを定義しています。

one：定数値 1 を持つIntWritableオブジェクト。これは生成されるキー-値ペアの値として使用されます。
word：入力行から抽出された各単語を格納するためのTextオブジェクト。

mapメソッドは、特定のマッピングロジックを提供するためにオーバーライドされています。

入力のText値を文字列に変換し、空白を基準に単語に分割します。
配列内の各単語に対して、wordオブジェクトをその単語に設定し、キーとしてその単語と値としてoneを持つキー-値ペアを生成します。これはcontext.writeメソッドを使用して行われます。

この Mapper クラスは、入力データ内の各単語に対して、キーとしてその単語と値として整数 1 を持つキー-値ペアを生成するように設計されています。この設定は、データセット内の各単語の出現回数をカウントすることを目的とする単語カウントアプリケーションで一般的に使用されます。

Combiner を実装する

このステップでは、データのシャッフルの前にローカルな集約を行うために Combiner クラスを実装します。

Combiner クラス用の Java ファイルを作成します。

nano /home/hadoop/SpaceBattleCombiner.java

次に、SpaceBattleCombiner.javaファイルに以下のコードを追加します。

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

public class SpaceBattleCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        // 各キーの値を合計する
        for (IntWritable val : values) {
            sum += val.get();
        }
        // キーとその値の合計を出力する
        context.write(key, new IntWritable(sum));
    }
}

SpaceBattleCombinerクラスは、Hadoop フレームワークのReducerクラスを拡張しています。これは、MapReduce プロセスにおいて Combiner として使用され、Mapper によって生成された中間のキー-値ペアのローカルな集約を行います。

このクラスは、特定の Combiner ロジックを提供するためにreduceメソッドをオーバーライドしています。

このメソッドは、Text型のキーとIntWritable型の値の反復可能オブジェクトを入力として受け取ります。キーは単語を表し、反復可能オブジェクトはその単語の出現回数を含んでいます。
このメソッドは値を反復処理し、それらを合計して単語の合計数を取得します。
最後に、このメソッドはcontext.writeメソッドを使用して、キーとして単語と値として合計数を持つキー-値ペアを出力します。

SpaceBattleCombinerの目的は、データがネットワークを介して Reducer にシャッフルされる前に、各単語のカウントのローカルな集約を行うことです。これにより、Mapper と Reducer のフェーズ間で転送されるデータ量が削減され、MapReduce ジョブの効率が向上します。

Reducer を実装する

このステップでは、キー-値ペアの最終的な集約を行うために Reducer クラスを実装します。

Reducer クラス用の Java ファイルを作成します。

nano /home/hadoop/SpaceBattleReducer.java

次に、SpaceBattleReducer.javaファイルに以下のコードを追加します。

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

public class SpaceBattleReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

SpaceBattleReducerクラスは、Hadoop フレームワークのReducerクラスを拡張しています。これは、Mapper によって生成され、任意に Combiner によって処理された中間のキー-値ペアの最終的な集約を行うために使用されます。

このクラスは、特定の Reducer ロジックを提供するためにreduceメソッドをオーバーライドしています。

このメソッドは、Text型のキーとIntWritable型の値の反復可能オブジェクトを入力として受け取ります。キーは単語を表し、反復可能オブジェクトはその単語の出現回数を含んでいます。
このメソッドは値を反復処理し、それらを合計して単語の合計数を取得します。
最後に、このメソッドはcontext.writeメソッドを使用して、キーとして単語と値として合計数を持つキー-値ペアを出力します。

SpaceBattleReducerは、すべての Mapper 出力にわたる各単語のカウントを合計することで、データの最終的な集約を行います。これにより、入力データ内の各単語の最終的な出現回数が提供されます。

Driver を記述する

このステップでは、MapReduce ジョブを管理する Java ファイルを作成します。これには、ジョブの設定を行い、Mapper、Combiner、Reducer クラスを指定することが含まれます。

Driver クラス用の Java ファイルを作成します。

nano /home/hadoop/SpaceBattleDriver.java

次に、SpaceBattleDriver.javaファイルに以下のコードを追加します。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SpaceBattleDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Space Battle Simulation");
        job.setJarByClass(SpaceBattleDriver.class);
        job.setMapperClass(SpaceBattleMapper.class);
        job.setCombinerClass(SpaceBattleCombiner.class);
        job.setReducerClass(SpaceBattleReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true)? 0 : 1);
    }
}

SpaceBattleDriverクラスは、宇宙戦闘シミュレーションの MapReduce ジョブを構成して実行する責任があります。

このクラスは、新しいConfigurationオブジェクトとこの設定を持つJobインスタンスを作成することから始まります。ジョブには識別用の名前「Space Battle Simulation」が付けられています。
setJarByClassメソッドにSpaceBattleDriverクラスを渡して、ジョブに必要なクラスが含まれる jar ファイルを設定します。
setMapperClass、setCombinerClass、setReducerClassメソッドを使用して、それぞれマッピング、結合、削減タスクを実行するクラスを指定します。
setOutputKeyClassとsetOutputValueClassメソッドで、出力キーと値の型を定義します。この場合、それぞれText型とIntWritable型です。
FileInputFormat.addInputPathとFileOutputFormat.setOutputPathメソッドで、入力データと出力データのパスを設定します。これらのパスは、main メソッドにコマンドライン引数として渡されます。
最後に、job.waitForCompletionメソッドを呼び出してジョブを送信し、完了を待ちます。ジョブが正常に完了するとtrueを返し、そうでなければfalseを返します。ジョブが成功するとプログラムはステータスコード 0 で終了し、失敗すると 1 で終了します。

この Driver クラスは、MapReduce ジョブのすべてのコンポーネントを結び付け、ジョブを実行するエントリポイントです。

まとめ

この実験では、宇宙戦闘シミュレーションのシナリオで Hadoop Shuffle Combiner 技術の実装を学びました。Mapper、Combiner、Reducer、Driver クラスを作成する手順に従うことで、Hadoop MapReduce 環境におけるデータ処理の最適化に関する実践的な経験を得ました。この実験の目的は、ビッグデータ処理タスクにおけるネットワークオーバーヘッドの削減と計算効率の向上に関する理解を深めることでした。