Implementieren der Join-Operation
In diesem Schritt werden wir einen MapReduce-Job implementieren, um eine Join-Operation auf den Dateien dinosaurs.txt und locations.txt durchzuführen.
Erstellen Sie in das Verzeichnis /home/hadoop/join-lab eine neue Java-Datei namens JoinDinosaurs.java mit dem folgenden Inhalt:
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);
}
}
Dieser Code definiert einen MapReduce-Job mit einem benutzerdefinierten JoinMapper und JoinReducer. Der Mapper liest die Eingabedaten aus dinosaurs.txt und locations.txt und emittiert Schlüssel-Wert-Paare mit dem Namen des Dinosauriers als Schlüssel und dem Datentyp ("DIN" oder "LOC") zusammen mit dem entsprechenden Wert. Der Reducer führt dann die Join-Operation durch, indem er die Werte nach Schlüssel gruppiert und die Dinosaurierinformationen mit dem Standort kombiniert.
Um den Code zu kompilieren, führen Sie folgenden Befehl aus:
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/.
Als Nächstes führen Sie den MapReduce-Job mit dem folgenden Befehl aus:
hadoop jar join-dinosaurs.jar JoinDinosaurs /home/hadoop/join-lab /home/hadoop/join-lab/output
Dieser Befehl führt die Klasse JoinDinosaurs aus der Datei join-dinosaurs.jar aus, mit dem Eingabeverzeichnis /home/hadoop/join-lab (enthaltend dinosaurs.txt und locations.txt) und dem Ausgabeverzeichnis /home/hadoop/join-lab/output.
Nachdem der Job erfolgreich abgeschlossen ist, können Sie die Ausgabe im Verzeichnis /home/hadoop/join-lab/output ansehen.
