在上一篇博客中我們講解了MapReduce的原理以及map和reduce的作用，相信你理解了他們的原理，今天講解的是mapreduce 的另一個就是倒排索引。

什麽是倒排索引呢？倒排索引源於實際應用中需要根據屬性的值來查找記錄。這種索引表中的每一項都包括一個屬性值和具有該屬性值的各記錄的地址。由於不是由記錄來確定屬性值，而是由屬性值來確定記錄的位置，因而稱為倒排索引(inverted index)。帶有倒排索引的文件我們稱為倒排索引文件，簡稱倒排文件。

　　我們通過一個例子來講解一下:

　　有3個文本文件，上面的內容分別為：

data01.TXT ---------I love Beijing and i love Shanghai

　　data02.TXT----------I love China

　　data03.TXT----------Beijing is the capital of China

　　現在我們需要統計每個文檔中每個單詞出現的次數，並且加上他的索引值也就是在哪個文檔，想要實現的結果就是下面的形式：

　　如果只是使用wordcount那種方式的話，我們來屢一下思路，按照最後的輸出格式，我們肯定要將單詞作為key，單詞的索引（文件名）和他在文件中出現的次數作為value值，那麽在map中我們的輸出格式就必須為key (單詞）value（文件名+次數），但是這樣的話，我們就無法統計他在每個文件中出現的次數的總和，他在map中的輸出格式只能是下面這種形式。

　　而以這種形式來在reduce中進行計算時，你只能將所有文本中相同的單詞數量統計，而無法統計一個文本中單詞出現的數量。最後輸出的結果只能為下面這種形式。

我們發現love在data01.TXT中出現了兩次，可是他卻分開了，這就是wordcount無法實現的功能。這是我們就需要使用倒排索引的方式，將索引值（文件名）作為key值進行提取，這裏我們需要學一個新內容叫做，Combine,那麽Combine的作用和原理是什麽呢？Combine就是一個相當於本地結合的東西，就是在map和reduce之間，在map將內容提交給reduce時，進行本地的Key結合，將本地文件這裏就是同一個文本的相同的key匯成一個塊，再將這個塊傳給reduce。下面我們來看看具體的流程：

上面的就是數據的走向，下面來看看代碼：

Map:

package com.zll.wordcount;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class WordcountMap extends Mapper<LongWritable,Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String s = value.toString();
        String[] split = s.split(" ");
        InputSplit inputSplit = context.getInputSplit();
        String fileName = ((FileSplit) inputSplit).getPath().toString();
        String lujin = fileName.substring(fileName.lastIndexOf("/")+1);

        for (String str:split) {
            context.write(new Text(str+":"+lujin),new Text("1"));
        }
    }
}

Combine:

package com.zll.wordcount;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class WordcountCombine extends Reducer<Text,Text,Text,Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        int count=0;
        for (Text s1:values) {
            String str=s1.toString();
           count=count+Integer.parseInt(s1.toString());
        }
        String text=key.toString();
        String[] split = text.split(":");
       

        context.write(new Text(split[0]),new Text(split[1]+":"+count));
    }
}

Reduce:

package com.zll.wordcount;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


public class WordcountReduce extends Reducer<Text,Text,Text,Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        String zhi="";
        for (Text s1:values) {
            zhi=zhi+"("+s1.toString()+")";
        }
        context.write(key,new Text(zhi));
    }
}

Job:

package com.zll.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


public class WordcountJob extends Configured implements Tool {

    public static void main(String[] args) {
        WordcountJob wordcountJob=new WordcountJob();
        try {
            ToolRunner.run(wordcountJob, null);
        } catch (Exception e) {
            // TODO 自動生成的 catch 塊
            e.printStackTrace();
        }
    }
    @Override
    public int run(String[] args) throws Exception {
        Configuration configuration=new Configuration();
        configuration.set("fs.defaultFS","hdfs://192.168.153.11:9000");
        Job job=Job.getInstance(configuration);
        job.setJarByClass(WordcountJob.class);
        job.setMapperClass(WordcountMap.class);
        job.setReducerClass(WordcountReduce.class);
        job.setCombinerClass(WordcountCombine.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job,new Path("/hadoop/data"));
        FileOutputFormat.setOutputPath(job,new Path("/hadoop/outdata"));
        job.waitForCompletion(true);

        return 0;
    }
}

大數據MapReduce入門之倒排索引