全网最详细的Hadoop文章系列&＃xff0c;强烈建议收藏加关注&＃xff01;

后面更新文章都会列出历史文章目录&＃xff0c;帮助大家回顾知识重点。

目录

本系列历史文章

前言

MapReduce高阶训练

一、上网流量统计

二、需求&＃xff1a;统计求和

1、思路分析

2、代码实现

本系列历史文章

2021年大数据Hadoop&＃xff08;三十&＃xff09;&＃xff1a;Hadoop3.x的介绍

2021年大数据Hadoop&＃xff08;二十九&＃xff09;&＃xff1a;关于YARN常用参数设置

2021年大数据Hadoop&＃xff08;二十八&＃xff09;&＃xff1a;YARN的调度器Scheduler

2021年大数据Hadoop&＃xff08;二十七&＃xff09;&＃xff1a;YARN运行流程

2021年大数据Hadoop&＃xff08;二十六&＃xff09;&＃xff1a;YARN三大组件介绍

2021年大数据Hadoop&＃xff08;二十五&＃xff09;&＃xff1a;YARN通俗介绍和基本架构

2021年大数据Hadoop&＃xff08;二十四&＃xff09;&＃xff1a;MapReduce高阶训练

2021年大数据Hadoop&＃xff08;二十三&＃xff09;&＃xff1a;MapReduce的运行机制详解

2021年大数据Hadoop&＃xff08;二十二&＃xff09;&＃xff1a;MapReduce的自定义分组

2021年大数据Hadoop&＃xff08;二十一&＃xff09;&＃xff1a;MapReuce的Combineer

2021年大数据Hadoop&＃xff08;二十&＃xff09;&＃xff1a;MapReduce的排序和序列化

2021年大数据Hadoop&＃xff08;十九&＃xff09;&＃xff1a;MapReduce分区

2021年大数据Hadoop&＃xff08;十八&＃xff09;&＃xff1a;MapReduce程序运行模式和深入解析

2021年大数据Hadoop&＃xff08;十七&＃xff09;&＃xff1a;MapReduce编程规范及示例编写

2021年大数据Hadoop&＃xff08;十六&＃xff09;&＃xff1a;MapReduce计算模型介绍

2021年大数据Hadoop&＃xff08;十五&＃xff09;&＃xff1a;Hadoop的联邦机制 Federation

2021年大数据Hadoop&＃xff08;十四&＃xff09;&＃xff1a;HDFS的高可用机制

2021年大数据Hadoop&＃xff08;十三&＃xff09;&＃xff1a;HDFS意想不到的其他功能

2021年大数据Hadoop&＃xff08;十二&＃xff09;&＃xff1a;HDFS的API操作

2021年大数据Hadoop&＃xff08;十一&＃xff09;&＃xff1a;HDFS的元数据辅助管理

2021年大数据Hadoop&＃xff08;十&＃xff09;&＃xff1a;HDFS的数据读写流程

2021年大数据Hadoop&＃xff08;九&＃xff09;&＃xff1a;HDFS的高级使用命令

2021年大数据Hadoop&＃xff08;八&＃xff09;&＃xff1a;HDFS的Shell命令行使用

2021年大数据Hadoop&＃xff08;七&＃xff09;&＃xff1a;HDFS分布式文件系统简介

2021年大数据Hadoop&＃xff08;六&＃xff09;&＃xff1a;全网最详细的Hadoop集群搭建

2021年大数据Hadoop&＃xff08;五&＃xff09;&＃xff1a;Hadoop架构

2021年大数据Hadoop&＃xff08;四&＃xff09;&＃xff1a;Hadoop发行版公司

2021年大数据Hadoop&＃xff08;三&＃xff09;&＃xff1a;Hadoop国内外应用

2021年大数据Hadoop&＃xff08;二&＃xff09;&＃xff1a;Hadoop发展简史和特性优点

2021年大数据Hadoop&＃xff08;一&＃xff09;&＃xff1a;Hadoop介绍

前言

2021大数据领域优质创作博客&＃xff0c;带你从入门到精通&＃xff0c;该博客每天更新&＃xff0c;逐渐完善大数据各个知识体系的文章&＃xff0c;帮助大家更高效学习。

有对大数据感兴趣的可以关注微信公众号&＃xff1a;三帮大数据

MapReduce高阶训练

一、上网流量统计

数据格式如下:

二、需求&＃xff1a;统计求和

统计每个手机号的上行数据包数总和&＃xff0c;下行数据包数总和&＃xff0c;上行总流量之和&＃xff0c;下行总流量之和

分析&＃xff1a;以手机号码作为key值&＃xff0c;上行数据包&＃xff0c;下行数据包&＃xff0c;上行总流量&＃xff0c;下行总流量四个字段作为value值&＃xff0c;然后以这个key和value作为map阶段的输出&＃xff0c;reduce阶段的输入。

1、思路分析

2、代码实现

第一步&＃xff1a;自定义map的输出value对象FlowBean

import org.apache.hadoop.io.Writable;import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;import java.io.DataOutput;import java.io.IOException;//如果MR中的JavaBean不是作为K2&＃xff0c;则只需要实现Writable接口即可public class FlowBean implements Writable{private Integer upFlow;    //上行包数private Integer  downFlow; //下行包数private Integer upCountFlow; //上行流量和private Integer downCountFlow; //下行流量和public FlowBean() {}public FlowBean(Integer upFlow, Integer downFlow, Integer upCountFlow, Integer downCountFlow) {this.upFlow &＃61; upFlow;this.downFlow &＃61; downFlow;this.upCountFlow &＃61; upCountFlow;this.downCountFlow &＃61; downCountFlow;}public Integer getUpFlow() {return upFlow;}&＃64;Overridepublic String toString() {return upFlow &＃43;"\t" &＃43; downFlow &＃43;"\t" &＃43; upCountFlow &＃43;"\t" &＃43; downCountFlow;}public void setUpFlow(Integer upFlow) {this.upFlow &＃61; upFlow;}public Integer getDownFlow() {return downFlow;}public void setDownFlow(Integer downFlow) {this.downFlow &＃61; downFlow;}public Integer getUpCountFlow() {return upCountFlow;}public void setUpCountFlow(Integer upCountFlow) {this.upCountFlow &＃61; upCountFlow;}public Integer getDownCountFlow() {return downCountFlow;}public void setDownCountFlow(Integer downCountFlow) {this.downCountFlow &＃61; downCountFlow;}//序列化&＃64;Overridepublic void write(DataOutput out) throws IOException {out.writeInt(upFlow);out.writeInt(downFlow);out.writeInt(upCountFlow);out.writeInt(downCountFlow);}//反序列化&＃64;Overridepublic void readFields(DataInput in) throws IOException {this.upFlow &＃61; in.readInt();this.downFlow &＃61; in.readInt();this.upCountFlow &＃61; in.readInt();this.downCountFlow &＃61; in.readInt();}}

第二步&＃xff1a;定义FlowMapper类

import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class FlowCountMapper extends Mapper {&＃64;Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//1:拆分行文本数据&＃xff08;拆分v1&＃xff09;String[] split &＃61; value.toString().split("\t");//2:从拆分数组中得到手机号&＃xff0c;得到K2String phoneNum &＃61; split[1];//3:从拆分数组中得到4个流量字段&＃xff0c;并封装到FlowBean,得到V2FlowBean flowBean &＃61; new FlowBean();flowBean.setUpFlow(Integer.parseInt(split[6]));flowBean.setDownFlow(Integer.parseInt(split[7]));flowBean.setUpCountFlow(Integer.parseInt(split[8]));flowBean.setDownCountFlow(Integer.parseInt(split[9]));//4:将K2和V2写入上下文中context.write(new Text(phoneNum), flowBean);}}

第三步&＃xff1a;定义FlowCountReducer类

import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class FlowCountReducer extends Reducer {&＃64;Overrideprotected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {//1:定义四个变量&＃xff0c;分别来存储上行包数&＃xff0c;下行包数&＃xff0c;上行流量总和&＃xff0c;下行流量总和int upFlow &＃61; 0;int downFlow &＃61; 0;int upCountFlow &＃61; 0;int downCountFlow &＃61; 0;//2:遍历集合&＃xff0c;将集合中每一个FlowBean的四个流量字段相加for (FlowBean flowBean : values) {upFlow &＃43;&＃61; flowBean.getUpFlow();downFlow &＃43;&＃61; flowBean.getDownFlow();upCountFlow &＃43;&＃61; flowBean.getUpCountFlow();downCountFlow &＃43;&＃61; flowBean.getDownCountFlow();}//3:K3就是原来的K2&＃xff0c;V3就是新的FlowBeanFlowBean flowBean &＃61; new FlowBean(upFlow, downFlow, upCountFlow, downCountFlow);//4:将K3和V3写入上下文中context.write(key, flowBean);}}

第四步&＃xff1a;程序main函数入口FlowCountRunner

import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import java.io.IOException;public class FlowCountRunner {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//1、创建建一个job任务对象Configuration configuration &＃61; new Configuration();Job job &＃61; Job.getInstance(configuration, "flowcount_demo");//2、指定job所在的jar包job.setJarByClass(FlowCountRunner.class);//3、指定源文件的读取方式类和源文件的读取路径job.setInputFormatClass(TextInputFormat.class); //按照行读取//TextInputFormat.addInputPath(job, new Path("hdfs://node1:8020/input/wordcount")); //只需要指定源文件所在的目录即可TextInputFormat.addInputPath(job, new Path("file:///E:\\input\\flowcount")); //只需要指定源文件所在的目录即可//4、指定自定义的Mapper类和K2、V2类型job.setMapperClass(FlowCountMapper.class); //指定Mapper类job.setMapOutputKeyClass(Text.class); //K2类型job.setMapOutputValueClass(FlowBean.class);//V2类型//5、指定自定义分区类&＃xff08;如果有的话&＃xff09;//6、指定自定义分组类&＃xff08;如果有的话&＃xff09;//7、指定自定义Combiner类(如果有的话)//设置ReduceTask个数//8、指定自定义的Reducer类和K3、V3的数据类型job.setReducerClass(FlowCountReducer.class); //指定Reducer类job.setOutputKeyClass(Text.class); //K3类型job.setOutputValueClass(FlowBean.class);  //V3类型//9、指定输出方式类和结果输出路径job.setOutputFormatClass(TextOutputFormat.class);//TextOutputFormat.setOutputPath(job, new  Path("hdfs://node1:8020/output/wordcount")); //目标目录不能存在&＃xff0c;否则报错TextOutputFormat.setOutputPath(job, new  Path("file:///E:\\output\\flowcount")); //目标目录不能存在&＃xff0c;否则报错//10、将job提交到yarn集群boolean bl &＃61; job.waitForCompletion(true); //true表示可以看到任务的执行进度//11.退出执行进程System.exit(bl?0:1);}}

• &＃x1f4e2;博客主页&＃xff1a;https://lansonli.blog.csdn.net
• &＃x1f4e2;欢迎点赞 &＃x1f44d; 收藏 ⭐留言 &＃x1f4dd; 如有错误敬请指正&＃xff01;
• &＃x1f4e2;本文由 Lansonli 原创&＃xff0c;首发于 CSDN博客&＃x1f649;
• &＃x1f4e2;大数据系列文章会每天更新&＃xff0c;停下休息的时候不要忘了别人还在奔跑&＃xff0c;希望大家抓紧时间学习&＃xff0c;全力奔赴更美好的生活✨