MapReduce编程——倒排索引

需求：统计每个单词在每个文件中出现的词频

有2个文件：a.txt b.txt
a.txt 文件中的内容：
<0,”hello tom”>
<1,”hello kitty”>
<2,”hello world”>
<3,”hello jerry”>

b.txt 文件中的内容：
<0,”hello tom”>
<1,”hello tom”>
<2,”hello jerry “>
想要得到的最终结果形式：
hello “a.txt->5 b.txt->3”
tom “a.txt->2 b.txt->1”
kitty “a.txt->1”

原理剖析：总共分３个阶段
——————— 第一阶段：Map阶段————————————-
输入：
a.txt 文件中的内容：
<0,”hello tom”>
<1,”hello kitty”>
<2,”hello world”>
<3,”hello jerry”>

b.txt 文件中的内容：
<0,”hello tom”>
<1,”hello tom”>
<2,”hello jerry “>

输出：
context.write(“hello->a.txt”,”1”);
context.write(“tom->a.txt”,”1”);
context.write(“hello->a.txt”,”1”);
context.write(“kitty->a.txt”,”1”);
context.write(“hello->a.txt”,”1”);
context.write(“world->a.txt”,”1”);
context.write(“hello->a.txt”,”1”);
context.write(“jerry->a.txt”,”1”);

context.write(“hello->b.txt”,”1”);
context.write(“tom->b.txt”,”1”);
context.write(“hello->b.txt”,”1”);
context.write(“tom->a.txt”,”1”);
context.write(“hello->a.txt”,”1”);
context.write(“jerry->a.txt”,”1”);

——————— Combiner阶段———————————–
输入：
<”hello->a.txt”,”1”>
<”hello->a.txt”,”1”>
<”hello->a.txt”,”1”>
<”hello->a.txt”,”1”>
<”hello->a.txt”,”1”>

<”hello->b.txt”,”1”>
<”hello->b.txt”,”1”>
<”hello->b.txt”,”1”>

输出：
context.write(“hello”,”a.txt->4”);
context.write(“hello”,”b.txt->3”);

————————— Reducer阶段—————————–
输入：

<”hello”,{“a.txt->5”,”b.txt->3”}>

输出：
context.write(“hello”,”a.txt->5 b.txt->3”);

---

最终结果：
hello 　“a.txt->5　b.txt->3”
tom　　“a.txt->2　b.txt->1”
kitty 　“a.txt->1”

实现代码如下：

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
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;

public class InverseIndex {

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

        Job job = Job.getInstance(conf);
        //设置jar
        job.setJarByClass(InverseIndex.class);

        //设置Mapper相关的属性
        job.setMapperClass(IndexMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));//words.txt

        //设置Reducer相关属性
        job.setReducerClass(IndexReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setCombinerClass(IndexCombiner.class);

        //提交任务
        job.waitForCompletion(true);
    }
    /**
      * Map阶段
     **/
    public static class IndexMapper extends Mapper<LongWritable, Text, Text, Text>{

        private Text k = new Text();
        private Text v = new Text();
        @Override//重写map方法
        protected void map(LongWritable key, Text value,
                Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            //将文件中的内容一行一行读出 输入内容的原型应该是<0,"hello tom"> 行号作为key，一行内容作为value
            String line = value.toString();
            //使用空格分割字符串
            String[] fields = line.split(" ");
            //得到该map的输入切片
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            //通过输入切片获取文件路径名称
            Path path = inputSplit.getPath();
            String name = path.getName();
            //拼接字符串  文件名->出现的次数
            for(String f : fields){
                k.set(f + "->" + name);
                v.set("1");
                context.write(k, v);
            }
        }

    }
    /**
      * combiner是一个特殊的reducer，先在combiner中进行一次合并
     **/
    public static class IndexCombiner extends Reducer<Text, Text, Text, Text>{

        private Text k = new Text();
        private Text v = new Text();
        @Override//重写reduce方法（不是conbiner）
        protected void reduce(Text key, Iterable<Text> values,
                Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            //将 文件名称 和 单词出现次数 通过"->" 切分
            String[] fields = key.toString().split("->");
            long sum = 0;
            //迭代器values中的内容应该是：【1,1,1...】
            for(Text t : values){
                sum += Long.parseLong(t.toString());
            }
            //输出Key应该是 单词
            k.set(fields[0]);
            //取出这个单词对应的文件名称，和出现次数进行拼接 作为输出Value的值
            v.set(fields[1] + "->" + sum);
            context.write(k, v);
        }

    }
    /**
      * Reduce阶段 在reduce阶段进行第二次合并
      */
    public static class IndexReducer extends Reducer<Text, Text, Text, Text>{

        private Text v = new Text();
        @Override//输入 K：单词   V：文件名—>出现次数
        protected void reduce(Text key, Iterable<Text> values,
                Reducer<Text, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            String value = "";
            //迭代器values中的内容应该是：【a.txt->5 ，b.txt->3】
            将这个值通过迭代器循环，使用"->"连接起来
            for(Text t : values){
                value += t.toString() + " ";
            }
            //key仍然是单词
            v.set(value);
            context.write(key, v);
        }

    }

}

运行：

编写好Map-Reduce程序之后，在hadoop的hdfs上传两个文件，存放至/ii目录下，即 ii/[ a.txt, b.txt ]
使用hadoop jar 命令启动运行MR例程：

hadoop　jar　　/root/mrs.jar（指定jar文件保存的目录）cn.itcast.hadoop.mr.ii.InverseIndex(指定jar文件中main方法的路径)　/ii(输入文件a.txt b.txt在hdfs中所在的目录)　/iiout(输出文件在hdfs中所在的目录)