結合操作の実装
このステップでは、MapReduce ジョブを実装して、dinosaurs.txt と locations.txt ファイルに対して結合操作を行います。
/home/hadoop/join-lab ディレクトリに、以下の内容を持つ新しい Java ファイル JoinDinosaurs.java を作成します。
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class JoinDinosaurs {
public static class JoinMapper extends Mapper<LongWritable, Text, Text, Text> {
private final Text outKey = new Text();
private final Text outValue = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
String[] parts = line.split(",");
if (parts.length == 2) { // locations.txt
outKey.set(parts[0]);
outValue.set("LOC:" + parts[1]);
} else if (parts.length == 3) { // dinosaurs.txt
outKey.set(parts[0]);
outValue.set("DIN:" + parts[1] + "," + parts[2]);
}
context.write(outKey, outValue);
}
}
public static class JoinReducer extends Reducer<Text, Text, Text, Text> {
private final Text outValue = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
Map<String, String> dinMap = new HashMap<>();
StringBuilder locBuilder = new StringBuilder();
for (Text value : values) {
String valStr = value.toString();
if (valStr.startsWith("DIN:")) {
dinMap.put("DIN", valStr.substring(4));
} else if (valStr.startsWith("LOC:")) {
locBuilder.append(valStr.substring(4)).append(",");
}
if (locBuilder.length() > 0) {
locBuilder.deleteCharAt(locBuilder.length() - 1);
}
}
StringBuilder outBuilder = new StringBuilder();
for (Map.Entry<String, String> entry : dinMap.entrySet()) {
outBuilder.append(entry.getValue()).append("\t").append(locBuilder.toString().trim());
}
outValue.set(outBuilder.toString());
context.write(key, outValue);
}
}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
if (args.length!= 2) {
System.err.println("Usage: JoinDinosaurs <input_dir> <output_dir>");
System.exit(1);
}
Job job = Job.getInstance();
job.setJarByClass(JoinDinosaurs.class);
job.setJobName("Join Dinosaurs");
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setMapperClass(JoinMapper.class);
job.setReducerClass(JoinReducer.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
System.exit(job.waitForCompletion(true)? 0 : 1);
}
}
このコードは、カスタムの JoinMapper と JoinReducer を持つ MapReduce ジョブを定義しています。マッパーは dinosaurs.txt と locations.txt から入力データを読み取り、恐竜の名前をキーとして、データ型（"DIN" または "LOC"）と対応する値を含むキーバリューペアを出力します。その後、リデューサはキーで値をグループ化し、恐竜の情報と場所を結合することで結合操作を行います。
コードをコンパイルするには、以下のコマンドを実行します。
mkdir classes
javac -source 8 -target 8 -cp "/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 classes JoinDinosaurs.java
jar -cvf join-dinosaurs.jar -C classes/.
次に、以下のコマンドを使用して MapReduce ジョブを実行します。
hadoop jar join-dinosaurs.jar JoinDinosaurs /home/hadoop/join-lab /home/hadoop/join-lab/output
このコマンドは、join-dinosaurs.jar ファイルから JoinDinosaurs クラスを実行し、入力ディレクトリ /home/hadoop/join-lab（dinosaurs.txt と locations.txt が含まれています）と出力ディレクトリ /home/hadoop/join-lab/output を指定します。
ジョブが正常に完了した後、/home/hadoop/join-lab/output ディレクトリに出力を表示できます。
