MR基础案例（二）倒排索引

向右看齐 2024-04-18 12:52 55阅读 0赞

**文本文档**  
三个文件：index.html,hadoop.html,spark.html  
每个文件里有一些单词

index.html : hadoop hadoop hadoop hadoop index bigdata
    hadoop.html : hadoop hadoop is nice nice best 
    spark.html : spark is best best best

**结果集**  
排序单词，在单词后罗列出现次数最多的网页，也要排序

best : spark.html:3;hadoop.html:1；
    hadoop : index.html:4;hadoop.html:2；
    ...

**实现思路：**

1.  先将数据分片，获取文件名做为key，value设为1计数
2.  将每行按空格拆分，拆出单词与文件名拼接成“index.html\_hadoop”形式做为键值
3.  自定义combine类，拆分键值，形成“index.html:3”形式作为value输出，hadoop单词作为key
4.  在reduce阶段，定义treemap，实现comparator接口，实现value按value.value排序，最后将结果拼接成"hadoop:index:3;"形式；

SortDev.java

package MR;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.InputSplit;
    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.input.FileSplit;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    import java.util.*;
    
    
    public class SortRev {
        
    
        //自定义的mapper类
        public static class MyMapper extends Mapper<Object, Text, Text, Text> {
        
            public Text k = new Text();
            public Text v = new Text("1");
            /**
             * 抽象map函数   (map阶段的核心业务逻辑)
             */
            @Override
            protected void map(Object key, Text value,Context context) throws IOException, InterruptedException {
        
                //获取文件名称
                InputSplit is = context.getInputSplit();
                String filename = ((FileSplit)is).getPath().getName();
                //读取行
                String []  fields = value.toString().split(" ");
                for (int i=0;i<fields.length;i++){
        
                    //输出
                    k.set(filename+"_"+fields[i]);
                    context.write(k,v);
                }
                /**
                 * index.html_hadoop 1
                 * index.html_hadoop 1
                 * index.html_hadoop 1
                 * index.html_best 1
                 *
                 * hadoop.html_hadoop 1
                 * ====================
                 * hadoop index.html:3
                 * hadoop hadoop.html:2
                 * ====================
                 * hadoop : index.html:3;hadoop.html:2
                 *
                 *
                 *
                 */
    //            context.write(value,NullWritable.get());
            }
        }
    
    
        //自定义reducer类
        public static class MyReducer extends Reducer<Text, Text, Text, Text> {
        
            public Text v = new Text();
            // 升序比较器
            Comparator<Map.Entry<String, Integer>> valueComparator = (o1, o2) -> {
        
                //return o1.getValue() - o2.getValue();
                int tmp = 0;
                tmp = o2.getValue() - o1.getValue();
                if(tmp != 0){
        
                    return tmp;
                }
                return o1.getKey().compareTo(o2.getKey());
            };
            Map<String, Integer> map = new TreeMap<>();
            /**
             * 抽象的reduce函数 (reduce阶段的核心业务逻辑)
             */
            @Override
            protected void reduce(Text key, Iterable<Text> value,Context context)
                    throws IOException, InterruptedException {
        
                /**
                 * hadoop list(index.html:3;hadoop.html:2)
                 */
                try {
        
                    //定义计数器
                    for (Text i : value) {
        
                        String[] kvs = i.toString().split(":");
                        map.put(kvs[0],Integer.parseInt(kvs[1]));
                    }
    
                    // map转换成list进行排序
                    List<Map.Entry<String, Integer>> list = new ArrayList<>(map.entrySet());
                    // 排序
                    list.sort(valueComparator);
    
                    //输出
                    String res = "";
                    for (Map.Entry<String, Integer> en : list) {
        
                        res += en.getKey()+":"+en.getValue()+";";
                    }
                    v.set(res.substring(0,res.length()-1));
                    context.write(key, v);
    
                } catch (NumberFormatException | InterruptedException | IOException e) {
        
                    e.printStackTrace();
                } finally {
        
                    map.clear(); //清空
                }
            }
        }
    
        /**
         * 驱动方法
         * @param args
         */
        public static void main(String[] args) {
        
            try {
        
                //1、获取配置对象并进行属性设置
                Configuration conf = new Configuration();
                //对conf设置
                //2、获取job
                Job job = Job.getInstance(conf, "myworldCount");
                //3、对job设置运行主类
                job.setJarByClass(SortRev.class);
    
                //4、对job的map端属性设置
                job.setMapperClass(MyMapper.class);
                job.setMapOutputKeyClass(Text.class);
                job.setMapOutputValueClass(Text.class);
    
                //5、对job的reduce端的属性设置
                job.setReducerClass(MyReducer.class);
                job.setOutputKeyClass(Text.class);
                job.setOutputValueClass(Text.class);
    
                //job设置combiner类
                //job.setCombinerClass(MyReducer.class);
                job.setCombinerClass(SortRevCombiner.class);
    
                //6、设置job的输入路径和输出路径
                FileInputFormat.addInputPath(job, new Path(args[0]));
                FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
                //7、提交作业
                int success = job.waitForCompletion(true) ? 0 : 1;
                //8、退出
                System.exit(success);
            } catch (IOException | ClassNotFoundException | InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

SortDevCombiner.java

package MR;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /**
     * 自定义commbiner
     */
    public class SortRevCombiner extends Reducer<Text, Text, Text, Text> {
        
        public Text k = new Text();
        public Text v = new Text();
        /**
         * 抽象的reduce函数 (reduce阶段的核心业务逻辑)
         */
        @Override
        protected void reduce(Text key, Iterable<Text> value,Context context)
                throws IOException, InterruptedException {
        
            //定义计数器
            String[] ks = key.toString().split("_");
            k.set(ks[1]);
            int counter = 0;
            for (Text l : value){
        
                counter += Integer.parseInt(l.toString());
            }
            v.set(ks[0]+":"+counter);
            context.write(k, v);
        }
    }