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

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

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含10180字，纯文字阅读大概需要15分钟。

内容图文

小知识点：

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. publicclassSortMR{
  22. publicstaticclassSortMapperextendsMapper<LongWritable,Text,FlowBean,NullWritable>{
  24. //拿到一行数据，切分出各字段，封装为一个flowbean，作为key输出
  25. @Override
  26. protectedvoid map(LongWritable key,Text value,Context context)
  27. throwsIOException,InterruptedException{
  29. String line = value.toString();
  31. String[] fields =StringUtils.split(line,"\t");
  33. String phoneNB = fields[0];
  34. long u_flow =Long.parseLong(fields[1]);
  35. long d_flow =Long.parseLong(fields[2]);
  37. 			context.write(newFlowBean(phoneNB, u_flow, d_flow),NullWritable.get());
  39. }
  42. }
  46. publicstaticclassSortReducerextendsReducer<FlowBean,NullWritable,Text,FlowBean>{
  48. @Override
  49. protectedvoid reduce(FlowBean key,Iterable<NullWritable> values,Context context)
  50. throwsIOException,InterruptedException{
  52. String phoneNB = key.getPhoneNB();
  53. 			context.write(newText(phoneNB), key);
  55. }
  57. }
  61. publicstaticvoid main(String[] args)throwsException{
  63. Configuration conf =newConfiguration();
  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);
  74. 		job.setOutputKeyClass(Text.class);
  75. 		job.setOutputValueClass(FlowBean.class);

  77. 		//这两个参数正好是 hadoop jar 。。 最后两个参数
  78. FileInputFormat.setInputPaths(job,newPath(args[0]));
  79. FileOutputFormat.setOutputPath(job,newPath(args[1]));
  80. 		//标准输出
  81. System.exit(job.waitForCompletion(true)?0:1);
  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. publicclassFlowSumArea{
  33. publicstaticclassFlowSumAreaMapperextendsMapper<LongWritable,Text,Text,FlowBean>{
  35. @Override
  36. protectedvoid map(LongWritable key,Text value,Context context)
  37. throwsIOException,InterruptedException{
  39. //拿一行数据
  40. String line = value.toString();
  42. //切分成各个字段
  43. String[] fields =StringUtils.split(line,"\t");
  45. //拿到我们的字段
  46. String phoneNB = fields[1];
  47. long u_flow =Long.parseLong(fields[7]);
  48. long d_flow =Long.parseLong(fields[8]);
  50. //封装数据为kv并输出
  51. 			context.write(newText(phoneNB),newFlowBean(phoneNB,u_flow,d_flow));
  52. }
  55. }
  57. publicstaticclassFlowSumAreaReducerextendsReducer<Text,FlowBean,Text,FlowBean>{
  59. @Override
  60. protectedvoid reduce(Text key,Iterable<FlowBean> values,Context context)
  61. throwsIOException,InterruptedException{
  63. long up_flow_counter =0;
  64. long d_flow_counter =0;
  66. for(FlowBean bean : values){
  67. 				up_flow_counter +=bean.getUp_flow();
  68. 				d_flow_counter += bean.getD_flow();
  71. }
  72. 			context.write(key,newFlowBean(key.toString(),up_flow_counter,d_flow_counter));
  75. }
  76. }
  78. publicstaticvoid main(String[] args)throwsIOException,ClassNotFoundException,InterruptedException{
  80. Configuration conf =newConfiguration();
  81. Job job =Job.getInstance(conf);
  83. 					job.setJarByClass(FlowSumArea.class);
  85. //job.setMapperClass(FlowSumAreaMapper.class);
  86. 					job.setMapperClass(FlowSumAreaMapper.class);
  87. 					job.setReducerClass(FlowSumAreaReducer.class);
  89. //设置我们自定义的分组逻辑定义
  90. 					job.setPartitionerClass(AreaPartitioner.class);
  93. 					job.setOutputKeyClass(Text.class);
  94. 					job.setOutputValueClass(FlowBean.class);
  96. //设置reduce的任务并发数，应该跟分组的数量保持一致
  97. 					job.setNumReduceTasks(6);
  98. //进程数如果大了，后面的文件为空，小了会出现错误，为1则没有分组
  101. FileInputFormat.setInputPaths(job,newPath(args[0]));
  102. FileOutputFormat.setOutputPath(job,newPath(args[1]));
  105. System.exit(job.waitForCompletion(true)?0:1);
  107. }
  109. }

AreaPartitioner ：
  1. package cn.itcast.hadoop.mr.areapartition;
  3. import java.util.HashMap;
  5. import org.apache.hadoop.mapreduce.Partitioner;
  7. publicclassAreaPartitioner<KEY, VALUE>extendsPartitioner<KEY, VALUE>{
  9. privatestaticHashMap<String,Integer> areaMap =newHashMap<>();
  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);
  19. }
  21. @Override
  22. publicint 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));
  26. return areaCoder;
  27. }
  30. }

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

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



























































附件列表

     

    原文：http://www.cnblogs.com/xiaoxiao5ya/p/c23cd7c85104ae4bc5875c798d81fb2e.html

    内容总结

    以上是互联网集市为您收集整理的流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计全部内容，希望文章能够帮你解决流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

    内容备注

    版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

    内容手机端

    扫描二维码推送至手机访问。

    本文链接：https://qyyshop.com/info/1112404.html

    来源：【匿名】