分布式缓存法计算矩阵乘法

  1 package matrix;
  2 
  3 import java.io.BufferedReader;
  4 import java.io.FileReader;
  5 import java.io.IOException;
  6 import java.net.URI;
  7 
  8 import org.apache.hadoop.conf.Configuration;
  9 import org.apache.hadoop.fs.Path;
 10 import org.apache.hadoop.io.*;
 11 import org.apache.hadoop.mapreduce.Job;
 12 import org.apache.hadoop.mapreduce.Mapper;
 13 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 14 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
 15 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 16 import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
 17 import org.apache.hadoop.filecache.DistributedCache;//分布式缓存库
 18 
 19 
 20 public class matrixProduct {
 21     public static class MyMapper extends Mapper<IntWritable, Text, 
 22     IntWritable, Text>{
 23         private int leftMatrixRowNum;
 24         private int leftMatrixColNum;
 25         private int rightMatrixRowNum;        
 26         private double[][] cacheMatrix;
 27         private double[] valueVector;
 28         private StringBuilder result=new StringBuilder();
 29         //@SuppressWarnings("resource")
 30         @Override
 31         protected void setup(
 32                 Mapper<IntWritable, Text, IntWritable, Text>.Context context)
 33                 throws IOException, InterruptedException {
 34             // TODO 自动生成的方法存根
 35             super.setup(context);
 36             leftMatrixRowNum=Integer.valueOf(context.getConfiguration().get("leftMatrixRowNum"));
 37             leftMatrixColNum=Integer.valueOf(context.getConfiguration().get("leftMatrixColNum"));
 38             rightMatrixRowNum=Integer.valueOf(context.getConfiguration().get("rightMatrixRowNum"));
 39             cacheMatrix=new double[leftMatrixRowNum][leftMatrixColNum];
 40             valueVector=new double[rightMatrixRowNum];
 41                         
 42             try {
 43                   Path[] cacheFiles=DistributedCache.getLocalCacheFiles(context.getConfiguration());
 44                 if(cacheFiles!=null&&cacheFiles.length>0){
 45                     String line;                    
 46                     BufferedReader dataReader=new BufferedReader(new FileReader(cacheFiles[0].toString()));
 47                     int i=-1;
 48                     while((line=dataReader.readLine())!=null){
 49                         ++i;
 50                         String[] eleStrings=line.split("\t");
 51                         for(int j=0;j<eleStrings.length;++j){
 52                             cacheMatrix[i][j]=Double.valueOf(eleStrings[j]).doubleValue();
 53                         }
 54                     }
 55                 }    
 56                                 
 57             } catch (Exception e) {
 58                 // TODO: handle exception        
 59                 System.out.println("setup exception");
 60             }
 61                     
 62         }
 63         
 64         @Override
 65         protected void map(IntWritable key, Text value,
 66                 Mapper<IntWritable, Text, IntWritable, Text>.Context context)
 67                 throws IOException, InterruptedException {
 68             // TODO 自动生成的方法存根
 69             super.map(key, value, context);
 70             String[] valueArray=value.toString().split("\t");
 71             
 72             for (int i = 0; i < valueArray.length; i++) {
 73                 valueVector[i] = Double.valueOf(valueArray[i]).doubleValue();                
 74             }
 75             double temp=0;
 76             for (int i=0;i<1043;++i) {
 77                 temp=0;
 78                 for (int j=0;j<1043;++j){
 79                 temp+=cacheMatrix[i][j]*valueVector[j];
 80                 }
 81                 if(i!=1042)
 82                     result.append(String.valueOf(temp)).append("\t");
 83                 else 
 84                     result.append(String.valueOf(temp)).append("\n");                    
 85                                     
 86             }
 87             context.write(key, new Text(result.toString()));
 88                     
 89         }
 90                         
 91     }
 92     public static void run(String s1,String s2,String s3,String leftMatrixRowNum,String leftMatrixColNum
 93             ,String rightMatrixRowNum)throws Exception{
 94         System.out.println("ewr");
 95         URI fileURI=new URI(s1);
 96         Configuration conf=new Configuration();
 97         conf.set("leftMatrixRowNum", leftMatrixRowNum);
 98         conf.set("leftMatrixColNum", leftMatrixColNum);
 99         conf.set("rightMatrixRowNum", rightMatrixRowNum);
100         Job job=new Job(conf,"matrix cache memory");
101         job.setJarByClass(matrixProduct.class);
102         job.setMapperClass(MyMapper.class);
103         job.setNumReduceTasks(0);
104         DistributedCache.addCacheFile(fileURI, conf);
105         job.setMapOutputKeyClass(IntWritable.class);
106         job.setMapOutputValueClass(Text.class);
107         job.setOutputKeyClass(IntWritable.class);
108         job.setOutputValueClass(Text.class);
109         job.setInputFormatClass(SequenceFileInputFormat.class);
110         job.setOutputFormatClass(SequenceFileOutputFormat.class);
111         FileInputFormat.setInputPaths(job, new Path(s2));
112         FileOutputFormat.setOutputPath(job, new Path(s3));    
113         System.exit(job.waitForCompletion(true)?0:1);                
114     }
115     
116     
117     
118     public static void main(String[] args) throws IOException, Exception{
119         
120         run(args[0], args[1],args[2],args[3],args[4],args[5]);
121         
122     }
123 
124 }