Vorbereiten des Datensatzes und des Codes
In diesem Schritt werden wir die erforderlichen Dateien und den Code einrichten, um das antike Datenverarbeitungssystem zu simulieren.
Ändern Sie zunächst den Benutzer in hadoop und wechseln Sie dann in das Home-Verzeichnis des hadoop-Benutzers:
su - hadoop
Erstellen Sie ein neues Verzeichnis namens distributed-cache-lab und navigieren Sie dorthin:
mkdir distributed-cache-lab
cd distributed-cache-lab
Als nächstes erstellen Sie eine Textdatei namens ancient-texts.txt mit dem folgenden Inhalt:
Die Weisheit der Jahrhunderte ist ewig.
Wissen ist der Weg zur Erleuchtung.
Nehmen Sie die Geheimnisse des Universums an.
Diese Datei wird die antiken Texte repräsentieren, die wir verarbeiten möchten.
Erstellen Sie jetzt eine Java-Datei namens AncientTextAnalyzer.java mit dem folgenden Code:
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.StringTokenizer;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class AncientTextAnalyzer {
public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
}
}
}
public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
context.write(key, result);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length!= 2) {
System.err.println("Usage: AncientTextAnalyzer <in> <out>");
System.exit(2);
}
Job job = Job.getInstance(conf, "Ancient Text Analyzer");
job.setJarByClass(AncientTextAnalyzer.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true)? 0 : 1);
}
}
Dieser Code ist ein einfaches MapReduce-Programm, das die Vorkommen jedes Worts in der Eingabedatei zählt. Wir werden diesen Code verwenden, um die Verwendung des verteilten Caches in Hadoop zu demonstrieren.
