Mapreduce实例——倒排索引

现有某电商网站的3张信息数据表，分别为商品库表goods3，商品访问情况表goods_visit3，订单明细表order_items3，goods表记录了商品的状态数据，goods_visit3记录了商品的点击情况，order_items3记录了用户购买的商品的信息数据，它们的表结构及内容如下：

商品ID,商品点击次数
1024600,2
1024593,0
1024592,0
1024590,0
1024589,0
1024588,0
1024587,0
1024586,0
1024585,0
1024584,0

明细ID,订单ID,商品ID,购买数据,商品销售价格,商品最终单价,商品金额
251688,52107,1024600,1,31.6,31.6,15.8
252165,52209,1024600,1,31.6,31.6,15.8
251870,52146,1024481,1,15.6,15.6,7.8
251935,52158,1024481,1,15.6,15.6,7.8
252415,52264,1024480,1,69.0,69.0,69.0
250983,51937,1024480,1,69.0,69.0,69.0
252609,52299,1024480,1,69.0,69.0,69.0
251689,52107,1024440,1,31.6,31.6,15.8
239369,49183,1024256,1,759.0,759.0,759.0
249222,51513,1024140,1,198.0,198.0,198.0

商品ID,商品状态,分类ID,评分
1024600,6,52006,0
1024593,1,52121,0
1024592,1,52121,0
1024590,1,52119,0
1024589,1,52119,0
1024588,1,52030,0
1024587,1,52021,0
1024586,1,52029,0
1024585,1,52014,0
1024584,1,52029,0

通过mapreduce统计goods_id相同的商品都在哪几张表并统计出现了多少次：

package mapreduce9;

import java.io.IOException;
import org.apache.hadoop.fs.Path;
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;

//08.Mapreduce实例——倒排索引
public class MyIndex {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance();
        job.setJobName("InversedIndexTest");
        job.setJarByClass(MyIndex.class);

        job.setMapperClass(doMapper.class);
        job.setCombinerClass(doCombiner.class);
        job.setReducerClass(doReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        Path in1 = new Path("hdfs://192.168.51.100:8020/mymapreduce9/in/goods3");
        Path in2 = new Path("hdfs://192.168.51.100:8020/mymapreduce9/in/goods_visit3");
        Path in3 = new Path("hdfs://192.168.51.100:8020/mymapreduce9/in/order_items3");
        Path out = new Path("hdfs://192.168.51.100:8020/mymapreduce9/out");

        FileInputFormat.addInputPath(job, in1);
        FileInputFormat.addInputPath(job, in2);
        FileInputFormat.addInputPath(job, in3);
        FileOutputFormat.setOutputPath(job, out);

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class doMapper extends Mapper{
        public static Text myKey = new Text();
        public static Text myValue = new Text();
        //private FileSplit filePath;

        @Override
        protected void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            String filePath=((FileSplit)context.getInputSplit()).getPath().toString();
            if(filePath.contains("goods")){
                String val[]=value.toString().split(",");
                int splitIndex =filePath.indexOf("goods");
                myKey.set(val[0] + ":" + filePath.substring(splitIndex));
            }else if(filePath.contains("order")){
                String val[]=value.toString().split(",");
                int splitIndex =filePath.indexOf("order");
                myKey.set(val[2] + ":" + filePath.substring(splitIndex));
            }
            myValue.set("1");
            context.write(myKey, myValue);
        }
    }
    public static class doCombiner extends Reducer{
        public static Text myK = new Text();
        public static Text myV = new Text();

        @Override
        protected void reduce(Text key, Iterable values, Context context)
                throws IOException, InterruptedException {
            int sum = 0 ;
            for (Text value : values) {
                sum += Integer.parseInt(value.toString());
            }
            int mysplit = key.toString().indexOf(":");
            myK.set(key.toString().substring(0, mysplit));
            myV.set(key.toString().substring(mysplit + 1) + ":" + sum);
            context.write(myK, myV);
        }
    }

    public static class doReducer extends Reducer{

        public static Text myK = new Text();
        public static Text myV = new Text();

        @Override
        protected void reduce(Text key, Iterable values, Context context)
                throws IOException, InterruptedException {

            String myList = new String();

            for (Text value : values) {
                myList += value.toString() + ";";
            }
            myK.set(key);
            myV.set(myList);
            context.write(myK, myV);
        }
    }
}

结果：

原理：

"倒排索引"是文档检索系统中最常用的数据结构，被广泛地应用于全文搜索引擎。它主要是用来存储某个单词（或词组）在一个文档或一组文档中的存储位置的映射，即提供了一种根据内容来查找文档的方式。由于不是根据文档来确定文档所包含的内容，而是进行相反的操作，因而称为倒排索引（Inverted Index）。

实现"倒排索引"主要关注的信息为：单词、文档URL及词频。

（1）Map过程

首先使用默认的TextInputFormat类对输入文件进行处理，得到文本中每行的偏移量及其内容。显然，Map过程首先必须分析输入的对，得到倒排索引中需要的三个信息：单词、文档URL和词频，接着我们对读入的数据利用Map操作进行预处理，如下图所示：

这里存在两个问题：第一，对只能有两个值，在不使用Hadoop自定义数据类型的情况下，需要根据情况将其中两个值合并成一个值，作为key或value值。第二，通过一个Reduce过程无法同时完成词频统计和生成文档列表，所以必须增加一个Combine过程完成词频统计。

这里将商品ID和URL组成key值（如"1024600：goods3"），将词频（商品ID出现次数）作为value，这样做的好处是可以利用MapReduce框架自带的Map端排序，将同一文档的相同单词的词频组成列表，传递给Combine过程，实现类似于WordCount的功能。

（2）Combine过程

经过map方法处理后，Combine过程将key值相同的value值累加，得到一个单词在文档中的词频，如下图所示。如果直接将下图所示的输出作为Reduce过程的输入，在Shuffle过程时将面临一个问题：所有具有相同单词的记录（由单词、URL和词频组成）应该交由同一个Reducer处理，但当前的key值无法保证这一点，所以必须修改key值和value值。这次将单词（商品ID）作为key值，URL和词频组成value值（如"goods3：1"）。这样做的好处是可以利用MapReduce框架默认的HashPartitioner类完成Shuffle过程，将相同单词的所有记录发送给同一个Reducer进行处理。

（3）Reduce过程

经过上述两个过程后，Reduce过程只需将相同key值的所有value值组合成倒排索引文件所需的格式即可，剩下的事情就可以直接交给MapReduce框架进行处理了。如下图所示