矩阵相乘的mapreduce操作fo

有借鉴，借鉴来源同上一条博客，侵删。

map

public class StepFourMapper extends Mapper {


    DecimalFormat df = new DecimalFormat("0.00");
    List usersList = new ArrayList();
    List> item_of_scoreList = new ArrayList>();
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        
        FileReader fileReader = new FileReader("itemUserScore2");

        BufferedReader bufferedReader = new BufferedReader(fileReader);
        // 将数据按需要的格式存入内存中
        String line = null;
        while((line = bufferedReader.readLine()) != null){
            usersList.add(line.split("t")[0]);
            String[] item_of_score = line.split("t")[1].split(",");
            MapsingleMap = new HashMap();
            for(String singleScore:item_of_score){
                String itemID = singleScore.split("_")[0];
                String score = singleScore.split("_")[1];
                singleMap.put(itemID,score);
            }
            item_of_scoreList.add(singleMap);
        }
        fileReader.close();
        bufferedReader.close();
    }

    Text outKey = new Text();
    Text outValue = new Text();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
   
        String itemID = value.toString().split("t")[0];
        String[] score_of_items = value.toString().split("t")[1].split(",");
        // 首先遍历每一个用户
        for(int i = 0; i < usersList.size(); i++){
            // 得到该用户评分过（购买过）的商品
            Map item_of_score = item_of_scoreList.get(i);
            // 得到当前商品的与其它所有商品的相似性列表
            double sum = 0.0;
            for(String singleItem:score_of_items){
                String singleItemID = singleItem.split("_")[0];
                // 判断当前用户是否评分过该商品，若没有，则默认为0
                if(item_of_score.containsKey(singleItemID)){
                    String score = singleItem.split("_")[1];
                    sum += Double.parseDouble(score)*Double.parseDouble(item_of_score.get(singleItemID));
                }
            }
            if(sum == 0){
                continue;
            }
            outKey.set(usersList.get(i));
            outValue.set(itemID+"_"+String.valueOf(df.format(sum)));
            context.write(outKey,outValue);
        }
    }
}

reduce

 @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        StringBuffer item_of_sup = new StringBuffer();
        for(Text val:values){
            item_of_sup.append(val).append(",");
        }
        // 去掉最后一个逗号
        String line = item_of_sup.substring(0,item_of_sup.length()-1);
        
        String tablename = new String("myprojectRepeat");
        if(isTableExist(tablename)){
            //实例：put student,95001,Sname,LiYing
            putDate(tablename,key.toString(),"recommendlist","items",line);
        }
        else{
            createTable(tablename,"recommendlist");
            putDate(tablename,key.toString(),"recommendlist","items",line);
        }
        outKey.set(key);
        outValue.set(line);
        context.write(outKey,outValue);
    }
}

driver

public class StepFourDriver {

    //转置评分矩阵路径
    static String cache = new String("/llk/myproject/output/step3/part-r-00000");
    //相似矩阵路径
    static String inputPath = new String(/output/step2/part-r-00000");
    static String outputPath = new String("/output/step4/");
    public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException, URISyntaxException {
        // 1 获取配置信息以及获取 job 对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        // 2 关联本 Driver 程序的 jar
        job.setJarByClass(StepFourDriver.class);
        // 3 关联 Mapper 和 Reducer 的 jar
        job.setMapperClass(StepFourMapper.class);
        job.setReducerClass(StepFourReducer.class);
        if(args.length > 2){
            inputPath = args[0];
            outputPath = args[1];
            cache = args[2];
        }
        // 将文件路径添加为URI
        job.addCacheArchive(new URI(cache + "#itemUserScore2"));

        // 4 设置 Mapper 输出的 kv 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        // 5 设置最终输出 kv 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        // 6 设置输入和输出路径
        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
     

        // 7 提交 job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0: 1);
    }
}