MapReduce Apache Hadoop Technology

Question

Я пытаюсь использовать программу wordcount с использованием технологии MapReduce Hadoop. Мне нужно создать приложение Indexed Word Count, которое будет подсчитывать количество вхождений каждого слова в каждом файле в заданном наборе входных файлов. Этот набор файлов присутствует в ведро Amazon S3. Он также будет учитывать общее количество каждого слова. Я прикрепил код, который учитывает вхождения слов в данном наборе файлов. После этого мне нужно напечатать то, что происходит в каком слове, в каком файле содержится число вхождений слова в этом конкретном файле.

Я знаю, что он немного сложный, но любой будет оценен.

Map.java

import java.io.IOException; 
import java.util.*; 

import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapreduce.*; 
import org.apache.hadoop.mapreduce.lib.input.FileSplit; 

public class Map extends Mapper<LongWritable, Text, Text, IntWritable> { 
    private final static IntWritable one = new IntWritable(1); 
    private Text word = new Text(); 
    private String pattern= "^[a-z][a-z0-9]*$"; 

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
     String line = value.toString(); 
     StringTokenizer tokenizer = new StringTokenizer(line); 
     InputSplit inputSplit = context.getInputSplit(); 
     String fileName = ((FileSplit) inputSplit).getPath().getName(); 
     while (tokenizer.hasMoreTokens()) { 
      word.set(tokenizer.nextToken()); 
      String stringWord = word.toString().toLowerCase(); 
      if (stringWord.matches(pattern)){ 
       context.write(new Text(stringWord), one); 
      } 

     } 
    } 
} 

Reduce.java

import java.io.IOException; 

import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapreduce.*; 

public class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> { 

    public 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)); 
    } 
} 

WordCount.java

import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.conf.*; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapreduce.*; 
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; 

public class WordCount { 
    public static void main(String[] args) throws Exception { 
     Configuration conf = new Configuration(); 

     Job job = new Job(conf, "WordCount"); 
     job.setJarByClass(WordCount.class); 
     job.setOutputKeyClass(Text.class); 
     job.setOutputValueClass(IntWritable.class); 

     job.setNumReduceTasks(3); 

     job.setMapperClass(Map.class); 
     job.setReducerClass(Reduce.class); 

     job.setInputFormatClass(TextInputFormat.class); 
     job.setOutputFormatClass(TextOutputFormat.class); 

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

     job.waitForCompletion(true); 
    } 
} 

Answer 1

В картографа, создать пользовательский записываемый textpair который будет ключом вывода, будет содержать имя файла и слово из вашего файла и значение как 1.

Mapper Выход:

<K,V> ==> <MytextpairWritable,new IntWritable(1) 

Вы можете получить имя файла в картографа с ниже фрагмент кода.

FileSplit fileSplit = (FileSplit)context.getInputSplit(); 
String filename = fileSplit.getPath().getName(); 

И передайте их в качестве конструктора для обычного класса, доступного для записи, в файле context.write. Что-то вроде этого.

context.write(new MytextpairWritable(filename,word),new IntWritable(1)); 

И в стороне редуктора просто суммировать значение, так что вы могли бы получить для каждого файла, сколько вхождений есть для конкретного слова. Код редуктора будет примерно таким.

public class Reduce extends Reducer<mytextpairWritable, IntWritable,mytextpairWritable, IntWritable> { 


    public void reduce(mytextpairWritable 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)); 
} 

Ваш результат будет примерно таким.

File1,hello,2 
File2,hello,3 
File3,hello,1