首页 > 代码库 > Hadoop实战-MapReduce之分组(group-by)统计(七)
Hadoop实战-MapReduce之分组(group-by)统计(七)
1、数据准备
使用MapReduce计算age.txt中年龄最大、最小、均值
name,min,max,count
Mike,35,20,1
Mike,5,15,2
Mike,20,13,1
Steven,40,20,10
Ken,28,68,1
Ken,14,198,10
Cindy,32,31,100
2、预期结果
Mike 5 20 4
Steven,40,20,10
Ken 14 198 11
Cindy,32,31,100
3、需要加入自定义输出类型MinMaxCountTuple
import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import org.apache.hadoop.io.Writable; public class MinMaxCountTuple implements Writable { private int min; private int max; private int count; public int getMin() { return min; } public void setMin(int min) { this.min = min; } public int getMax() { return max; } public void setMax(int max) { this.max = max; } public int getCount() { return count; } public void setCount(int count) { this.count = count; } public void readFields(DataInput in) throws IOException { min = in.readInt(); max = in.readInt(); count = in.readInt(); } public void write(DataOutput out) throws IOException { out.writeInt(min); out.writeInt(max); out.writeInt(count); } @Override public String toString() { return min + "\t" + max + "\t" + count; } }
4、MapReduce编程
import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; 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.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class Age { public static class AgeMap extends Mapper<Object, Text, Text, MinMaxCountTuple> { private Text userName = new Text(); private MinMaxCountTuple outTuple = new MinMaxCountTuple(); @Override public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { String content = itr.nextToken(); String[] splits = content.split(","); String name = splits[0]; int min = Integer.valueOf(splits[1]); int max = Integer.valueOf(splits[2]); int count = Integer.valueOf(splits[3]); outTuple.setMin(min); outTuple.setMax(max); outTuple.setCount(count); userName.set(name); context.write(userName, outTuple); } } } public static class AgeReduce extends Reducer<Text, MinMaxCountTuple, Text, MinMaxCountTuple> { private MinMaxCountTuple result = new MinMaxCountTuple(); public void reduce(Text key, Iterable<MinMaxCountTuple> values, Context context) throws IOException, InterruptedException { int sum = 0; result.setMax(0); result.setMin(Integer.MAX_VALUE); for (MinMaxCountTuple tmp : values) { if (tmp.getMin() < result.getMin()) { result.setMin(tmp.getMin()); } if (tmp.getMax() > result.getMax()) { result.setMax(tmp.getMax()); } sum += tmp.getCount(); } result.setCount(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args) .getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: MinMaxCountDriver <in> <out>"); System.exit(2); } Job job = new Job(conf, "StackOverflow Comment Date Min Max Count"); job.setJarByClass(Age.class); job.setMapperClass(AgeMap.class); job.setCombinerClass(AgeReduce.class); job.setReducerClass(AgeReduce.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(MinMaxCountTuple.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
Hadoop实战-MapReduce之分组(group-by)统计(七)
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。