MapReduce功能实现十---倒排索引(Inverted Index)
生活随笔
收集整理的這篇文章主要介紹了
MapReduce功能实现十---倒排索引(Inverted Index)
小編覺得挺不錯的,現在分享給大家,幫大家做個參考.
MapReduce功能實現系列：
??????MapReduce功能實現一—Hbase和Hdfs之間數據相互轉換
??????MapReduce功能實現二—排序
??????MapReduce功能實現三—Top N
??????MapReduce功能實現四—小綜合(從hbase中讀取數據統計并在hdfs中降序輸出Top 3)
??????MapReduce功能實現五—去重(Distinct)、計數(Count)
??????MapReduce功能實現六—最大值(Max)、求和(Sum)、平均值(Avg)
??????MapReduce功能實現七—小綜合(多個job串行處理計算平均值)
??????MapReduce功能實現八—分區(Partition)
??????MapReduce功能實現九—Pv、Uv
??????MapReduce功能實現十—倒排索引(Inverted Index)
??????MapReduce功能實現十一—join
?
1.前言：
??"倒排索引"是文檔檢索系統中最常用的數據結構，被廣泛地應用于全文搜索引擎。它主要是用來存儲某個單詞（或詞組）在一個文檔或一組文檔中的存儲位置的映射，即提供了一種根據內容來查找文檔的方式。由于不是根據文檔來確定文檔所包含的內容，而是進行相反的操作，因而稱為倒排索引（Inverted Index）。
?
2.模擬數據：
[hadoop@h71 q1]$ vi file1.txt mapreduce is simple[hadoop@h71 q1]$ vi file2.txt mapreduce is powerful is simple[hadoop@h71 q1]$ vi file3.txt hello mapreduce bye mapreduce說明：
- 這里存在兩個問題：第一，<key,value>對只能有兩個值，在不使用Hadoop自定義數據類型的情況下，需要根據情況將其中兩個值合并成一個值，作為key或value值；第二，通過一個Reduce過程無法同時完成詞頻統計和生成文檔列表，所以必須增加一個Combine過程完成詞頻統計。
- 這里講單詞和URL組成key值（如"MapReduce：file1.txt"），將詞頻作為value，這樣做的好處是可以利用MapReduce框架自帶的Map端排序，將同一文檔的相同單詞的詞頻組成列表，傳遞給Combine過程，實現類似于WordCount的功能。
- Combine過程：經過map方法處理后，Combine過程將key值相同的value值累加，得到一個單詞在文檔在文檔中的詞頻，如果直接輸出作為Reduce過程的輸入，在Shuffle過程時將面臨一個問題：所有具有相同單詞的記錄（由單詞、URL和詞頻組成）應該交由同一個Reducer處理，但當前的key值無法保證這一點，所以必須修改key值和value值。這次將單詞作為key值，URL和詞頻組成value值（如"file1.txt：1"）。這樣做的好處是可以利用MapReduce框架默認的HashPartitioner類完成Shuffle過程，將相同單詞的所有記錄發送給同一個Reducer進行處理。
3.將數據上傳到hdfs上：
[hadoop@h71 q1]$ hadoop fs -mkdir /user/hadoop/index_in[hadoop@h71 q1]$ hadoop fs -put file1.txt /user/hadoop/index_in [hadoop@h71 q1]$ hadoop fs -put file2.txt /user/hadoop/index_in [hadoop@h71 q1]$ hadoop fs -put file3.txt /user/hadoop/index_in4.Java代碼：
[hadoop@h71 q1]$ vi InvertedIndex.java
import java.io.IOException; import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration; 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; import org.apache.hadoop.util.GenericOptionsParser;public class InvertedIndex {public static class Map extends Mapper<Object, Text, Text, Text> {private Text keyInfo = new Text(); // 存儲單詞和URL組合private Text valueInfo = new Text(); // 存儲詞頻private FileSplit split; // 存儲Split對象// 實現map函數public void map(Object key, Text value, Context context) throws IOException, InterruptedException {// 獲得<key,value>對所屬的FileSplit對象split = (FileSplit) context.getInputSplit();StringTokenizer itr = new StringTokenizer(value.toString());while (itr.hasMoreTokens()) {// key值由單詞和URL組成，如"MapReduce：file1.txt"// 獲取文件的完整路徑// keyInfo.set(itr.nextToken()+":"+split.getPath().toString());// 這里為了好看，只獲取文件的名稱。int splitIndex = split.getPath().toString().indexOf("file");keyInfo.set(itr.nextToken() + ":" + split.getPath().toString().substring(splitIndex));// 詞頻初始化為1valueInfo.set("1"); context.write(keyInfo, valueInfo);}}}public static class Combine extends Reducer<Text, Text, Text, Text> {private Text info = new Text();// 實現reduce函數public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {// 統計詞頻int sum = 0;for (Text value : values) {sum += Integer.parseInt(value.toString());}int splitIndex = key.toString().indexOf(":");// 重新設置value值由URL和詞頻組成info.set(key.toString().substring(splitIndex + 1) + ":" + sum);// 重新設置key值為單詞key.set(key.toString().substring(0, splitIndex));context.write(key, info);}}public static class Reduce extends Reducer<Text, Text, Text, Text> {private Text result = new Text();// 實現reduce函數public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {// 生成文檔列表String fileList = new String();for (Text value : values) {fileList += value.toString() + ";";} result.set(fileList);context.write(key, result);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.set("mapred.jar", "ii.jar");String[] ioArgs = new String[] { "index_in", "index_out" };String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs();if (otherArgs.length != 2) {System.err.println("Usage: Inverted Index <in> <out>");System.exit(2);}Job job = new Job(conf, "Inverted Index");job.setJarByClass(InvertedIndex.class);// 設置Map、Combine和Reduce處理類job.setMapperClass(Map.class);job.setCombinerClass(Combine.class);job.setReducerClass(Reduce.class);// 設置Map輸出類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);// 設置Reduce輸出類型job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);// 設置輸入和輸出目錄FileInputFormat.addInputPath(job, new Path(otherArgs[0]));FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);} }5.知識點延伸：
- int indexOf(String str) ：返回第一次出現的指定子字符串在此字符串中的索引。
- int indexOf(String str, int startIndex)：從指定的索引處開始，返回第一次出現的指定子字符串在此字符串中的索引。
- int lastIndexOf(String str) ：返回在此字符串中最右邊出現的指定子字符串的索引。
- int lastIndexOf(String str, int startIndex) ：從指定的索引處開始向后搜索，返回在此字符串中最后一次出現的指定子字符串的索引。
- indexOf(“a”)是從字符串的0個位置開始查找的。比如你的字符串：“abca”，那么程序將會輸出0，之后的a是不判斷的。
- str＝str.substring(int beginIndex);截取掉str從首字母起長度為beginIndex的字符串，將剩余字符串賦值給str。
- str＝str.substring(int beginIndex，int endIndex);截取str中從beginIndex開始至endIndex結束時的字符串，并將其賦值給str。
6.執行：
[hadoop@h71 q1]$ /usr/jdk1.7.0_25/bin/javac InvertedIndex.java [hadoop@h71 q1]$ /usr/jdk1.7.0_25/bin/jar cvf xx.jar InvertedIndex*class [hadoop@h71 q1]$ hadoop jar xx.jar InvertedIndex7.查看結果：
[hadoop@h71 q1]$ hadoop fs -cat /user/hadoop/index_out/part-r-00000 bye file3.txt:1; hello file3.txt:1; is file1.txt:1;file2.txt:2; mapreduce file2.txt:1;file3.txt:2;file1.txt:1; powerful file2.txt:1; simple file2.txt:1;file1.txt:1;8.號外：
有一網友給我發私信說遇到了個問題，讓我幫一下他。
題目：將數據源文件上傳到HDFS系統進行存儲，然后基于MapReduce編程進行數據分析，測試數據來源于美國專利文獻數據。
數據源：
解答代碼：
import java.io.IOException;import org.apache.hadoop.conf.Configuration; 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.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser;public class InvertedIndex { public static class Map extends Mapper<Object, Text, Text, Text> { private final static Text one = new Text("1"); private Text word = new Text();public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line[] = value.toString().split(","); word.set(line[4].substring(1, line[4].length()-1)+":"+line[1]); context.write(word, one);}} public static class Combine extends Reducer<Text, Text, Text, Text> { private Text info = new Text(); // 實現reduce函數 public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { // 統計詞頻 int sum = 0; for (Text value : values) { sum += Integer.parseInt(value.toString()); } int splitIndex = key.toString().indexOf(":"); // 重新設置value值由URL和詞頻組成 info.set("<"+key.toString().substring(splitIndex + 1) + ":" + sum+">"); // 重新設置key值為單詞 key.set(key.toString().substring(0, splitIndex)); context.write(key, info); } } public static class Reduce extends Reducer<Text, Text, Text, Text> { private Text result = new Text(); // 實現reduce函數 public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { // 生成文檔列表 String fileList = new String(); for (Text value : values) {fileList += value.toString() + ",";} result.set(fileList.substring(0,fileList.length()-1)); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); conf.set("mapred.jar", "ii.jar"); String[] ioArgs = new String[] { "index_in", "index_out" }; String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: Inverted Index <in> <out>"); System.exit(2); } Job job = new Job(conf, "Inverted Index"); job.setJarByClass(InvertedIndex.class); // 設置Map、Combine和Reduce處理類 job.setMapperClass(Map.class); job.setCombinerClass(Combine.class); job.setReducerClass(Reduce.class); // 設置Map輸出類型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); // 設置Reduce輸出類型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); // 設置輸入和輸出目錄 FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }運行結果：
[hadoop@h40 hui]$ hadoop fs -cat /user/hadoop/index_out/part-r-00000 GB <1999:1> IL <1998:1> US <1998:4>,<1999:6>總結
以上是生活随笔為你收集整理的MapReduce功能实现十---倒排索引(Inverted Index)的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： matlab处理大量数据
- 下一篇： 加密锁新功能应用