大数据学习（6）MapReduce应用

 

倒排索引

/**
 * 
 *
 * <pre>
 *file1.txt:
 *hello ketty
 *hello tomcat
 *
 *file2.txt:
 *hello hadoop
 *
 *map1:
 *hello:file1.txt 1
 *hello:file1.txt 1
 *ketty:file1.txt 1
 *tomcat:file1.txt 1
 *hello:file2.txt 1
 *hadoop:file2.txt 1
 *
 *reduce1:
 *hello:file1.txt 2
 *ketty:file1.txt 1
 *tomcat:file1.txt 1
 *hello:file2.txt 1
 *hadoop:file2.txt 1
 *
 *reduce2:
 *hello file1.txt 2,file2.txt 1
 *ketty file1.txt 1
 *tomcat file1.txt 1
 *hadoop file2.txt 1
 *</pre>
 * @author huqiao
 */
public class InvertedIndex {
    
    /**
     * input:files to be inverted index<br/>
     * output: someword:filename  count
     * @author huqiao
     */
    static class WordInFileCountMapper extends Mapper<LongWritable,Text,Text,LongWritable>{

        @Override
        protected void map(LongWritable key, Text value,Context ctx)
                throws IOException, InterruptedException {
            String line = value.toString();
            String[] words = line.split(" ");
            
            FileSplit fileSplit = (FileSplit)ctx.getInputSplit();
            String fileName = fileSplit.getPath().getName();
            for(String word : words) {
                ctx.write(new Text(word + ":" + fileName), new LongWritable(1));
            }
        }
        
    }
    
    /**
     * output:
     * <pre>
     *hello:file1.txt 2
     *ketty:file1.txt 1
     *tomcat:file1.txt 1
     *hello:file2.txt 1
     *hadoop:file2.txt 1
     *</pre>
     * @author huqiao
     */
    static class WordInFileCountReducer extends Reducer<Text,LongWritable,Text,LongWritable>{

        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context ctx) throws IOException, InterruptedException {
            int total = 0;
            for(LongWritable value : values) {
                total += value.get();
            }
            ctx.write(key, new LongWritable(total));
        }
        
    }
    
    
    /**
     * output:
     * <pre>
     * hello-->WordCountRecord{fileName:file1.txt,count:2}
     * ...
     * </pre>
     * @author huqiao
     */
    static class InvertedIndexMapper extends Mapper<LongWritable,Text,Text,WordCountRecord>{

        @Override
        protected void map(LongWritable key, Text value,Context ctx)
                throws IOException, InterruptedException {
            String line = value.toString();
            String[] lineArray = line.split("	");
            String[] wordAndFileName = lineArray[0].split(":");
            String word = wordAndFileName[0];
            String fileName = wordAndFileName[1];
            Long count = Long.parseLong(lineArray[1]);
            
            ctx.write(new Text(word), new WordCountRecord(fileName, count));
            
        }
        
    }
    
    /**
     * output:
     * <pre>
     * hello-->file1.txt 2,file2.txt 1
     * ...
     * </pre>
     * @author huqiao
     */
    static class InvertedIndexReducer extends Reducer<Text,WordCountRecord,Text,Text>{

        @Override
        protected void reduce(Text key, Iterable<WordCountRecord> values, Context ctx) throws IOException, InterruptedException {
             StringBuffer output = new StringBuffer();
             for(WordCountRecord value : values) {
                 output.append(value.getFileName() + " " + value.getCount()+",");
             }
             ctx.write(key, new Text(output.toString()));
        }
        
    }
    
    public static void main(String[] args) throws Exception{
        
        String inputPath = args[0];
        String outputPath = args[1];
        String phase = args[2];
        
        FileSystem fs = FileSystem.get(new URI("hdfs://vcentos1:9000"),new Configuration(),"root");
        
        //delete output path when it existed
        Path output = new Path(outputPath);
        if(fs.exists(output)) {
            fs.delete(output,true);
        }
        
        if("phase1".equals(phase)) {
             doPhase1(inputPath,outputPath);
        }else {
            doPhase2(inputPath,outputPath);
        }
       
        
    }

    private static void doPhase1(String inputPath,String outputPath)throws Exception {
             Job job = Job.getInstance();
            
            job.setJarByClass(InvertedIndex.class);
            
            job.setMapperClass(WordInFileCountMapper.class);
            job.setReducerClass(WordInFileCountReducer.class);
            
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);
            
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);
            
            FileInputFormat.setInputPaths(job, new Path(inputPath));
            FileOutputFormat.setOutputPath(job, new Path(outputPath));
            
            boolean success = job.waitForCompletion(true);
            
            System.exit(success ? 0 : 1);
    }
    
    private static void doPhase2(String inputPath,String outputPath)throws Exception {
        Job job = Job.getInstance();
        
        job.setJarByClass(InvertedIndex.class);
        
        job.setMapperClass(InvertedIndexMapper.class);
        job.setReducerClass(InvertedIndexReducer.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(WordCountRecord.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        FileInputFormat.setInputPaths(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        
        boolean success = job.waitForCompletion(true);
        
        System.exit(success ? 0 : 1);
    }

}

 执行时分两个阶段：

 hadoop jar mr.jar me.huqiao.hadoop.demo_code.invertedsort.InvertedIndex /invertedindex/input /invertedindex/phase-a-output/ phase1

然后以第一个阶段的输出作为第二个阶段的输入：

hadoop jar mr.jar me.huqiao.hadoop.demo_code.invertedsort.InvertedIndex /invertedindex/phase-a-output /invertedindex/phase-b-output/ phase2

最终效果类似于：

about   logs.txt 1,
are     text.txt 1,
hadoop  file1.txt 1,
hdfs    file1.txt 1,
hello   text.txt 1,logs.txt 1,file1.txt 2,
how     logs.txt 1,text.txt 1,
kitty   logs.txt 1,
today   logs.txt 1,
tom     text.txt 1,
you     text.txt 1,

找出价格最贵的商品

共同QQ好友