zoukankan      html  css  js  c++  java
  • 大数据学习(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好友

  • 相关阅读:
    jQuery为链接添加链接图像
    jQuery插件Toggle text value
    用ajax清除浏览器缓存的js、css、图片等
    Ajax 解决ie缓存问题
    jQuery插件slidergallery.(仿MAC展示.)
    javascript计算器Calculator
    回车自动提交内容
    个性化表单五技巧
    教你怎么样用CSS和图片做搜索框.
    浅谈css框架开发
  • 原文地址:https://www.cnblogs.com/at0x7c00/p/8083167.html
Copyright © 2011-2022 走看看