码迷，mamicode.com
首页 > 编程语言 > 详细

流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计

时间：2015-12-04 12:29:51      阅读：198      评论：0      收藏：0      [点我收藏+]

标签：

小知识点：

half：关机
yarn端口：8088
删除hdfs目录：hadoop fs -rm -r /wc/output

namenode两个状态都是standby原因：zookeeper没有比hdfs先启动

现在来做一个流量统计的例子：
首先数据是这样一张表：见附件
技术分享

统计：（代码）
技术分享
1，flowbean：
package cn.itcast.hadoop.mr.flowsum;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;

public class FlowBean implements WritableComparable<FlowBean> {

private String phoneNB;
private long up_flow;
private long d_flow;
private long s_flow;
//在反序列化时候反射机制需要调用空参数构造方法，所以显示定义了一个空参构造函数
public FlowBean() {}
//为了对象数据的初始化方便，加入一个带参数的构造函数
public FlowBean(String phoneNB, long up_flow, long d_flow) {
super();
this.phoneNB = phoneNB;
this.up_flow = up_flow;
this.d_flow = d_flow;
this.s_flow = up_flow+d_flow;
}

@Override
public String toString() {
return ""+up_flow +"\t" +d_flow + "\t"+ s_flow;
}

public String getPhoneNB() {
return phoneNB;
}

public void setPhoneNB(String phoneNB) {
this.phoneNB = phoneNB;
}

public long getUp_flow() {
return up_flow;
}

public void setUp_flow(long up_flow) {
this.up_flow = up_flow;
}

public long getD_flow() {
return d_flow;
}

public void setD_flow(long d_flow) {
this.d_flow = d_flow;
}

public long getS_flow() {
return s_flow;
}

public void setS_flow(long s_flow) {
this.s_flow = s_flow;
}

//从数据流中反序列化出对象的数据
// 从数据流中独处对象字段时候，必须跟序列化的顺序保持一致
@Override
public void readFields(DataInput in) throws IOException {
phoneNB = in.readUTF();
up_flow=in.readLong();
d_flow=in.readLong();
s_flow=in.readLong();
}

//将对象数据序列化到流中
@Override
public void write(DataOutput out) throws IOException {
 
out.writeUTF(phoneNB);
out.writeLong(up_flow);
out.writeLong(d_flow);
out.writeLong(s_flow);
}
//比较，在这里实现了排序
@Override
public int compareTo(FlowBean o) {
return s_flow>o.getS_flow()?-1:1;
}

}

2，flowsumMapper：
package cn.itcast.hadoop.mr.flowsum;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * @author yw.wang
 * FlowBean 是我们自定义的一种数据类型，要在hadoop的各个节点之间传输，所以应该遵循hadoop的序列化机制
 * 就必须实现hadoop的序列化接口
 *
 */
public class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
// 拿到日志中的一行数据，切分各个字段，抽取我们需要的字段：手机号，上行流量，下行流量，然后封装成kv类型发送出去，到reduce
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
//拿一行数据
String line = value.toString();
//切分成各个字段
String[] fields = StringUtils.split(line,"\t");
//拿到我们需要的字段
String phoneNB = fields[0];
long u_flow =Long.parseLong(fields[7]);
long d_flow =Long.parseLong(fields[8]);
//封装数据为kv类型并输出
context.write(new Text(phoneNB), new FlowBean(phoneNB,u_flow,d_flow));
}
}


3，flowsumreducer
package cn.itcast.hadoop.mr.flowsum;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class FlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean>{
//框架每传递一组数据<1237435262,{flowbean,flowbean,flowbean....}>
//reduce中的业务逻辑就是遍历values，然后进行累加求和再输出
@Override
protected void reduce(Text key, Iterable<FlowBean> values,Context context)
throws IOException, InterruptedException {
long up_flow_counter= 0;
long d_flow_counter=0;
for (FlowBean bean : values) {
up_flow_counter +=bean.getUp_flow();
d_flow_counter+=bean.getD_flow();
}
context.write(key, new FlowBean(key.toString(),up_flow_counter,d_flow_counter));
}

}

4，flowsumrunner：
package cn.itcast.hadoop.mr.flowsum;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

//这是job描述和提交类的规范写法
public class FlowSumRunner extends Configured implements Tool{

@Override
public int run(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(FlowSumRunner.class);
job.setMapperClass(FlowSumMapper.class);
job.setReducerClass(FlowSumReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(FlowBean.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
return job.waitForCompletion(true)?0:1;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new FlowSumRunner(), args);
System.exit(res);
}

}

打成jar包：
技术分享

在集群中使用命令：
hadoop  jar  /root/Documents/sum.jar   cn.itcast.hadoop.mr.flowsum.FlowSumRunner  /wc/data/  /wc/sumoutput

解释：
技术分享


排序：
技术分享
代码：
  1. package cn.itcast.hadoop.mr.flowsort;
  3. import java.io.IOException;
  5. import org.apache.commons.lang.StringUtils;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.fs.Path;
  8. import org.apache.hadoop.io.LongWritable;
  9. import org.apache.hadoop.io.NullWritable;
  10. import org.apache.hadoop.io.Text;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.Mapper;
  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;
  17. import cn.itcast.hadoop.mr.flowsum.FlowBean;
  19. public class SortMR {
  20. 	
  21. 	
  22. 	public static class SortMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable>{
  23. 		
  24. 		//拿到一行数据，切分出各字段，封装为一个flowbean，作为key输出
  25. 		@Override
  26. 		protected void map(LongWritable key, Text value,Context context)
  27. 				throws IOException, InterruptedException {
  29. 			String line = value.toString();
  30. 			
  31. 			String[] fields = StringUtils.split(line, "\t");
  32. 			
  33. 			String phoneNB = fields[0];
  34. 			long u_flow = Long.parseLong(fields[1]);
  35. 			long d_flow = Long.parseLong(fields[2]);
  36. 			
  37. 			context.write(new FlowBean(phoneNB, u_flow, d_flow), NullWritable.get());
  38. 			
  39. 		}
  40. 		
  41. 		
  42. 	}
  43. 	
  44. 	
  45. 	
  46. 	public static class SortReducer extends Reducer<FlowBean, NullWritable, Text, FlowBean>{
  47. 		
  48. 		@Override
  49. 		protected void reduce(FlowBean key, Iterable<NullWritable> values,Context context)
  50. 				throws IOException, InterruptedException {
  52. 			String phoneNB = key.getPhoneNB();
  53. 			context.write(new Text(phoneNB), key);
  54. 			
  55. 		}
  56. 		
  57. 	}
  58. 	
  59. 	
  60. 	
  61. 	public static void main(String[] args) throws Exception {
  62. 		
  63. 		Configuration conf = new Configuration();	
  64. 		Job job = Job.getInstance(conf);
  65. 		// main方法所在的类，此处表示自身的类
  66. 		job.setJarByClass(SortMR.class);
  67. 	//会代表map，reduce的output，如果不一样可以申明mapoutput类型，像下面的一样
  68. 		job.setMapperClass(SortMapper.class);
  69. 		job.setReducerClass(SortReducer.class);
  70. 	// mapoutput类型
  71. 		job.setMapOutputKeyClass(FlowBean.class);
  72. 		job.setMapOutputValueClass(NullWritable.class);
  73. 		
  74. 		job.setOutputKeyClass(Text.class);
  75. 		job.setOutputValueClass(FlowBean.class);

  77. 		//这两个参数正好是 hadoop jar 。。 最后两个参数
  78. 		FileInputFormat.setInputPaths(job, new Path(args[0]));
  79. 		FileOutputFormat.setOutputPath(job, new Path(args[1]));
  80. 		//标准输出
  81. 		System.exit(job.waitForCompletion(true)?0:1);
  82. 			
  83. 	}
  85. }
排序是针对统计的结果进行排序，故数据元是统计完成之后的00000success那个文件


分组：
技术分享

FlowSumArea ：
  1. package cn.itcast.hadoop.mr.areapartition;
  3. import java.io.IOException;
  5. import org.apache.commons.lang.StringUtils;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.fs.Path;
  8. import org.apache.hadoop.io.LongWritable;
  9. import org.apache.hadoop.io.Text;
  10. import org.apache.hadoop.mapreduce.Job;
  11. import org.apache.hadoop.mapreduce.Mapper;
  12. import org.apache.hadoop.mapreduce.Reducer;
  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  15. import org.apache.hadoop.metrics2.impl.ConfigBuilder;
  23. import cn.itcast.hadoop.mr.flowsum.FlowBean;
  25. /**
  26.  * 对流量原始日志进行流量统计，将不同省份的用户统计结果输出到不同文件
  27.  * 需要自定义改造两个机制
  28.  * 1，改造分区的逻辑，自定义一个partitioneer
  29.  * 2，自定义reduer task的并发任务数
  30.  */
  31. public class FlowSumArea {
  33. 	public static class FlowSumAreaMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
  34. 		
  35. 		@Override
  36. 		protected void map(LongWritable key, Text value,Context context)
  37. 				throws IOException, InterruptedException {
  38. 		
  39. 			//拿一行数据
  40. 			String line = value.toString();
  41. 			
  42. 			//切分成各个字段
  43. 			String[] fields = StringUtils.split(line,"\t");
  44. 			
  45. 			//拿到我们的字段
  46. 			String phoneNB = fields[1];
  47. 			long u_flow = Long.parseLong(fields[7]);
  48. 			long d_flow = Long.parseLong(fields[8]);
  49. 			
  50. 			//封装数据为kv并输出
  51. 			context.write(new Text(phoneNB), new FlowBean(phoneNB,u_flow,d_flow));
  52. 		}
  53. 		
  54. 		
  55. 	}
  56. 	
  57. 	public static class FlowSumAreaReducer extends Reducer<Text, FlowBean, Text, FlowBean>{
  58. 		
  59. 		@Override
  60. 		protected void reduce(Text key, Iterable<FlowBean> values,Context context)
  61. 				throws IOException, InterruptedException {
  62. 			
  63. 			long up_flow_counter = 0;
  64. 			long d_flow_counter = 0;
  65. 			
  66. 			for (FlowBean bean : values) {
  67. 				up_flow_counter +=bean.getUp_flow();
  68. 				d_flow_counter += bean.getD_flow();
  69. 				
  70. 				
  71. 			}
  72. 			context.write(key, new  FlowBean(key.toString(),up_flow_counter,d_flow_counter));
  73. 			
  74. 			
  75. 		}
  76. 	}
  77. 	
  78. 				public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
  79. 					
  80. 					Configuration conf = new Configuration();
  81. 					Job job = Job.getInstance(conf);
  82. 					
  83. 					job.setJarByClass(FlowSumArea.class);
  84. 					
  85. 					//job.setMapperClass(FlowSumAreaMapper.class);
  86. 					job.setMapperClass(FlowSumAreaMapper.class);
  87. 					job.setReducerClass(FlowSumAreaReducer.class);
  88. 					
  89. 					//设置我们自定义的分组逻辑定义
  90. 					job.setPartitionerClass(AreaPartitioner.class);
  91. 					
  92. 					
  93. 					job.setOutputKeyClass(Text.class);
  94. 					job.setOutputValueClass(FlowBean.class);
  95. 					
  96. 					//设置reduce的任务并发数，应该跟分组的数量保持一致
  97. 					job.setNumReduceTasks(6);
  98. //进程数如果大了，后面的文件为空，小了会出现错误，为1则没有分组
  99. 					
  100. 					
  101. 					FileInputFormat.setInputPaths(job, new Path(args[0]));
  102. 					FileOutputFormat.setOutputPath(job, new Path(args[1]));
  103. 					
  104. 					
  105. 					System.exit(job.waitForCompletion(true)?0:1);
  106. 					
  107. 				}
  108. 	
  109. }

AreaPartitioner ：
  1. package cn.itcast.hadoop.mr.areapartition;
  3. import java.util.HashMap;
  5. import org.apache.hadoop.mapreduce.Partitioner;
  7. public class AreaPartitioner<KEY, VALUE> extends Partitioner<KEY, VALUE> {
  9. 	private static HashMap<String,Integer> areaMap = new HashMap<>();
  10. 	
  11. 	static{
  12. 		areaMap.put("135", 0);
  13. 		areaMap.put("136", 1);
  14. 		areaMap.put("137", 2);
  15. 		areaMap.put("138", 3);
  16. 		areaMap.put("139", 4);
  17. 	
  18. 		
  19. 	}
  20. 	
  21. 	@Override
  22. 	public int getPartition(KEY key, VALUE value, int numPartitions) {
  23. 		//从key中拿到手机号，查询手机归属地字典，不同省份返回不同的组号
  24. 		int areaCoder = areaMap.get(key.toString().substring(0,3))==null?5:areaMap.get(key.toString().substring(0,3));
  25. 		
  26. 		return areaCoder;
  27. 	}
  29. 	
  30. }

运行：
hadoop jar /root/Documents/area.jar cn.itcast.hadoop.mr.areapartition.FlowSumArea /wc/data /wc/areasoutput
技术分享

至此，mapreduce的流量统计，分组，排序工作完成了
 
 





























































附件列表

     

    流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计

    标签：

    原文地址：http://www.cnblogs.com/xiaoxiao5ya/p/c23cd7c85104ae4bc5875c798d81fb2e.html
    踩
    (0)
    赞
    (0)
       
    举报
    评论 一句话评论（0）
    登录后才能评论！
    分享档案
    更多>
    2021年07月29日 (22)
    2021年07月28日 (40)
    2021年07月27日 (32)
    2021年07月26日 (79)
    2021年07月23日 (29)
    2021年07月22日 (30)
    2021年07月21日 (42)
    2021年07月20日 (16)
    2021年07月19日 (90)
    2021年07月16日 (35)
    周排行
    mamicode.com排行更多图片
    更多
    友情链接
    兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
    关于我们 - 联系我们 - 留言反馈
    © 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
    迷上了代码！