javamap倒序索引_MapReduce案例之倒排索引

1. 倒排索引

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

2. 实例描述

通常情况下&＃xff0c;倒排索引由一个单词(或词组)以及相关的文档列表组成&＃xff0c;文档列表中的文档或者是标识文档的 ID 号&＃xff0c;或者是指文档所在位置的 URL。如下图所示&＃xff1a;

最常用的是使用词频作为权重&＃xff0c;即记录单词在文档中出现的次数。以英文为例&＃xff0c;如下图所示&＃xff0c;索引文件中的“ MapReduce”一行表示&＃xff1a;“ MapReduce”这个单词在文本 T0 中 出现过 1 次&＃xff0c;T1 中出现过 1 次&＃xff0c;T2 中出现过 2 次。

这里存在两个问题&＃xff1a;第一&＃xff0c; key/value对只能有两个值&＃xff0c;需要根据情况将其中两个值合并成一个值&＃xff0c;作为 key 或 value 值&＃xff1b;

第二&＃xff0c;通过一个 Reduce 过程无法同时完成词频统计和生成文档列表&＃xff0c;所以必须增加一个 Combine 过程完成词频统计。

这里将单词和 URL 组成 key 值(如“ MapReduce&＃xff1a; file1.txt”)&＃xff0c;将词频作为value&＃xff0c;这样做的好处是可以利用 MapReduce 框架自带的Map 端排序&＃xff0c;将同一文档的相同单词的词频组成列表&＃xff0c;传递给 Combine 过程&＃xff0c;实现类似于 WordCount 的功能。

3.2 Combine 过程

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

3.4 程序代码

pom文件

xsi:schemaLocation&＃61;"http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

com.itcast

invertedIndex

1.0-SNAPSHOT

jar

invertedIndex

http://maven.apache.org

UTF-8

org.apache.hadoop

hadoop-common

2.6.4

org.apache.hadoop

hadoop-hdfs

2.6.4

org.apache.hadoop

hadoop-client

2.6.4

org.apache.hadoop

hadoop-mapreduce-client-core

2.6.4

org.apache.maven.plugins

maven-jar-plugin

2.4

true

lib/

cn.itcast.hadoop.mrwc.WordCountDriver

Map程序

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class InvertedIndexMapper extends Mapper{

private static Text keyInfo &＃61; new Text();// 存储单词和 URL 组合

private static final Text valueInfo &＃61; new Text("1");// 存储词频,初始化为1

&＃64;Override

protected void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

String line &＃61; value.toString();

String[] fields &＃61; line.split(" ");// 得到字段数组

FileSplit fileSplit &＃61; (FileSplit) context.getInputSplit();// 得到这行数据所在的文件切片

String fileName &＃61; fileSplit.getPath().getName();// 根据文件切片得到文件名

for (String field : fields) {

// key值由单词和URL组成&＃xff0c;如“MapReduce:file1”

keyInfo.set(field &＃43; ":" &＃43; fileName);

context.write(keyInfo, valueInfo);

}

}

}

combine程序

import java.io.IOException;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class InvertedIndexCombiner extends Reducer{

private static Text info &＃61; new Text();

// 输入&＃xff1a;

// 输出&＃xff1a;

&＃64;Override

protected void reduce(Text key, Iterable values, Context context)

throws IOException, InterruptedException {

int sum &＃61; 0;// 统计词频

for (Text value : values) {

sum &＃43;&＃61; Integer.parseInt(value.toString());

}

int splitIndex &＃61; key.toString().indexOf(":");

// 重新设置 value 值由 URL 和词频组成

info.set(key.toString().substring(splitIndex &＃43; 1) &＃43; ":" &＃43; sum);

// 重新设置 key 值为单词

key.set(key.toString().substring(0, splitIndex));

context.write(key, info);

}

}

reduce程序

import java.io.IOException;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class InvertedIndexReducer extends Reducer{

private static Text result &＃61; new Text();

// 输入&＃xff1a;

// 输出&＃xff1a;

&＃64;Override

protected void reduce(Text key, Iterable values, Context context)

throws IOException, InterruptedException {

// 生成文档列表

String fileList &＃61; new String();

for (Text value : values) {

fileList &＃43;&＃61; value.toString() &＃43; ";";

}

result.set(fileList);

context.write(key, result);

}

}

主程序

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

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;

public class InvertedIndexRunner {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

Configuration conf &＃61; new Configuration();

Job job &＃61; Job.getInstance(conf);

job.setJarByClass(InvertedIndexRunner.class);

job.setMapperClass(InvertedIndexMapper.class);

job.setCombinerClass(InvertedIndexCombiner.class);

job.setReducerClass(InvertedIndexReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(Text.class);

FileInputFormat.setInputPaths(job, new Path("D:\\ziliao\\data\\InvertedIndex\\input"));

// 指定处理完成之后的结果所保存的位置

FileOutputFormat.setOutputPath(job, new Path("D:\\ziliao\\data\\InvertedIndex\\output"));

// 向 yarn 集群提交这个 job

boolean res &＃61; job.waitForCompletion(true);

System.exit(res ? 0 : 1);

}

}

按权重排序

/**

* Created by Administrator on 2018/8/15.

*/

public class FileCount implements Comparable {

private String filename;

private long count;

//按照总流量倒序排

public int compareTo(FileCount bean) {

return bean.count>this.count?1:-1;

}

public FileCount(String filename, long count) {

this.filename &＃61; filename;

this.count &＃61; count;

}

&＃64;Override

public String toString() {

return filename &＃43; ":" &＃43; count;

}

}

新reduce程序

import java.io.IOException;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class InvertedIndexReducer extends Reducer{

private static Text result &＃61; new Text();

// 输入&＃xff1a;

// 输出&＃xff1a;

&＃64;Override

protected void reduce(Text key, Iterable values, Context context)

throws IOException, InterruptedException {

// 生成文档列表

String fileList &＃61; new String();

List FileCountList &＃61; new ArrayList();

for (Text value : values) {

String[] arr &＃61; value.toString().split(":");

FileCount FileCount &＃61; new FileCount(arr[0],Long.parseLong(arr[1]));

FileCountList.add(FileCount);

}

Collections.sort(FileCountList);

for(FileCount FileCount : FileCountList)

{

fileList &＃43;&＃61; FileCount.toString() &＃43; ";";

}

result.set(fileList);

context.write(key, result);

}

}