forked from bsspirit/maven_hadoop_template
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
257 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
141 changes: 141 additions & 0 deletions
141
src/main/java/org/conan/myhadoop/recommend/Step4_Update.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,141 @@ | ||
| package org.conan.myhadoop.recommend; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.HashMap; | ||
| import java.util.Iterator; | ||
| import java.util.Map; | ||
|
|
||
| import org.apache.hadoop.fs.Path; | ||
| import org.apache.hadoop.io.LongWritable; | ||
| import org.apache.hadoop.io.Text; | ||
| import org.apache.hadoop.mapred.JobConf; | ||
| import org.apache.hadoop.mapreduce.Job; | ||
| import org.apache.hadoop.mapreduce.Mapper; | ||
| import org.apache.hadoop.mapreduce.Reducer; | ||
| import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; | ||
| import org.apache.hadoop.mapreduce.lib.input.FileSplit; | ||
| import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; | ||
| import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; | ||
| import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; | ||
| import org.conan.myhadoop.hdfs.HdfsDAO; | ||
|
|
||
| public class Step4_Update { | ||
|
|
||
| public static class Step4_PartialMultiplyMapper extends Mapper<LongWritable, Text, Text, Text> { | ||
|
|
||
| private String flag;// A同现矩阵 or B评分矩阵 | ||
|
|
||
| @Override | ||
| protected void setup(Context context) throws IOException, InterruptedException { | ||
| FileSplit split = (FileSplit) context.getInputSplit(); | ||
| flag = split.getPath().getParent().getName();// 判断读的数据集 | ||
|
|
||
| // System.out.println(flag); | ||
| } | ||
|
|
||
| @Override | ||
| public void map(LongWritable key, Text values, Context context) throws IOException, InterruptedException { | ||
| String[] tokens = Recommend.DELIMITER.split(values.toString()); | ||
|
|
||
| if (flag.equals("step3_2")) {// 同现矩阵 | ||
| String[] v1 = tokens[0].split(":"); | ||
| String itemID1 = v1[0]; | ||
| String itemID2 = v1[1]; | ||
| String num = tokens[1]; | ||
|
|
||
| Text k = new Text(itemID1); | ||
| Text v = new Text("A:" + itemID2 + "," + num); | ||
|
|
||
| context.write(k, v); | ||
| // System.out.println(k.toString() + " " + v.toString()); | ||
|
|
||
| } else if (flag.equals("step3_1")) {// 评分矩阵 | ||
| String[] v2 = tokens[1].split(":"); | ||
| String itemID = tokens[0]; | ||
| String userID = v2[0]; | ||
| String pref = v2[1]; | ||
|
|
||
| Text k = new Text(itemID); | ||
| Text v = new Text("B:" + userID + "," + pref); | ||
|
|
||
| context.write(k, v); | ||
| // System.out.println(k.toString() + " " + v.toString()); | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| public static class Step4_AggregateReducer extends Reducer<Text, Text, Text, Text> { | ||
|
|
||
| @Override | ||
| public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { | ||
| System.out.println(key.toString() + ":"); | ||
|
|
||
| Map<String, String> mapA = new HashMap<String, String>(); | ||
| Map<String, String> mapB = new HashMap<String, String>(); | ||
|
|
||
| for (Text line : values) { | ||
| String val = line.toString(); | ||
| System.out.println(val); | ||
|
|
||
| if (val.startsWith("A:")) { | ||
| String[] kv = Recommend.DELIMITER.split(val.substring(2)); | ||
| mapA.put(kv[0], kv[1]); | ||
|
|
||
| } else if (val.startsWith("B:")) { | ||
| String[] kv = Recommend.DELIMITER.split(val.substring(2)); | ||
| mapB.put(kv[0], kv[1]); | ||
|
|
||
| } | ||
| } | ||
|
|
||
| double result = 0; | ||
| Iterator<String> iter = mapA.keySet().iterator(); | ||
| while (iter.hasNext()) { | ||
| String mapk = iter.next();// itemID | ||
|
|
||
| int num = Integer.parseInt(mapA.get(mapk)); | ||
| Iterator<String> iterb = mapB.keySet().iterator(); | ||
| while (iterb.hasNext()) { | ||
| String mapkb = iterb.next();// userID | ||
| double pref = Double.parseDouble(mapB.get(mapkb)); | ||
| result = num * pref;// 矩阵乘法相乘计算 | ||
|
|
||
| Text k = new Text(mapkb); | ||
| Text v = new Text(mapk + "," + result); | ||
| context.write(k, v); | ||
| System.out.println(k.toString() + " " + v.toString()); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public static void run(Map<String, String> path) throws IOException, InterruptedException, ClassNotFoundException { | ||
| JobConf conf = Recommend.config(); | ||
|
|
||
| String input1 = path.get("Step5Input1"); | ||
| String input2 = path.get("Step5Input2"); | ||
| String output = path.get("Step5Output"); | ||
|
|
||
| HdfsDAO hdfs = new HdfsDAO(Recommend.HDFS, conf); | ||
| hdfs.rmr(output); | ||
|
|
||
| Job job = new Job(conf); | ||
| job.setJarByClass(Step4_Update.class); | ||
|
|
||
| job.setOutputKeyClass(Text.class); | ||
| job.setOutputValueClass(Text.class); | ||
|
|
||
| job.setMapperClass(Step4_Update.Step4_PartialMultiplyMapper.class); | ||
| job.setReducerClass(Step4_Update.Step4_AggregateReducer.class); | ||
|
|
||
| job.setInputFormatClass(TextInputFormat.class); | ||
| job.setOutputFormatClass(TextOutputFormat.class); | ||
|
|
||
| FileInputFormat.setInputPaths(job, new Path(input1), new Path(input2)); | ||
| FileOutputFormat.setOutputPath(job, new Path(output)); | ||
|
|
||
| job.waitForCompletion(true); | ||
| } | ||
|
|
||
| } |
91 changes: 91 additions & 0 deletions
91
src/main/java/org/conan/myhadoop/recommend/Step4_Update2.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| package org.conan.myhadoop.recommend; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.HashMap; | ||
| import java.util.Iterator; | ||
| import java.util.Map; | ||
|
|
||
| import org.apache.hadoop.fs.Path; | ||
| import org.apache.hadoop.io.LongWritable; | ||
| import org.apache.hadoop.io.Text; | ||
| import org.apache.hadoop.mapred.JobConf; | ||
| import org.apache.hadoop.mapreduce.Job; | ||
| import org.apache.hadoop.mapreduce.Mapper; | ||
| import org.apache.hadoop.mapreduce.Reducer; | ||
| import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; | ||
| import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; | ||
| import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; | ||
| import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; | ||
| import org.conan.myhadoop.hdfs.HdfsDAO; | ||
|
|
||
| public class Step4_Update2 { | ||
|
|
||
| public static class Step4_RecommendMapper extends Mapper<LongWritable, Text, Text, Text> { | ||
|
|
||
| @Override | ||
| public void map(LongWritable key, Text values, Context context) throws IOException, InterruptedException { | ||
| String[] tokens = Recommend.DELIMITER.split(values.toString()); | ||
| Text k = new Text(tokens[0]); | ||
| Text v = new Text(tokens[1]+","+tokens[2]); | ||
| context.write(k, v); | ||
| } | ||
| } | ||
|
|
||
| public static class Step4_RecommendReducer extends Reducer<Text, Text, Text, Text> { | ||
|
|
||
| @Override | ||
| public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { | ||
| System.out.println(key.toString() + ":"); | ||
| Map<String, Double> map = new HashMap<String, Double>();// 结果 | ||
|
|
||
| for (Text line : values) { | ||
| System.out.println(line.toString()); | ||
| String[] tokens = Recommend.DELIMITER.split(line.toString()); | ||
| String itemID = tokens[0]; | ||
| Double score = Double.parseDouble(tokens[1]); | ||
|
|
||
| if (map.containsKey(itemID)) { | ||
| map.put(itemID, map.get(itemID) + score);// 矩阵乘法求和计算 | ||
| } else { | ||
| map.put(itemID, score); | ||
| } | ||
| } | ||
|
|
||
| Iterator<String> iter = map.keySet().iterator(); | ||
| while (iter.hasNext()) { | ||
| String itemID = iter.next(); | ||
| double score = map.get(itemID); | ||
| Text v = new Text(itemID + "," + score); | ||
| context.write(key, v); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public static void run(Map<String, String> path) throws IOException, InterruptedException, ClassNotFoundException { | ||
| JobConf conf = Recommend.config(); | ||
|
|
||
| String input = path.get("Step6Input"); | ||
| String output = path.get("Step6Output"); | ||
|
|
||
| HdfsDAO hdfs = new HdfsDAO(Recommend.HDFS, conf); | ||
| hdfs.rmr(output); | ||
|
|
||
| Job job = new Job(conf); | ||
| job.setJarByClass(Step4_Update2.class); | ||
|
|
||
| job.setOutputKeyClass(Text.class); | ||
| job.setOutputValueClass(Text.class); | ||
|
|
||
| job.setMapperClass(Step4_Update2.Step4_RecommendMapper.class); | ||
| job.setReducerClass(Step4_Update2.Step4_RecommendReducer.class); | ||
|
|
||
| job.setInputFormatClass(TextInputFormat.class); | ||
| job.setOutputFormatClass(TextOutputFormat.class); | ||
|
|
||
| FileInputFormat.setInputPaths(job, new Path(input)); | ||
| FileOutputFormat.setOutputPath(job, new Path(output)); | ||
|
|
||
| job.waitForCompletion(true); | ||
| } | ||
|
|
||
| } |