Hadoop オリンピア大会のパーティショニング

はじめに

古代ギリシャのオリンピアでは、各地からのアスリートが集まり、その腕前を披露し、様々な競技イベントに挑戦しました。そんなアスリートの一人であるアレクシオスは、近い将来の試合に向けて必死にトレーニングを重ね、彼の都市国家に栄光をもたらす決意を固めていました。

目的は、参加者をイベントごとに分類してグループ化することで、公正で効率的な競争を保証することでした。しかし、何百人ものアスリートが栄光をかけて競う中で、彼らをそれぞれのイベントに振り分けることは、脅威的な作業でした。

Skills Graph

Mapper を実装する

このステップでは、入力データを読み取り、パーティショナが処理するためのキーと値のペアを生成する Mapper クラスを作成します。

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

su - hadoop

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

touch /home/hadoop/OlympicMapper.java

OlympicMapper.java ファイルに次のコードを追加します。

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

public class OlympicMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] fields = value.toString().split(",");
        String athlete = fields[0];
        String event = fields[1];
        context.write(new Text(event), new Text(athlete));
    }
}

OlympicMapper クラスでは、入力キーを LongWritable（行オフセットを表す）とし、入力値を Text（入力ファイルの 1 行のテキストを表す）として定義します。出力キーはイベントを表す Text オブジェクトであり、出力値は選手の名前を表す Text オブジェクトです。

map メソッドは、入力データの各行をコンマ区切りで分割し、選手の名前とイベントを抽出し、イベントをキーとし、選手の名前を値とするキーと値のペアを emit します。

パーティショナを実装する

このステップでは、イベントに基づいてキーと値のペアを分割するカスタム パーティショナ クラスを作成します。

まず、パーティショナ クラス用の Java ファイルを作成します。

touch /home/hadoop/OlympicPartitioner.java

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

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class OlympicPartitioner extends Partitioner<Text, Text> {

    @Override
    public int getPartition(Text key, Text value, int numPartitions) {
        return Math.abs(key.hashCode() % numPartitions);
    }
}

OlympicPartitioner クラスは、Hadoop が提供する Partitioner クラスを拡張します。これは、キー、値、およびパーティション数を入力として受け取る getPartition メソッドをオーバーライドします。

getPartition メソッドは、イベント (キー) のハッシュ コードを計算し、ハッシュ コードの絶対値をパーティション数で割った余りを返すことでパーティション番号を返します。これにより、同じイベントを持つすべてのレコードが、Reducer による処理のために同じパーティションに送信されることが保証されます。

リデューサを実装する

このステップでは、分割されたデータを処理し、最終出力を生成する Reducer クラスを作成します。

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

touch /home/hadoop/OlympicReducer.java

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

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

public class OlympicReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        StringBuilder athletes = new StringBuilder();
        for (Text value : values) {
            athletes.append(value.toString()).append(",");
        }
        if (athletes.length() > 0) {
            athletes.deleteCharAt(athletes.length() - 1);
        }
        context.write(key, new Text(athletes.toString()));
    }
}

OlympicReducer クラスは、Hadoop が提供する Reducer クラスを拡張します。入力キーを Text（イベントを表す）とし、入力値を Text（選手の名前を表す）とし、出力キーと値を Text オブジェクトとして定義します。

reduce メソッドは、各一意のイベント キーに対して呼び出され、そのイベントに関連付けられた選手名のイテレータが渡されます。このメソッドは、各イベントに対してコンマ区切りの選手のリストを作成し、イベントをキーとし、選手のリストを値とするキーと値のペアを emit します。

ドライバを書く

このステップでは、Mapper、Partitioner、Reducer クラスを結び付け、MapReduce ジョブを実行する Driver クラスを作成します。

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

touch /home/hadoop/OlympicDriver.java

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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 OlympicDriver {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Olympic Partitioner");

        job.setJarByClass(OlympicDriver.class);
        job.setMapperClass(OlympicMapper.class);
        job.setPartitionerClass(OlympicPartitioner.class);
        job.setReducerClass(OlympicReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true)? 0 : 1);
    }
}

OlympicDriver クラスは、MapReduce ジョブのエントリ ポイントです。このクラスはジョブ構成を設定し、Mapper、Partitioner、Reducer クラスを指定し、入力および出力パスを構成します。

main メソッドでは、新しい Configuration オブジェクトとジョブ名 "Olympic Partitioner" の Job インスタンスを作成します。対応するセッター メソッドを使用して Mapper、Partitioner、Reducer クラスを設定します。

また、出力キーと値のクラスを Text に設定します。入力および出力パスは、ドライバに渡されるコマンド ライン引数を使用して指定されます。

最後に、Job インスタンスの waitForCompletion メソッドを呼び出して MapReduce ジョブを実行し、適切なステータス コード (成功時は 0、失敗時は 1) で終了します。

ジョブを実行するには、Java クラスをコンパイルして jar ファイルを作成する必要があります。その後、次のコマンドを使用して jar ファイルを実行できます。

javac -source 8 -target 8 -classpath "/home/hadoop/:/home/hadoop/hadoop/share/hadoop/common/hadoop-common-3.3.6.jar:/home/hadoop/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.3.6.jar:/home/hadoop/hadoop/share/hadoop/common/lib/*" -d /home/hadoop /home/hadoop/OlympicMapper.java /home/hadoop/OlympicPartitioner.java /home/hadoop/OlympicReducer.java /home/hadoop/OlympicDriver.java
jar cvf olympic.jar *.class
hadoop jar olympic.jar OlympicDriver /input /output

最後に、次のコマンドを実行して結果を確認できます。

hadoop fs -cat /output/*

出力例:

Event_1 Athlete_17,Athlete_18,Athlete_79,Athlete_71,Athlete_77,Athlete_75,Athlete_19,Athlete_24,Athlete_31,Athlete_32,Athlete_39,Athlete_89,Athlete_88,Athlete_87,Athlete_100,Athlete_13,Athlete_52,Athlete_53,Athlete_58
Event_2 Athlete_1,Athlete_97,Athlete_96,Athlete_85,Athlete_81,Athlete_80,Athlete_72,Athlete_68,Athlete_64,Athlete_61,Athlete_54,Athlete_48,Athlete_47,Athlete_43,Athlete_28,Athlete_23,Athlete_21,Athlete_15,Athlete_12,Athlete_3
Event_3 Athlete_11,Athlete_55,Athlete_8,Athlete_46,Athlete_42,Athlete_41,Athlete_40,Athlete_38,Athlete_33,Athlete_92,Athlete_29,Athlete_27,Athlete_25,Athlete_93,Athlete_22,Athlete_20,Athlete_98,Athlete_14,Athlete_69,Athlete_99,Athlete_66,Athlete_65
Event_4 Athlete_90,Athlete_50,Athlete_37,Athlete_36,Athlete_91,Athlete_74,Athlete_73,Athlete_63,Athlete_26,Athlete_78,Athlete_5,Athlete_62,Athlete_60,Athlete_59,Athlete_82,Athlete_4,Athlete_51,Athlete_86,Athlete_2,Athlete_94,Athlete_7,Athlete_95
Event_5 Athlete_34,Athlete_76,Athlete_57,Athlete_56,Athlete_30,Athlete_16,Athlete_6,Athlete_10,Athlete_83,Athlete_84,Athlete_70,Athlete_45,Athlete_44,Athlete_49,Athlete_9,Athlete_67,Athlete_35

まとめ

この実験では、古代ギリシャのオリンピア大会に着想を得たシナリオを設計することで、Hadoop Shuffle Partitioner の概念を探りました。入力データを読み取り、キーと値のペアを生成する Mapper クラス、イベントに基づいてデータを分割するカスタム パーティショナ クラス、および分割されたデータを処理して最終出力を生成する Reducer クラスを実装しました。

この実験を通じて、MapReduce プログラミング モデルに関する実践的な経験を得るとともに、パーティショナ クラスを活用してデータを効率的にパーティションに分散させる方法を学びました。古代ギリシャのオリンピア大会のシナリオは、現実世界の設定における Shuffle Partitioner の実際の応用を理解するのに魅力的なコンテキストを提供しました。