您的位置: 首页  >  IT文章  >  流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计

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

分类: IT文章 • 2023-12-22 16:06:18
小知识点：

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

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

现在来做一个流量统计的例子：
首先数据是这样一张表：见附件
流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计

统计：（代码）
流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计
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 +" " +d_flow + " "+ 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," ");
//拿到我们需要的字段
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包：
流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计

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

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


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

  1. package cn.itcast.hadoop.mr.flowsort;
    2. 

    3. 

  3. import java.io.IOException;
    4. 

    5. 

  5. import org.apache.commons.lang.StringUtils;
    6. 

  6. import org.apache.hadoop.conf.Configuration;
    7. 

  7. import org.apache.hadoop.fs.Path;
    8. 

  8. import org.apache.hadoop.io.LongWritable;
    9. 

  9. import org.apache.hadoop.io.NullWritable;
    10. 

  10. import org.apache.hadoop.io.Text;
    11. 

  11. import org.apache.hadoop.mapreduce.Job;
    12. 

  12. import org.apache.hadoop.mapreduce.Mapper;
    13. 

  13. import org.apache.hadoop.mapreduce.Reducer;
    14. 

  14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    15. 

  15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    16. 

    17. 

  17. import cn.itcast.hadoop.mr.flowsum.FlowBean;
    18. 

    19. 

  19. public class SortMR {
    20. 

  20. 	
    21. 

  21. 	
    22. 

  22. 	public static class SortMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable>{
    23. 

  23. 		
    24. 

  24. 		//拿到一行数据，切分出各字段，封装为一个flowbean，作为key输出
    25. 

  25. 		@Override
    26. 

  26. 		protected void map(LongWritable key, Text value,Context context)
    27. 

  27. 				throws IOException, InterruptedException {
    28. 

    29. 

  29. 			String line = value.toString();
    30. 

  30. 			
    31. 

  31. 			String[] fields = StringUtils.split(line, "	");
    32. 

  32. 			
    33. 

  33. 			String phoneNB = fields[0];
    34. 

  34. 			long u_flow = Long.parseLong(fields[1]);
    35. 

  35. 			long d_flow = Long.parseLong(fields[2]);
    36. 

  36. 			
    37. 

  37. 			context.write(new FlowBean(phoneNB, u_flow, d_flow), NullWritable.get());
    38. 

  38. 			
    39. 

  39. 		}
    40. 

  40. 		
    41. 

  41. 		
    42. 

  42. 	}
    43. 

  43. 	
    44. 

  44. 	
    45. 

  45. 	
    46. 

  46. 	public static class SortReducer extends Reducer<FlowBean, NullWritable, Text, FlowBean>{
    47. 

  47. 		
    48. 

  48. 		@Override
    49. 

  49. 		protected void reduce(FlowBean key, Iterable<NullWritable> values,Context context)
    50. 

  50. 				throws IOException, InterruptedException {
    51. 

    52. 

  52. 			String phoneNB = key.getPhoneNB();
    53. 

  53. 			context.write(new Text(phoneNB), key);
    54. 

  54. 			
    55. 

  55. 		}
    56. 

  56. 		
    57. 

  57. 	}
    58. 

  58. 	
    59. 

  59. 	
    60. 

  60. 	
    61. 

  61. 	public static void main(String[] args) throws Exception {
    62. 

  62. 		
    63. 

  63. 		Configuration conf = new Configuration();	
    64. 

  64. 		Job job = Job.getInstance(conf);
    65. 

  65. 		// main方法所在的类，此处表示自身的类
    66. 

  66. 		job.setJarByClass(SortMR.class);
    67. 

  67. 
	//会代表map，reduce的output，如果不一样可以申明mapoutput类型，像下面的一样

    68. 

  68. 		job.setMapperClass(SortMapper.class);
    69. 

  69. 		job.setReducerClass(SortReducer.class);
    70. 

  70. 	// mapoutput类型
    71. 

  71. 		job.setMapOutputKeyClass(FlowBean.class);
    72. 

  72. 		job.setMapOutputValueClass(NullWritable.class);
    73. 

  73. 		
    74. 

  74. 		job.setOutputKeyClass(Text.class);
    75. 

  75. 		job.setOutputValueClass(FlowBean.class);
    76. 


    77. 

  77. 		//这两个参数正好是 hadoop jar 。。 最后两个参数
    78. 

  78. 		FileInputFormat.setInputPaths(job, new Path(args[0]));
    79. 

  79. 		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    80. 

  80. 		//标准输出
    81. 

  81. 		System.exit(job.waitForCompletion(true)?0:1);
    82. 

  82. 
			

    83. 

  83. 	}
    84. 

    85. 

  85. }
    86.
排序是针对统计的结果进行排序，故数据元是统计完成之后的00000success那个文件


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

FlowSumArea ：
    

  1. package cn.itcast.hadoop.mr.areapartition;
    2. 

    3. 

  3. import java.io.IOException;
    4. 

    5. 

  5. import org.apache.commons.lang.StringUtils;
    6. 

  6. import org.apache.hadoop.conf.Configuration;
    7. 

  7. import org.apache.hadoop.fs.Path;
    8. 

  8. import org.apache.hadoop.io.LongWritable;
    9. 

  9. import org.apache.hadoop.io.Text;
    10. 

  10. import org.apache.hadoop.mapreduce.Job;
    11. 

  11. import org.apache.hadoop.mapreduce.Mapper;
    12. 

  12. import org.apache.hadoop.mapreduce.Reducer;
    13. 

  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    14. 

  14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    15. 

  15. import org.apache.hadoop.metrics2.impl.ConfigBuilder;
    16. 

    17. 

    18. 

    19. 

    20. 

    21. 

    22. 

    23. 

  23. import cn.itcast.hadoop.mr.flowsum.FlowBean;
    24. 

    25. 

  25. /**
    26. 

  26.  * 对流量原始日志进行流量统计，将不同省份的用户统计结果输出到不同文件
    27. 

  27.  * 需要自定义改造两个机制
    28. 

  28.  * 1，改造分区的逻辑，自定义一个partitioneer
    29. 

  29.  * 2，自定义reduer task的并发任务数
    30. 

  30.  */
    31. 

  31. public class FlowSumArea {
    32. 

    33. 

  33. 	public static class FlowSumAreaMapper extends Mapper<LongWritable, Text, Text, FlowBean>{
    34. 

  34. 		
    35. 

  35. 		@Override
    36. 

  36. 		protected void map(LongWritable key, Text value,Context context)
    37. 

  37. 				throws IOException, InterruptedException {
    38. 

  38. 		
    39. 

  39. 			//拿一行数据
    40. 

  40. 			String line = value.toString();
    41. 

  41. 			
    42. 

  42. 			//切分成各个字段
    43. 

  43. 			String[] fields = StringUtils.split(line,"	");
    44. 

  44. 			
    45. 

  45. 			//拿到我们的字段
    46. 

  46. 			String phoneNB = fields[1];
    47. 

  47. 			long u_flow = Long.parseLong(fields[7]);
    48. 

  48. 			long d_flow = Long.parseLong(fields[8]);
    49. 

  49. 			
    50. 

  50. 			//封装数据为kv并输出
    51. 

  51. 			context.write(new Text(phoneNB), new FlowBean(phoneNB,u_flow,d_flow));
    52. 

  52. 		}
    53. 

  53. 		
    54. 

  54. 		
    55. 

  55. 	}
    56. 

  56. 	
    57. 

  57. 	public static class FlowSumAreaReducer extends Reducer<Text, FlowBean, Text, FlowBean>{
    58. 

  58. 		
    59. 

  59. 		@Override
    60. 

  60. 		protected void reduce(Text key, Iterable<FlowBean> values,Context context)
    61. 

  61. 				throws IOException, InterruptedException {
    62. 

  62. 			
    63. 

  63. 			long up_flow_counter = 0;
    64. 

  64. 			long d_flow_counter = 0;
    65. 

  65. 			
    66. 

  66. 			for (FlowBean bean : values) {
    67. 

  67. 				up_flow_counter +=bean.getUp_flow();
    68. 

  68. 				d_flow_counter += bean.getD_flow();
    69. 

  69. 				
    70. 

  70. 				
    71. 

  71. 			}
    72. 

  72. 			context.write(key, new  FlowBean(key.toString(),up_flow_counter,d_flow_counter));
    73. 

  73. 			
    74. 

  74. 			
    75. 

  75. 		}
    76. 

  76. 	}
    77. 

  77. 	
    78. 

  78. 				public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    79. 

  79. 					
    80. 

  80. 					Configuration conf = new Configuration();
    81. 

  81. 					Job job = Job.getInstance(conf);
    82. 

  82. 					
    83. 

  83. 					job.setJarByClass(FlowSumArea.class);
    84. 

  84. 					
    85. 

  85. 					//job.setMapperClass(FlowSumAreaMapper.class);
    86. 

  86. 					job.setMapperClass(FlowSumAreaMapper.class);
    87. 

  87. 					job.setReducerClass(FlowSumAreaReducer.class);
    88. 

  88. 					
    89. 

  89. 					//设置我们自定义的分组逻辑定义
    90. 

  90. 					job.setPartitionerClass(AreaPartitioner.class);
    91. 

  91. 					
    92. 

  92. 					
    93. 

  93. 					job.setOutputKeyClass(Text.class);
    94. 

  94. 					job.setOutputValueClass(FlowBean.class);
    95. 

  95. 					
    96. 

  96. 					//设置reduce的任务并发数，应该跟分组的数量保持一致
    97. 

  97. 					job.setNumReduceTasks(6);
    98. 

  98. //进程数如果大了，后面的文件为空，小了会出现错误，为1则没有分组
    99. 

  99. 					
    100. 

  100. 					
    101. 

  101. 					FileInputFormat.setInputPaths(job, new Path(args[0]));
    102. 

  102. 					FileOutputFormat.setOutputPath(job, new Path(args[1]));
    103. 

  103. 					
    104. 

  104. 					
    105. 

  105. 					System.exit(job.waitForCompletion(true)?0:1);
    106. 

  106. 					
    107. 

  107. 				}
    108. 

  108. 	
    109. 

  109. }
    110. 


    111.
AreaPartitioner ：
    

  1. package cn.itcast.hadoop.mr.areapartition;
    2. 

    3. 

  3. import java.util.HashMap;
    4. 

    5. 

  5. import org.apache.hadoop.mapreduce.Partitioner;
    6. 

    7. 

  7. public class AreaPartitioner<KEY, VALUE> extends Partitioner<KEY, VALUE> {
    8. 

    9. 

  9. 	private static HashMap<String,Integer> areaMap = new HashMap<>();
    10. 

  10. 	
    11. 

  11. 	static{
    12. 

  12. 		areaMap.put("135", 0);
    13. 

  13. 		areaMap.put("136", 1);
    14. 

  14. 		areaMap.put("137", 2);
    15. 

  15. 		areaMap.put("138", 3);
    16. 

  16. 		areaMap.put("139", 4);
    17. 

  17. 	
    18. 

  18. 		
    19. 

  19. 	}
    20. 

  20. 	
    21. 

  21. 	@Override
    22. 

  22. 	public int getPartition(KEY key, VALUE value, int numPartitions) {
    23. 

  23. 		//从key中拿到手机号，查询手机归属地字典，不同省份返回不同的组号
    24. 

  24. 		int areaCoder = areaMap.get(key.toString().substring(0,3))==null?5:areaMap.get(key.toString().substring(0,3));
    25. 

  25. 		
    26. 

  26. 		return areaCoder;
    27. 

  27. 	}
    28. 

    29. 

  29. 	
    30. 

  30. }
    31.

运行：
hadoop jar /root/Documents/area.jar cn.itcast.hadoop.mr.areapartition.FlowSumArea /wc/data /wc/areasoutput
流量汇总（自定义jar包，在hadoop集群上 统计，排序，分组）之统计

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





























































附件列表

    相关推荐