zoukankan      html  css  js  c++  java

  • mapreduce实现"浏览该商品的人大多数还浏览了"经典应用

    转自：http://blog.csdn.net/u011750989/article/details/12004065

    看着思路不错。是协同过滤的一种实现。

    注意：01是分隔符，16进制，ascii码里001，叫SOH(start of heading)。用vi编辑器Ctrl+v然后Ctrl+a。其他的分隔符有tab键（ascii 为9），空格（32）

    输入:

    日期 ...cookie id. ...商品id..

    xx xx xx

    输出:

    商品id 商品id列表(按优先级排序,用逗号分隔)

    xx xx

    比如:

    id1 id3,id0,id4,id2

    id2 id0,id5

    整个计算过程分为4步

    1、提取原始日志中的（日期,cookie id,商品id）信息，按天处理，最后输出数据格式

    商品id-0 商品id-1

    xx x x

    这一步做了次优化,商品id-0一定比商品id-1小，为了减少存储，在最后汇总数据转置下即可

    reduce做局部排序及排重

    2、基于上次的结果做汇总,按天计算

    商品id-0 商品id-1 关联值(关联值即同时访问这两个商品的用户数)

    xx x x xx

    3、汇总最近三个月数据,同时考虑时间衰减,时间越久关联值的贡献越低,最后输出两两商品的关联值（包括转置后)

    4、行列转换，生成最后要的推荐结果数据,按关联值排序生成

    第一个MR

    [java] view plaincopyprint?
    1. import java.io.IOException;
    2. import java.util.ArrayList;
    3. import org.apache.hadoop.conf.Configuration;
    4. import org.apache.hadoop.fs.FileSystem;
    5. import org.apache.hadoop.fs.Path;
    6. import org.apache.hadoop.io.LongWritable;
    7. import org.apache.hadoop.io.Text;
    8. import org.apache.hadoop.io.WritableComparable;
    9. import org.apache.hadoop.io.WritableComparator;
    10. import org.apache.hadoop.mapreduce.Job;
    11. import org.apache.hadoop.mapreduce.Mapper;
    12. import org.apache.hadoop.mapreduce.Partitioner;
    13. import org.apache.hadoop.mapreduce.Reducer;
    14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    16. import org.apache.hadoop.util.GenericOptionsParser;
    17. import org.apache.log4j.Logger;
    20. /*
    21. * 输入:原始数据,会有重复
    22. *日期 cookie 楼盘id
    23. *
    24. * 输出:
    25. * 日期 楼盘id1 楼盘id2 //楼盘id1一定小于楼盘id2 ,按日期 cookie进行分组
    26. *
    27. */
    29. public class HouseMergeAndSplit {
    31. public staticclass Partitioner1 extends Partitioner<TextPair, Text> {
    32. @Override
    33. public int getPartition(TextPair key, Text value,int numParititon) {
    34. return Math.abs((new Text(key.getFirst().toString()+key.getSecond().toString())).hashCode() *127) % numParititon;
    36. }
    37. }
    38. public staticclass Comp1 extends WritableComparator {
    39. public Comp1() {
    40. super(TextPair.class,true);
    41. }
    42. @SuppressWarnings("unchecked")
    43. public int compare(WritableComparable a, WritableComparable b) {
    44. TextPair t1 = (TextPair) a;
    45. TextPair t2 = (TextPair) b;
    46. int comp= t1.getFirst().compareTo(t2.getFirst());
    47. if (comp!=0)
    48. return comp;
    49. return t1.getSecond().compareTo(t2.getSecond());
    50. }
    51. }
    52. public staticclass TokenizerMapper
    53. extends Mapper<LongWritable, Text, TextPair, Text>{
    54. Text val=new Text("test");
    55. public void map(LongWritable key, Text value, Context context
    56. ) throws IOException, InterruptedException {
    57. String s[]=value.toString().split("01");
    58. TextPair tp=new TextPair(s[0],s[1],s[4]+s[3]);//thedate cookie city+houseid
    59. context.write(tp, val);
    60. }
    61. }
    63. public staticclass IntSumReducer
    64. extends Reducer<TextPair,Text,Text,Text> {
    65. private static String comparedColumn[] =new String[3];
    66. ArrayList<String> houselist= new ArrayList<String>();
    67. private static Text keyv =new Text();
    69. private static Text valuev =new Text();
    70. static Logger logger = Logger.getLogger(HouseMergeAndSplit.class.getName());
    72. public void reduce(TextPair key, Iterable<Text> values,
    73. Context context
    74. ) throws IOException, InterruptedException {
    76. houselist.clear();
    77. String thedate=key.getFirst().toString();
    78. String cookie=key.getSecond().toString();
    80. for (int i=0;i<3;i++)
    81. comparedColumn[i]="";
    83. //first+second为分组键,每次不同重新调用reduce函数
    84. for (Text val:values)
    85. {
    87. if (thedate.equals(comparedColumn[0]) && cookie.equals(comparedColumn[1])&& !key.getThree().toString().equals(comparedColumn[2]))
    88. {
    89. // context.write(new Text(key.getFirst()+" "+key.getSecond().toString()), new Text(key.getThree().toString()+" first"+ " "+comparedColumn[0]+" "+comparedColumn[1]+" "+comparedColumn[2]));
    90. houselist.add(key.getThree().toString());
    92. comparedColumn[0]=key.getFirst().toString();
    93. comparedColumn[1]=key.getSecond().toString();
    94. comparedColumn[2]=key.getThree().toString();
    96. }
    98. if (!thedate.equals(comparedColumn[0])||!cookie.equals(comparedColumn[1]))
    100. {
    102. // context.write(new Text(key.getFirst()+" "+key.getSecond().toString()), new Text(key.getThree().toString()+" second"+ " "+comparedColumn[0]+" "+comparedColumn[1]+" "+comparedColumn[2]));
    103. houselist.add(key.getThree().toString());
    104. comparedColumn[0]=key.getFirst().toString();
    105. comparedColumn[1]=key.getSecond().toString();
    106. comparedColumn[2]=key.getThree().toString();
    108. }
    112. }
    116. keyv.set(comparedColumn[0]); //日期
    117. //valuev.set(houselist.toString());
    118. //logger.info(houselist.toString());
    119. //context.write(keyv,valuev);
    122. for (int i=0;i<houselist.size()-1;i++)
    123. {
    124. for (int j=i+1;j<houselist.size();j++)
    125. { valuev.set(houselist.get(i)+" "+houselist.get(j));//关联的楼盘
    126. context.write(keyv,valuev);
    127. }
    128. }
    130. }
    131. }
    133. public staticvoid main(String[] args) throws Exception {
    134. Configuration conf = new Configuration();
    135. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    136. if (otherArgs.length != 2) {
    137. System.err.println("Usage: wordcount <in> <out>");
    138. System.exit(2);
    139. }
    141. FileSystem fstm = FileSystem.get(conf);
    142. Path outDir = new Path(otherArgs[1]);
    143. fstm.delete(outDir, true);
    145. conf.set("mapred.textoutputformat.separator"," "); //reduce输出时key value中间的分隔符
    146. Job job = new Job(conf, "HouseMergeAndSplit");
    147. job.setNumReduceTasks(4);
    148. job.setJarByClass(HouseMergeAndSplit.class);
    149. job.setMapperClass(TokenizerMapper.class);
    151. job.setMapOutputKeyClass(TextPair.class);
    152. job.setMapOutputValueClass(Text.class);
    153. // 设置partition
    154. job.setPartitionerClass(Partitioner1.class);
    155. // 在分区之后按照指定的条件分组
    156. job.setGroupingComparatorClass(Comp1.class);
    157. // 设置reduce
    158. // 设置reduce的输出
    159. job.setReducerClass(IntSumReducer.class);
    160. job.setOutputKeyClass(Text.class);
    161. job.setOutputValueClass(Text.class);
    162. //job.setNumReduceTasks(18);
    163. FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    164. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    165. System.exit(job.waitForCompletion(true) ?0 : 1);
    166. }
    167. }

    TextPair

    [java] view plaincopyprint?
    1. import java.io.DataInput;
    2. import java.io.DataOutput;
    3. import java.io.IOException;
    5. import org.apache.hadoop.io.Text;
    6. import org.apache.hadoop.io.WritableComparable;
    8. public class TextPairimplements WritableComparable<TextPair> {
    9. private Text first;
    10. private Text second;
    11. private Text three;
    12. public TextPair() {
    13. set(new Text(), new Text(),new Text());
    14. }
    15. public TextPair(String first, String second,String three) {
    16. set(new Text(first), new Text(second),new Text(three));
    17. }
    18. public TextPair(Text first, Text second,Text Three) {
    19. set(first, second,three);
    20. }
    21. public void set(Text first, Text second,Text three) {
    22. this.first = first;
    23. this.second = second;
    24. this.three=three;
    25. }
    26. public Text getFirst() {
    27. return first;
    28. }
    29. public Text getSecond() {
    30. return second;
    31. }
    32. public Text getThree() {
    33. return three;
    34. }
    35. public void write(DataOutput out)throws IOException {
    36. first.write(out);
    37. second.write(out);
    38. three.write(out);
    39. }
    40. public void readFields(DataInput in)throws IOException {
    41. first.readFields(in);
    42. second.readFields(in);
    43. three.readFields(in);
    44. }
    45. public int compareTo(TextPair tp) {
    46. int cmp = first.compareTo(tp.first);
    47. if (cmp != 0) {
    48. return cmp;
    49. }
    50. cmp= second.compareTo(tp.second);
    51. if (cmp != 0) {
    52. return cmp;
    53. }
    54. return three.compareTo(tp.three);
    55. }
    56. }


    TextPairSecond

    [java] view plaincopyprint?
    1. import java.io.DataInput;
    2. import java.io.DataOutput;
    3. import java.io.IOException;
    5. import org.apache.hadoop.io.FloatWritable;
    6. import org.apache.hadoop.io.Text;
    7. import org.apache.hadoop.io.WritableComparable;
    9. public class TextPairSecondimplements WritableComparable<TextPairSecond> {
    10. private Text first;
    11. private FloatWritable second;
    12. public TextPairSecond() {
    13. set(new Text(), new FloatWritable());
    14. }
    15. public TextPairSecond(String first,float second) {
    16. set(new Text(first), new FloatWritable(second));
    17. }
    18. public TextPairSecond(Text first, FloatWritable second) {
    19. set(first, second);
    20. }
    21. public void set(Text first, FloatWritable second) {
    22. this.first = first;
    23. this.second = second;
    24. }
    25. public Text getFirst() {
    26. return first;
    27. }
    28. public FloatWritable getSecond() {
    29. return second;
    30. }
    31. public void write(DataOutput out)throws IOException {
    32. first.write(out);
    33. second.write(out);
    34. }
    35. public void readFields(DataInput in)throws IOException {
    36. first.readFields(in);
    37. second.readFields(in);
    38. }
    39. public int compareTo(TextPairSecond tp) {
    40. int cmp = first.compareTo(tp.first);
    41. if (cmp != 0) {
    42. return cmp;
    43. }
    44. return second.compareTo(tp.second);
    45. }
    47. }

    第二个MR

    [java] view plaincopyprint?
    1. import java.io.IOException;
    2. import java.text.SimpleDateFormat;
    3. import java.util.ArrayList;
    4. import java.util.Date;
    6. import org.apache.hadoop.conf.Configuration;
    7. import org.apache.hadoop.fs.FileSystem;
    8. import org.apache.hadoop.fs.Path;
    9. import org.apache.hadoop.io.IntWritable;
    10. import org.apache.hadoop.io.LongWritable;
    11. import org.apache.hadoop.io.NullWritable;
    12. import org.apache.hadoop.io.Text;
    13. import org.apache.hadoop.io.WritableComparable;
    14. import org.apache.hadoop.io.WritableComparator;
    15. import org.apache.hadoop.mapred.OutputCollector;
    16. import org.apache.hadoop.mapreduce.Job;
    17. import org.apache.hadoop.mapreduce.Mapper;
    18. import org.apache.hadoop.mapreduce.Partitioner;
    19. import org.apache.hadoop.mapreduce.Reducer;
    21. import org.apache.hadoop.mapreduce.Mapper.Context;
    22. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    23. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    24. import org.apache.hadoop.util.GenericOptionsParser;
    25. import org.apache.log4j.Logger;
    28. /*
    29. * 统计楼盘之间共同出现的次数
    30. * 输入:
    31. * 日期 楼盘1 楼盘2
    32. *
    33. * 输出：
    34. * 日期 楼盘1 楼盘2 共同出现的次数
    35. *
    36. */
    38. public class HouseCount {
    41. public staticclass TokenizerMapper
    42. extends Mapper<LongWritable, Text, Text, IntWritable>{
    45. IntWritable iw=new IntWritable(1);
    46. public void map(LongWritable key, Text value, Context context
    47. ) throws IOException, InterruptedException {
    50. context.write(value, iw);
    51. }
    52. }
    54. public staticclass IntSumReducer
    55. extends Reducer<Text,IntWritable,Text,IntWritable> {
    57. IntWritable result=new IntWritable();
    58. public void reduce(Text key, Iterable<IntWritable> values,
    59. Context context
    60. ) throws IOException, InterruptedException {
    62. int sum=0;
    63. for (IntWritable iw:values)
    64. {
    65. sum+=iw.get();
    66. }
    67. result.set(sum);
    68. context.write(key, result) ;
    70. }
    71. }
    73. public staticvoid main(String[] args) throws Exception {
    74. Configuration conf = new Configuration();
    75. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    76. if (otherArgs.length != 2) {
    77. System.err.println("Usage: wordcount <in> <out>");
    78. System.exit(2);
    79. }
    81. FileSystem fstm = FileSystem.get(conf);
    82. Path outDir = new Path(otherArgs[1]);
    83. fstm.delete(outDir, true);
    85. conf.set("mapred.textoutputformat.separator"," "); //reduce输出时key value中间的分隔符
    86. Job job = new Job(conf, "HouseCount");
    87. job.setNumReduceTasks(2);
    88. job.setJarByClass(HouseCount.class);
    89. job.setMapperClass(TokenizerMapper.class);
    91. job.setMapOutputKeyClass(Text.class);
    92. job.setMapOutputValueClass(IntWritable.class);
    94. // 设置reduce
    95. // 设置reduce的输出
    96. job.setReducerClass(IntSumReducer.class);
    97. job.setOutputKeyClass(Text.class);
    98. job.setOutputValueClass(IntWritable.class);
    99. //job.setNumReduceTasks(18);
    100. FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    101. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    102. System.exit(job.waitForCompletion(true) ?0 : 1);
    103. }
    104. }


    第三个MR

    [java] view plaincopyprint?
    1. import java.io.IOException;
    2. import java.text.ParseException;
    3. import java.text.SimpleDateFormat;
    4. import java.util.ArrayList;
    5. import java.util.Calendar;
    6. import java.util.Date;
    8. import org.apache.hadoop.conf.Configuration;
    9. import org.apache.hadoop.fs.FileSystem;
    10. import org.apache.hadoop.fs.Path;
    11. import org.apache.hadoop.io.FloatWritable;
    12. import org.apache.hadoop.io.IntWritable;
    13. import org.apache.hadoop.io.LongWritable;
    14. import org.apache.hadoop.io.NullWritable;
    15. import org.apache.hadoop.io.Text;
    16. import org.apache.hadoop.io.WritableComparable;
    17. import org.apache.hadoop.io.WritableComparator;
    18. import org.apache.hadoop.mapred.OutputCollector;
    19. import org.apache.hadoop.mapreduce.Job;
    20. import org.apache.hadoop.mapreduce.Mapper;
    21. import org.apache.hadoop.mapreduce.Partitioner;
    22. import org.apache.hadoop.mapreduce.Reducer;
    24. import org.apache.hadoop.mapreduce.Mapper.Context;
    25. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    26. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    27. import org.apache.hadoop.util.GenericOptionsParser;
    28. import org.apache.log4j.Logger;
    31. /*
    32. * 汇总近三个月统计楼盘之间共同出现的次数,考虑衰减系数， 并最后a b 转成 b a输出一次
    33. * 输入:
    34. * 日期 楼盘1 楼盘2 共同出现的次数
    35. *
    36. * 输出
    37. * 楼盘1 楼盘2 共同出现的次数（考虑了衰减系数，每天的衰减系数不一样)
    38. *
    39. */
    41. public class HouseCountHz {
    44. public staticclass HouseCountHzMapper
    45. extends Mapper<LongWritable, Text, Text, FloatWritable>{
    47. Text keyv=new Text();
    49. FloatWritable valuev=new FloatWritable();
    50. public void map(LongWritable key, Text value, Context context
    51. ) throws IOException, InterruptedException {
    53. String[] s=value.toString().split(" ");
    54. keyv.set(s[1]+" "+s[2]);//楼盘1,楼盘2
    55. Calendar date1=Calendar.getInstance();
    56. Calendar d2=Calendar.getInstance();
    58. Date b = null;
    59. SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd");
    60. try {
    61. b=sdf.parse(s[0]);
    62. } catch (ParseException e) {
    63. e.printStackTrace();
    64. }
    65. d2.setTime(b);
    66. long n=date1.getTimeInMillis();
    67. long birth=d2.getTimeInMillis();
    68. long sss=n-birth;
    69. int day=(int)((sss)/(3600*24*1000));//该条记录的日期与当前日期的日期差
    70. float factor=1/(1+(float)(day-1)/10);//衰减系数
    71. valuev.set(Float.parseFloat(s[3])*factor);
    73. context.write(keyv, valuev);
    74. }
    75. }
    77. public staticclass HouseCountHzReducer
    78. extends Reducer<Text,FloatWritable,Text,FloatWritable> {
    80. FloatWritable result=new FloatWritable();
    81. Text keyreverse=new Text();
    82. public void reduce(Text key, Iterable<FloatWritable> values,
    83. Context context
    84. ) throws IOException, InterruptedException {
    86. float sum=0;
    87. for (FloatWritable iw:values)
    88. {
    89. sum+=iw.get();
    90. }
    91. result.set(sum);
    92. String[] keys=key.toString().split(" ");
    93. keyreverse.set(keys[1]+" "+keys[0]);
    94. context.write(key, result) ;
    95. context.write(keyreverse, result) ;
    97. }
    98. }
    100. public staticvoid main(String[] args) throws Exception {
    101. Configuration conf = new Configuration();
    102. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    103. if (otherArgs.length !=2) {
    104. System.err.println("Usage: wordcount <in> <out>");
    105. System.exit(2);
    106. }
    108. FileSystem fstm = FileSystem.get(conf);
    109. Path outDir = new Path(otherArgs[1]);
    110. fstm.delete(outDir, true);
    112. conf.set("mapred.textoutputformat.separator"," "); //reduce输出时key value中间的分隔符
    113. Job job = new Job(conf,"HouseCountHz");
    114. job.setNumReduceTasks(2);
    115. job.setJarByClass(HouseCountHz.class);
    116. job.setMapperClass(HouseCountHzMapper.class);
    118. job.setMapOutputKeyClass(Text.class);
    119. job.setMapOutputValueClass(FloatWritable.class);
    121. // 设置reduce
    122. // 设置reduce的输出
    123. job.setReducerClass(HouseCountHzReducer.class);
    124. job.setOutputKeyClass(Text.class);
    125. job.setOutputValueClass(FloatWritable.class);
    126. //job.setNumReduceTasks(18);
    127. FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    128. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    129. System.exit(job.waitForCompletion(true) ?0 : 1);
    130. }
    131. }


    第四个MR

    [java] view plaincopyprint?
    1. import java.io.IOException;
    2. import java.util.Iterator;
    5. import org.apache.hadoop.conf.Configuration;
    6. import org.apache.hadoop.fs.FileSystem;
    7. import org.apache.hadoop.fs.Path;
    8. import org.apache.hadoop.io.FloatWritable;
    10. import org.apache.hadoop.io.LongWritable;
    12. import org.apache.hadoop.io.Text;
    13. import org.apache.hadoop.io.WritableComparable;
    14. import org.apache.hadoop.io.WritableComparator;
    16. import org.apache.hadoop.mapreduce.Job;
    17. import org.apache.hadoop.mapreduce.Mapper;
    18. import org.apache.hadoop.mapreduce.Partitioner;
    19. import org.apache.hadoop.mapreduce.Reducer;
    21. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    22. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    23. import org.apache.hadoop.util.GenericOptionsParser;
    27. /*
    28. * 输入数据:
    29. * 楼盘1 楼盘2 共同出现的次数
    30. *
    31. * 输出数据
    32. * 楼盘1 楼盘2,楼盘3,楼盘4 (按次数排序)
    33. */
    35. public class HouseRowToCol {
    37. public staticclass Partitioner1 extends Partitioner<TextPairSecond, Text> {
    38. @Override
    39. //分区
    40. public int getPartition(TextPairSecond key, Text value,int numParititon) {
    41. return Math.abs((new Text(key.getFirst().toString()+key.getSecond().toString())).hashCode() *127) % numParititon;
    43. }
    44. }
    45. //分组
    46. public staticclass Comp1 extends WritableComparator {
    47. public Comp1() {
    48. super(TextPairSecond.class,true);
    49. }
    50. @SuppressWarnings("unchecked")
    51. public int compare(WritableComparable a, WritableComparable b) {
    52. TextPairSecond t1 = (TextPairSecond) a;
    53. TextPairSecond t2 = (TextPairSecond) b;
    54. return t1.getFirst().compareTo(t2.getFirst());
    56. }
    57. }
    59. //排序
    60. public staticclass KeyComp extends WritableComparator {
    61. public KeyComp() {
    62. super(TextPairSecond.class,true);
    63. }
    64. @SuppressWarnings("unchecked")
    65. public int compare(WritableComparable a, WritableComparable b) {
    66. TextPairSecond t1 = (TextPairSecond) a;
    67. TextPairSecond t2 = (TextPairSecond) b;
    68. int comp= t1.getFirst().compareTo(t2.getFirst());
    69. if (comp!=0)
    70. return comp;
    71. return -t1.getSecond().compareTo(t2.getSecond());
    72. }
    73. }
    74. public staticclass HouseRowToColMapper
    75. extends Mapper<LongWritable, Text, TextPairSecond, Text>{
    77. Text houseid1=new Text();
    78. Text houseid2=new Text();
    79. FloatWritable weight=new FloatWritable();
    80. public void map(LongWritable key, Text value, Context context
    81. ) throws IOException, InterruptedException {
    83. String s[]=value.toString().split(" ");
    85. weight.set(Float.parseFloat(s[2]));
    86. houseid1.set(s[0]);
    87. houseid2.set(s[1]);
    88. TextPairSecond tp=new TextPairSecond(houseid1,weight);
    89. context.write(tp, houseid2);
    90. }
    91. }
    93. public staticclass HouseRowToColReducer
    94. extends Reducer<TextPairSecond,Text,Text,Text> {
    96. Text valuev=new Text();
    97. public void reduce(TextPairSecond key, Iterable<Text> values,
    98. Context context
    99. ) throws IOException, InterruptedException {
    100. Text keyv=key.getFirst();
    101. Iterator<Text> it=values.iterator();
    102. StringBuilder sb=new StringBuilder(it.next().toString());
    103. while(it.hasNext())
    104. {
    105. sb.append(","+it.next().toString());
    106. }
    107. valuev.set(sb.toString());
    108. context.write(keyv, valuev);
    112. }
    113. }
    115. public staticvoid main(String[] args) throws Exception {
    116. Configuration conf = new Configuration();
    117. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    118. if (otherArgs.length != 2) {
    119. System.err.println("Usage: wordcount <in> <out>");
    120. System.exit(2);
    121. }
    123. FileSystem fstm = FileSystem.get(conf);
    124. Path outDir = new Path(otherArgs[1]);
    125. fstm.delete(outDir, true);
    127. conf.set("mapred.textoutputformat.separator"," "); //reduce输出时key value中间的分隔符
    128. Job job = new Job(conf, "HouseRowToCol");
    129. job.setNumReduceTasks(4);
    130. job.setJarByClass(HouseRowToCol.class);
    131. job.setMapperClass(HouseRowToColMapper.class);
    133. job.setMapOutputKeyClass(TextPairSecond.class);
    134. job.setMapOutputValueClass(Text.class);
    135. // 设置partition
    136. job.setPartitionerClass(Partitioner1.class);
    137. // 在分区之后按照指定的条件分组
    138. job.setGroupingComparatorClass(Comp1.class);
    139. job.setSortComparatorClass(KeyComp.class);
    140. // 设置reduce
    141. // 设置reduce的输出
    142. job.setReducerClass(HouseRowToColReducer.class);
    143. job.setOutputKeyClass(Text.class);
    144. job.setOutputValueClass(Text.class);
    145. //job.setNumReduceTasks(18);
    146. FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    147. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    148. System.exit(job.waitForCompletion(true) ?0 : 1);
    149. }
    150. }
  • 相关阅读:
    接口优先于抽象类
    接口优先于抽象类
    PHP的isset()函数
    mysql linux安装
    为mediawiki用户重置密码
    explode在PHP中的用法
    ERROR 1045 (28000): Access denied for user 'ODBC'@'localhost' (using password: N)
    http://blog.csdn.net/spidertiger/archive/2006/09/11/1206512.aspx
    最新linux+vsftpd配置详解
    widows下安装mediawiki
  • 原文地址：https://www.cnblogs.com/cl1024cl/p/6205450.html
Copyright © 2011-2022 走看看