hadoop mapreduce를 사용하는 파일의 상호 단어

Question

여러 파일에 존재하는 단어를 '유일한'목록으로 표시 할 수있는 코드를 실행하려고했습니다. 지금까지 내가 한 일은 Chris White에게 wordcount 예제와 고맙습니다. 저는 그것을 컴파일 할 수있었습니다. 나는 여기저기서 코드를 작동 시키려고 노력했지만, 나는 데이터가없는 빈 페이지 만 얻는다. 매퍼는 각 단어를 해당 위치와 함께 수집한다고 가정합니다. 감속기는 무엇이 문제인지에 관한 어떤 생각이라도 공통 단어를 수집한다고 가정합니다. 코드는 다음과 같습니다.

package org.myorg; 

import java.io.IOException; 
import java.util.*; 
import java.lang.*; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.conf.*; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapred.*; 
import org.apache.hadoop.util.*; 

public class WordCount { 



    public static class Map extends MapReduceBase implements Mapper<Text, Text, Text, Text> 
    { 

     private final static IntWritable one = new IntWritable(1); 
     private Text word = new Text(); 

      private Text outvalue=new Text(); 
      private String filename = null; 

     public void map(Text key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException 
     { 
     if (filename == null) 
     { 
      filename = ((FileSplit) reporter.getInputSplit()).getPath().getName(); 
     } 

     String line = value.toString(); 
     StringTokenizer tokenizer = new StringTokenizer(line); 

     while (tokenizer.hasMoreTokens()) 
     { 
      word.set(tokenizer.nextToken()); 
      outvalue.set(filename); 
      output.collect(word, outvalue); 
     } 

     } 
    } 



    public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> 
    { 


     private Text src = new Text(); 
     public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException 
     { 


     int sum = 0; 
     //List<Text> list = new ArrayList<Text>(); 

      while (values.hasNext()) // I believe this would have all locations of the same word in different files? 
      { 

       sum += values.next().get(); 
       src =values.next().get(); 

      } 
     output.collect(key, src); 
      //while(values.hasNext()) 
      //{ 
       //Text value = values.next(); 
       //list.add(new Text(value)); 
       //System.out.println(value.toString());  
      //} 
      //System.out.println(values.toString()); 
      //for(Text value : list) 
      //{ 
       //System.out.println(value.toString()); 
      //} 


     } 

    } 



    public static void main(String[] args) throws Exception 
    { 

    JobConf conf = new JobConf(WordCount.class); 
    conf.setJobName("wordcount"); 
    conf.setInputFormat(KeyValueTextInputFormat.class); 
    conf.setOutputKeyClass(Text.class); 
    conf.setOutputValueClass(Text.class); 
    conf.setMapperClass(Map.class); 
    conf.setCombinerClass(Reduce.class); 
    conf.setReducerClass(Reduce.class); 
    //conf.setInputFormat(TextInputFormat.class); 
    conf.setOutputFormat(TextOutputFormat.class); 
    FileInputFormat.setInputPaths(conf, new Path(args[0])); 
    FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
    JobClient.runJob(conf); 

    } 

} 

나는 무엇이 누락 되었습니까? 많은 의무 ... 내 하둡 버전 : 당신이 이전 하둡 API (mapred)를 사용하고, 조언의 단어가 새로운 하둡 API를 사용하는 것 같다 모든 0.20.203

Answer 1

먼저 (맵리 듀스

mkdir classes 
javac -classpath /path/to/hadoop-0.20.203/hadoop-0.20.203-core.jar:/path/to/hadoop- 0.20.203/lib/commons-cli-1.2.jar -d classes WordCount.java && jar -cvf wordcount.jar -C classes/ . 

:) 새로운 API에서 0.20.203

와 호환되는, 여기 그런 다음, 우리가이 파일을 구축

import java.io.IOException; 
import java.lang.InterruptedException; 
import java.util.StringTokenizer; 

import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
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 WordCount { 
/** 
* The map class of WordCount. 
*/ 
public static class TokenCounterMapper 
    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); 
     } 
    } 
} 
/** 
* The reducer class of WordCount 
*/ 
public static class TokenCounterReducer 
    extends Reducer<Text, IntWritable, Text, IntWritable> { 
    public void reduce(Text key, Iterable<IntWritable> values, Context context) 
     throws IOException, InterruptedException { 
     int sum = 0; 
     for (IntWritable value : values) { 
      sum += value.get(); 
     } 
     context.write(key, new IntWritable(sum)); 
    } 
} 
/** 
* The main entry point. 
*/ 
public static void main(String[] args) throws Exception { 
    Configuration conf = new Configuration(); 
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 
    Job job = new Job(conf, "Example Hadoop 0.20.1 WordCount"); 
    job.setJarByClass(WordCount.class); 
    job.setMapperClass(TokenCounterMapper.class); 
    job.setReducerClass(TokenCounterReducer.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); 
} 
} 

일을하고 jar 파일에 결과를 팩하는 단어 수있다

마지막으로, 우리는이 모든 값을 섭취 한 후 경우, 관찰 된 값의 집합 (매퍼에서 방출되는 파일 이름)을 유지, 감속기에서 하둡

echo "hello world bye world" > /tmp/in/0.txt 
echo "hello hadoop goodebye hadoop" > /tmp/in/1.txt 
hadoop jar wordcount.jar org.packagename.WordCount /tmp/in /tmp/out 

Answer 2

의 독립 실행 형 모드에서 jar 파일을 실행 set size가 1이면 단어는 하나의 파일에서만 사용됩니다.

public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> 
{ 
    private TreeSet<Text> files = new TreeSet<Text>(); 

    public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException 
    { 
     files.clear(); 

     for (Text file : values) 
     { 
      if (!files.contains(value)) 
      { 
       // make a copy of value as hadoop re-uses the object 
       files.add(new Text(value)); 
      } 
     } 

     if (files.size() == 1) { 
      output.collect(key, files.first()); 
     } 

     files.clear(); 
    } 
}