集群环境下使用分布式缓存作矩阵乘法

虽然是在IDEA中完成了使用分布式缓存的方法进行矩阵乘积，不过当将源码编译，打成jar包，上传到集群上运行的时候却一直报“ no job jar file set”,"ClassNotFound"，其实就是找不到jar包，原因应该是与昨天更改了系统环境有关(因为使用hadoop自带的WordCount也不行)，经历了重新格式话HDFS、关闭什么防火墙....最后终于解决了问题，请参考(“no job jar file set”原因)。具体的代码如下(第132行是解决问题的关键)：

/**
 * Created with IntelliJ IDEA.
 * User: hadoop
 * Date: 16-3-6
 * Time: 下午12:47
 * To change this template use File | Settings | File Templates.
 */
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import java.io.IOException;
import java.lang.reflect.Array;
import java.net.URI;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.util.ReflectionUtils;

public class MutiDoubleInputMatrixProduct {
    public static  class MyMapper extends Mapper<IntWritable,DoubleArrayWritable,IntWritable,DoubleArrayWritable>{
        public DoubleArrayWritable map_value=new DoubleArrayWritable();
        public  double[][] leftMatrix=null;/******************************************/
        public Object obValue=null;
        public DoubleWritable[] arraySum=null;
        public double sum=0;
        public void setup(Context context) throws IOException {
            Configuration conf=context.getConfiguration();
            leftMatrix=new double[conf.getInt("leftMatrixRowNum",10)][conf.getInt("leftMatrixColumnNum",10)];
            System.out.println("map setup() start!");
            //URI[] cacheFiles=DistributedCache.getCacheFiles(context.getConfiguration());
            Path[] cacheFiles=DistributedCache.getLocalCacheFiles(conf);
            String localCacheFile="file://"+cacheFiles[0].toString();
            //URI[] cacheFiles=DistributedCache.getCacheFiles(conf);
            //DistributedCache.
            System.out.println("local path is:"+cacheFiles[0].toString());
            // URI[] cacheFiles=DistributedCache.getCacheFiles(context.getConfiguration());
            FileSystem fs =FileSystem.get(URI.create(localCacheFile), conf);
            SequenceFile.Reader reader=null;
            reader=new SequenceFile.Reader(fs,new Path(localCacheFile),conf);
            IntWritable key= (IntWritable)ReflectionUtils.newInstance(reader.getKeyClass(),conf);
            DoubleArrayWritable value= (DoubleArrayWritable)ReflectionUtils.newInstance(reader.getValueClass(),conf);
            int valueLength=0;
            int rowIndex=0;
            while (reader.next(key,value)){
                obValue=value.toArray();
                rowIndex=key.get();
                if(rowIndex<1){
                    valueLength=Array.getLength(obValue);
                }
                leftMatrix[rowIndex]=new double[conf.getInt("leftMatrixColumnNum",10)];
                //this.leftMatrix=new double[valueLength][Integer.parseInt(context.getConfiguration().get("leftMatrixColumnNum"))];
                for (int i=0;i<valueLength;++i){
                    leftMatrix[rowIndex][i]=Double.parseDouble(Array.get(obValue, i).toString());
                }

            }
        }
        public void map(IntWritable key,DoubleArrayWritable value,Context context) throws IOException, InterruptedException {
            obValue=value.toArray();
            InputSplit inputSplit=context.getInputSplit();
            String fileName=((FileSplit)inputSplit).getPath().getName();
            if (fileName.startsWith("FB")) {
                context.write(key,value);
            }
            else{
                arraySum=new DoubleWritable[this.leftMatrix.length];
                for (int i=0;i<this.leftMatrix.length;++i){
                    sum=0;
                    for (int j=0;j<this.leftMatrix[0].length;++j){
                        sum+= this.leftMatrix[i][j]*Double.parseDouble(Array.get(obValue,j).toString())*(double)(context.getConfiguration().getFloat("u",1f));
                    }
                    arraySum[i]=new DoubleWritable(sum);
                    //arraySum[i].set(sum);
                }
                map_value.set(arraySum);
                context.write(key,map_value);
            }
        }
    }
    public static class MyReducer extends Reducer<IntWritable,DoubleArrayWritable,IntWritable,DoubleArrayWritable>{
        public DoubleWritable[] sum=null;
        public Object obValue=null;
        public DoubleArrayWritable valueArrayWritable=null;

        public void setup(Context context){
            int leftMatrixColumnNum=context.getConfiguration().getInt("leftMatrixColumnNum",100);
            sum=new DoubleWritable[leftMatrixColumnNum];
            for (int i=0;i<leftMatrixColumnNum;++i){
                sum[i]=new DoubleWritable(0.0);
            }
        }

        public void reduce(IntWritable key,Iterable<DoubleArrayWritable>value,Context context) throws IOException, InterruptedException {
            int valueLength=0;
            for(DoubleArrayWritable doubleValue:value){
                obValue=doubleValue.toArray();
                valueLength=Array.getLength(obValue);
                for (int i=0;i<valueLength;++i){
                    sum[i]=new DoubleWritable(Double.parseDouble(Array.get(obValue,i).toString())+sum[i].get());
                }
            }
            valueArrayWritable=new DoubleArrayWritable();
            valueArrayWritable.set(sum);
            context.write(key,valueArrayWritable);
            for (int i=0;i<sum.length;++i){
                sum[i].set(0.0);
            }

        }
    }

    public static void main(String[]args) throws IOException, ClassNotFoundException, InterruptedException {
        String uri="/testData/input/b100";
        String outUri="/sOutput";
        String cachePath="/testData/F100";
        Configuration conf=new Configuration();
        DistributedCache.addCacheFile(URI.create(cachePath),conf);//添加分布式缓存
        FileSystem fs=FileSystem.get(URI.create(uri),conf);
        fs.delete(new Path(outUri),true);
        conf.setInt("leftMatrixColumnNum",100);
        conf.setInt("leftMatrixRowNum",100);
        conf.setFloat("u",0.5f);
        conf.set("mapred.jar","MutiDoubleInputMatrixProduct.jar");
        Job job=new Job(conf,"MultiMatrix2");       
        job.setJarByClass(MutiDoubleInputMatrixProduct.class);
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(DoubleArrayWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(DoubleArrayWritable.class);
        FileInputFormat.setInputPaths(job, new Path(uri));
        FileOutputFormat.setOutputPath(job,new Path(outUri));
        System.exit(job.waitForCompletion(true)?0:1);
    }


}
 class DoubleArrayWritable extends ArrayWritable {
    public DoubleArrayWritable(){
        super(DoubleWritable.class);
    }
    public String toString(){
        StringBuilder sb=new StringBuilder();
        for (Writable val:get()){
            DoubleWritable doubleWritable=(DoubleWritable)val;
            sb.append(doubleWritable.get());
            sb.append(",");
        }
        sb.deleteCharAt(sb.length()-1);
        return sb.toString();
    }
}