1、MapReduce实现矩阵相乘
一. 准备数据
1#!/bin/bash 2if [ $# -ne 3 ] 3then 4 echo "there must be 3 arguments to generate the two matries file!" 5 exit 1 6fi 7cat /dev/null > M_$1_$2 8cat /dev/null > N_$2_$3 9for i in `seq 1 $1` 10do 11 for j in `seq 1 $2` 12 do 13 s=$((RANDOM%100)) 14 echo -e "$i,$j\t$s" >>M_$1_$2 15 done 16done 17echo "we have built the matrix file M" 18 19for i in `seq 1 $2` 20 do 21 for j in ` seq 1 $3` 22 do 23 s=$((RANDOM%100)) 24 echo -e "$i,$j\t$s" >>N_$2_$3 25 done 26done 27echo "we have built the matrix file N"
用一下脚本语言准备数组数据
1M_3_2: 21,1 81 31,2 13 42,1 38 52,2 46 63,1 0 73,2 2 8 9N_2_4: 101,1 99 111,2 38 121,3 34 131,4 19 142,1 21 152,2 4 162,3 36 172,4 64
二. 计算
1public class Matrix { 2 3 private static class MatrixMapper extends 4 Mapper<LongWritable, Text, Text, Text> { 5 6 private static int colN = 0; 7 private static int rowM = 0; 8 9 @Override 10 protected void setup( 11 Mapper<LongWritable, Text, Text, Text>.Context context) 12 throws IOException, InterruptedException { 13 14 Configuration configuration = context.getConfiguration(); 15 colN = configuration.getInt("colN", 0); 16 rowM = configuration.getInt("rowM", 0); 17 18 } 19 20 @Override 21 protected void map(LongWritable key, Text value, 22 Mapper<LongWritable, Text, Text, Text>.Context context) 23 throws IOException, InterruptedException { 24 25 FileSplit fileSplit = (FileSplit) context.getInputSplit(); 26 String fileName = fileSplit.getPath().getName(); 27 String[] strings = value.toString().split(","); 28 int i = Integer.parseInt(strings[0]); 29 String[] ser = strings[1].split("\t"); 30 int j = Integer.parseInt(ser[0]); 31 int val = Integer.parseInt(ser[1]); 32 33 if (fileName.startsWith("M")) { 34 35 for (int count = 1; count <= colN; count++) { 36 context.write(new Text(i + "," + count), new Text("M," + j 37 + "," + val + "")); 38 } 39 40 } else { 41 42 for (int count = 1; count <= rowM; count++) { 43 context.write(new Text(count + "," + j), new Text("N," + i 44 + "," + val + "")); 45 } 46 47 } 48 } 49 } 50 51 private static class MatrixReduce extends 52 Reducer<Text, Text, Text, IntWritable> { 53 54 private static int rowM = 0; 55 56 @Override 57 protected void setup( 58 Reducer<Text, Text, Text, IntWritable>.Context context) 59 throws IOException, InterruptedException { 60 61 Configuration configuration = context.getConfiguration(); 62 rowM = configuration.getInt("rowM", 0); 63 64 } 65 66 @Override 67 protected void reduce(Text key, Iterable<Text> values, 68 Reducer<Text, Text, Text, IntWritable>.Context context) 69 throws IOException, InterruptedException { 70 71 int sumValue = 0; 72 int[] m_Arr = new int[rowM + 1]; 73 int[] n_Arr = new int[rowM + 1]; 74 75 for (Text value : values) { 76 77 String string = value.toString(); 78 String[] strings = string.split(","); 79 80 if (strings[0].equals("M")) { 81 m_Arr[Integer.parseInt(strings[1])] = Integer 82 .parseInt(strings[2]); 83 } else { 84 n_Arr[Integer.parseInt(strings[1])] = Integer 85 .parseInt(strings[2]); 86 } 87 } 88 89 for (int i = 1; i < rowM + 1; i++) { 90 sumValue += m_Arr[i] * n_Arr[i]; 91 } 92 93 context.write(key, new IntWritable(sumValue)); 94 } 95 96 } 97 98 public static void main(String[] args) throws IllegalArgumentException, 99 IOException, ClassNotFoundException, InterruptedException { 100 101 Configuration configuration = HadoopConfig.getConfiguration(); 102 configuration.setInt("colN", 4); 103 configuration.setInt("rowN", 2); 104 configuration.setInt("colM", 2); 105 configuration.setInt("rowM", 3); 106 107 Job job = Job.getInstance(configuration, "矩阵相乘"); 108 109 job.setJarByClass(Sort.class); 110 job.setMapperClass(MatrixMapper.class); 111 112 job.setMapOutputKeyClass(Text.class); 113 job.setMapOutputValueClass(Text.class); 114 job.setOutputKeyClass(Text.class); 115 job.setOutputValueClass(IntWritable.class); 116 117 job.setReducerClass(MatrixReduce.class); 118 FileInputFormat.addInputPath(job, new Path("/matrix")); 119 FileOutputFormat.setOutputPath(job, new Path("/matrixOutput")); 120 job.waitForCompletion(true); 121 System.exit(job.waitForCompletion(true) ? 0 : 1); 122 123 } 124 125}
三. 结果
11,1 8292 21,2 3130 31,3 3222 41,4 2371 52,1 4728 62,2 1628 72,3 2948 82,4 3666 93,1 42 103,2 8 113,3 72 123,4 128
2、MapReduce实现倒排索引
一、准备数据
1file1: 2one fish 3two bird 4two monkey 5 6file2: 7two peach 8three watermelon
二、计算
1public class InvertIndex { 2 3 private static class InvertIndexMapper extends 4 Mapper<LongWritable, Text, Text, Text> { 5 6 @Override 7 protected void map(LongWritable key, Text value, 8 Mapper<LongWritable, Text, Text, Text>.Context context) 9 throws IOException, InterruptedException { 10 11 FileSplit fileSplit = (FileSplit) context.getInputSplit(); 12 String fileName = fileSplit.getPath().toString(); 13 String[] words = value.toString().split(" "); 14 for (String string : words) { 15 context.write(new Text(string), new Text(fileName + "#" + key.toString())); 16 } 17 18 } 19 20 } 21 22 private static class InvertIndexReduce extends 23 Reducer<Text, Text, Text, Text> { 24 25 @Override 26 protected void reduce(Text key, Iterable<Text> values, 27 Reducer<Text, Text, Text, Text>.Context context) 28 throws IOException, InterruptedException { 29 30 StringBuilder stringBuilder = new StringBuilder(); 31 32 for (Text text : values) { 33 stringBuilder.append(text.toString()).append(";"); 34 } 35 36 context.write(key, new Text(stringBuilder.toString())); 37 } 38 } 39 40 public static void main(String[] args) throws IOException, 41 ClassNotFoundException, InterruptedException{ 42 43 Configuration configuration = HadoopConfig.getConfiguration(); 44 Job job = Job.getInstance(configuration, "倒排索引"); 45 job.setJarByClass(InvertIndex.class); 46 job.setMapperClass(InvertIndexMapper.class); 47 job.setMapOutputKeyClass(Text.class); 48 job.setMapOutputValueClass(Text.class); 49 job.setReducerClass(InvertIndexReduce.class); 50 job.setOutputKeyClass(Text.class); 51 job.setOutputValueClass(Text.class); 52 53 FileInputFormat.addInputPath(job, new Path("/data")); 54 FileOutputFormat.setOutputPath(job, new Path("/ouput")); 55 job.waitForCompletion(true); 56 System.exit(job.waitForCompletion(true) ? 0 : 1); 57 58 }
三、结果
1bird hdfs://127.0.0.1:8020/data/file1#9; 2fish hdfs://127.0.0.1:8020/data/file1#0; 3monkey hdfs://127.0.0.1:8020/data/file1#18; 4one hdfs://127.0.0.1:8020/data/file1#0; 5peach hdfs://127.0.0.1:8020/data/file2#0; 6three hdfs://127.0.0.1:8020/data/file2#10; 7two hdfs://127.0.0.1:8020/data/file2#0;hdfs://127.0.0.1:8020/data/file1#18;hdfs://127.0.0.1:8020/data/file1#9; 8watermelon hdfs://127.0.0.1:8020/data/file2#10;
3、MapReduce实现复杂倒排索引
一、准备数据
1file1: 2one fish 3two bird 4two monkey 5 6file2: 7two peach 8three watermelon
二、计算
1public class ComplexInvertIndex { 2 3 private static class FileNameRecordReader extends RecordReader<Text, Text> { 4 5 LineRecordReader lineRecordReader = new LineRecordReader(); 6 String fileName; 7 8 @Override 9 public void initialize(InputSplit split, TaskAttemptContext context) 10 throws IOException, InterruptedException { 11 lineRecordReader.initialize(split, context); 12 fileName = ((FileSplit) split).getPath().getName(); 13 } 14 15 @Override 16 public boolean nextKeyValue() throws IOException, InterruptedException { 17 return lineRecordReader.nextKeyValue(); 18 } 19 20 @Override 21 public Text getCurrentKey() throws IOException, InterruptedException { 22 return new Text(fileName); 23 } 24 25 @Override 26 public Text getCurrentValue() throws IOException, InterruptedException { 27 return lineRecordReader.getCurrentValue(); 28 } 29 30 @Override 31 public float getProgress() throws IOException, InterruptedException { 32 return lineRecordReader.getProgress(); 33 } 34 35 @Override 36 public void close() throws IOException { 37 lineRecordReader.close(); 38 } 39 40 } 41 42 private static class FileNameInputFormat extends 43 FileInputFormat<Text, Text> { 44 45 @Override 46 public RecordReader<Text, Text> createRecordReader(InputSplit split, 47 TaskAttemptContext context) throws IOException, 48 InterruptedException { 49 FileNameRecordReader fileNameRecordReader = new FileNameRecordReader(); 50 fileNameRecordReader.initialize(split, context); 51 return fileNameRecordReader; 52 } 53 54 } 55 56 private static class ComplexInvertIndexMapper extends 57 Mapper<Text, Text, Text, IntWritable> { 58 59 @Override 60 protected void map(Text key, Text value, 61 Mapper<Text, Text, Text, IntWritable>.Context context) 62 throws IOException, InterruptedException { 63 64 String[] strs = value.toString().split(" "); 65 for (String string : strs) { 66 context.write(new Text( string+"#"+key.toString() ),new IntWritable(1)); 67 } 68 69 } 70 71 } 72 73 private static class ComplexInvertIndexCombiner extends 74 Reducer<Text, IntWritable, Text, IntWritable> { 75 76 @Override 77 protected void reduce(Text key, Iterable<IntWritable> values, 78 Reducer<Text, IntWritable, Text, IntWritable>.Context context) 79 throws IOException, InterruptedException { 80 81 int sum = 0; 82 for (IntWritable value : values) { 83 sum += value.get(); 84 } 85 context.write(key,new IntWritable(sum)); 86 System.out.println(key.toString() + sum +""); 87 } 88 89 } 90 91 //把key的前面字段聚合，排序 92 private static class InvertIndexPartitioner extends 93 HashPartitioner<Text, IntWritable> { 94 95 @Override 96 public int getPartition(Text key, IntWritable value, int numReduceTasks) { 97 String[] strs = key.toString().split("#"); 98 return super.getPartition(new Text(strs[0]), value, numReduceTasks); 99 } 100 101 } 102 103 private static class ComplexInvertIndexReduce extends 104 Reducer<Text, IntWritable, Text, Text> { 105 106 static Map<String, String> map = new HashMap<String, String>(); 107 108 @Override 109 protected void reduce(Text key, Iterable<IntWritable> values, 110 Reducer<Text, IntWritable, Text, Text>.Context context) 111 throws IOException, InterruptedException { 112 113 String[] strings = key.toString().split("#"); 114 String word = strings[0]; 115 String doc = strings[1]; 116 int sum = 0; 117 for(IntWritable value : values){ 118 sum = sum + value.get(); 119 } 120 if( map.get(word) == null ){ 121 map.put(word," ("+doc+","+sum+") "); 122 }else{ 123 map.put(word,map.get(word)+" ("+doc+","+sum+") "); 124 } 125 126 } 127 128 @Override 129 protected void cleanup( 130 Reducer<Text, IntWritable, Text, Text>.Context context) 131 throws IOException, InterruptedException { 132 for(String key:map.keySet()){ 133 context.write(new Text(key), new Text(map.get(key))); 134 } 135 } 136 137 } 138 139 public static void main(String[] args)throws IOException, 140 ClassNotFoundException, InterruptedException{ 141 142 Configuration configuration = HadoopConfig.getConfiguration(); 143 Job job = Job.getInstance(configuration, "复杂倒排索引"); 144 job.setJarByClass(ComplexInvertIndex.class); 145 job.setInputFormatClass(FileNameInputFormat.class); 146 job.setMapperClass(ComplexInvertIndexMapper.class); 147 job.setMapOutputKeyClass(Text.class); 148 job.setMapOutputValueClass(IntWritable.class); 149 job.setCombinerClass(ComplexInvertIndexCombiner.class); 150 job.setReducerClass(ComplexInvertIndexReduce.class); 151 job.setPartitionerClass(InvertIndexPartitioner.class); 152 job.setOutputKeyClass(Text.class); 153 job.setOutputValueClass(Text.class); 154 155 FileInputFormat.addInputPath(job, new Path("/data")); 156 FileOutputFormat.setOutputPath(job, new Path("/ouputdata")); 157 job.waitForCompletion(true); 158 System.exit(job.waitForCompletion(true) ? 0 : 1); 159 160 }
三、结果查看
1monkey (file1,1) 2bird (file1,1) 3fish (file1,1) 4one (file1,1) 5peach (file2,1) 6watermelon (file2,1) 7three (file2,1) 8two (file1,2) (file2,1)