基于Hadoop的商品推荐（老师版）

package com.briup.bigdata.project.grms.main; import com.briup.bigdata.project.grms.step1.UserBuyGoodsList; import com.briup.bigdata.project.grms.step2.GoodsCooccurrenceList; import com.briup.bigdata.project.grms.step3.GoodsCooccurrenceMatrix; import com.briup.bigdata.project.grms.step4.UserBuyGoodsVector; import com.briup.bigdata.project.grms.step5.MultiplyGoodsMatrixAndUserVector; import com.briup.bigdata.project.grms.step6.MakeSumForMultiplication; import com.briup.bigdata.project.grms.step7.DuplicateDataForResult; import com.briup.bigdata.project.grms.step8.RecommendResultDB; import com.briup.bigdata.project.grms.step8.SaveRecommendResultToDB; import java.util.Properties; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.db.DBConfiguration; import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat; import org.apache.hadoop.mapreduce.lib.input.MultipleInputs; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob; import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; public class JobController extends Configured implements Tool{ @Override public final int run(String[] args) throws Exception{ Configuration conf=getConf(); Path inStep1=new Path(conf.get("inStep1")); Path outStep1=new Path(conf.get("outStep1")); Path outStep2=new Path(conf.get("outStep2")); Path outStep3=new Path(conf.get("outStep3")); Path outStep4=new Path(conf.get("outStep4")); Path outStep5=new Path(conf.get("outStep5")); Path outStep6=new Path(conf.get("outStep6")); Path outStep7=new Path(conf.get("outStep7")); //----第1步作业配置-------------------- Job job1=Job.getInstance(conf,UserBuyGoodsList.class.getSimpleName()); job1.setJarByClass(this.getClass()); job1.setMapperClass(UserBuyGoodsList.UserBuyGoodsListMapper.class); job1.setMapOutputKeyClass(Text.class); job1.setMapOutputValueClass(Text.class); job1.setInputFormatClass(TextInputFormat.class); // 数据来源：原始数据 TextInputFormat.addInputPath(job1,inStep1); job1.setReducerClass(UserBuyGoodsList.UserBuyGoodsListReducer.class); job1.setOutputKeyClass(Text.class); job1.setOutputValueClass(Text.class); job1.setOutputFormatClass(TextOutputFormat.class); TextOutputFormat.setOutputPath(job1,outStep1); //----第2步作业配置-------------------- Job job2=Job.getInstance(conf,GoodsCooccurrenceList.class.getSimpleName()); job2.setJarByClass(this.getClass()); job2.setMapperClass(GoodsCooccurrenceList.GoodsCooccurrenceListMapper.class); job2.setMapOutputKeyClass(Text.class); job2.setMapOutputValueClass(Text.class); job2.setInputFormatClass(TextInputFormat.class); // 数据来源：第1步的计算结果 TextInputFormat.addInputPath(job2,outStep1); job2.setReducerClass(GoodsCooccurrenceList.GoodsCooccurrenceListReducer.class); job2.setOutputKeyClass(Text.class); job2.setOutputValueClass(Text.class); job2.setOutputFormatClass(TextOutputFormat.class); TextOutputFormat.setOutputPath(job2,outStep2); //----第3步作业配置-------------------- Job job3=Job.getInstance(conf,GoodsCooccurrenceMatrix.class.getSimpleName()); job3.setJarByClass(this.getClass()); job3.setMapperClass(GoodsCooccurrenceMatrix.GoodsCooccurrenceMatrixMapper.class); job3.setMapOutputKeyClass(Text.class); job3.setMapOutputValueClass(Text.class); job3.setInputFormatClass(TextInputFormat.class); // 数据来源：第2步的计算结果 TextInputFormat.addInputPath(job3,outStep2); job3.setReducerClass(GoodsCooccurrenceMatrix.GoodsCooccurrenceMatrixReducer.class); job3.setOutputKeyClass(Text.class); job3.setOutputValueClass(Text.class); job3.setOutputFormatClass(TextOutputFormat.class); TextOutputFormat.setOutputPath(job3,outStep3); //----第4步作业配置-------------------- Job job4=Job.getInstance(conf,UserBuyGoodsVector.class.getSimpleName()); job4.setJarByClass(this.getClass()); job4.setMapperClass(UserBuyGoodsVector.UserBuyGoodsVectorMapper.class); job4.setMapOutputKeyClass(Text.class); job4.setMapOutputValueClass(Text.class); job4.setInputFormatClass(TextInputFormat.class); // 数据来源：第1步的计算结果或者原始数据 TextInputFormat.addInputPath(job4,outStep1); job4.setReducerClass(UserBuyGoodsVector.UserBuyGoodsVectorReducer.class); job4.setOutputKeyClass(Text.class); job4.setOutputValueClass(Text.class); job4.setOutputFormatClass(TextOutputFormat.class); TextOutputFormat.setOutputPath(job4,outStep4); //----第5步作业配置-------------------- Job job5=Job.getInstance(conf,MultiplyGoodsMatrixAndUserVector.class.getSimpleName()); job5.setJarByClass(this.getClass()); // 数据来源：第1步的计算结果或者原始数据 MultipleInputs.addInputPath(job5,outStep3,TextInputFormat.class,MultiplyGoodsMatrixAndUserVector.MultiplyGoodsMatrixAndUserVectorFirstMapper.class); MultipleInputs.addInputPath(job5,outStep4,TextInputFormat.class,MultiplyGoodsMatrixAndUserVector.MultiplyGoodsMatrixAndUserVectorSecondMapper.class); job5.setMapOutputKeyClass(Text.class); job5.setMapOutputValueClass(Text.class); job5.setReducerClass(MultiplyGoodsMatrixAndUserVector.MultiplyGoodsMatrixAndUserVectorReducer.class); job5.setOutputKeyClass(Text.class); job5.setOutputValueClass(LongWritable.class); job5.setOutputFormatClass(TextOutputFormat.class); TextOutputFormat.setOutputPath(job5,outStep5); //----第6步作业配置-------------------- Job job6=Job.getInstance(conf,MakeSumForMultiplication.class.getSimpleName()); job6.setJarByClass(this.getClass()); job6.setMapperClass(MakeSumForMultiplication.MakeSumForMultiplicationMapper.class); job6.setMapOutputKeyClass(Text.class); job6.setMapOutputValueClass(LongWritable.class); job6.setInputFormatClass(TextInputFormat.class); // 数据来源：第5步的计算结果 TextInputFormat.addInputPath(job6,outStep5); job6.setReducerClass(MakeSumForMultiplication.MakeSumForMultiplicationReducer.class); job6.setOutputKeyClass(Text.class); job6.setOutputValueClass(LongWritable.class); job6.setOutputFormatClass(TextOutputFormat.class); TextOutputFormat.setOutputPath(job6,outStep6); //----第7步作业配置-------------------- Job job7=Job.getInstance(conf,DuplicateDataForResult.class.getSimpleName()); job7.setJarByClass(this.getClass()); MultipleInputs.addInputPath(job7,outStep1,TextInputFormat.class,DuplicateDataForResult.DuplicateDataForResultFirstMapper.class); MultipleInputs.addInputPath(job7,outStep6,TextInputFormat.class,DuplicateDataForResult.DuplicateDataForResultSecondMapper.class); job7.setMapOutputKeyClass(Text.class); job7.setMapOutputValueClass(Text.class); job7.setReducerClass(DuplicateDataForResult.DuplicateDataForResultReducer.class); job7.setOutputKeyClass(Text.class); job7.setOutputValueClass(Text.class); job7.setOutputFormatClass(TextOutputFormat.class); TextOutputFormat.setOutputPath(job7,outStep7); job7.setNumReduceTasks(1); //----第8步作业配置-------------------- Job job8=Job.getInstance(conf,SaveRecommendResultToDB.class.getSimpleName()); job8.setJarByClass(this.getClass()); job8.setMapperClass(SaveRecommendResultToDB.SaveRecommendResultToDBMapper.class); job8.setMapOutputKeyClass(Text.class); job8.setMapOutputValueClass(Text.class); job8.setInputFormatClass(TextInputFormat.class); TextInputFormat.addInputPath(job8,outStep7); job8.se