devbox
Cpp五条
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Hadoop的MapReduce使用_hadoop mapreduce使用

作者：Cpp五条 | 2024-05-14 06:44:16

赞

踩

hadoop mapreduce使用

一、MapReduce框架结构

一个完整的mapreduce程序在分布式运行时有三类实例进程：

1、MRAppMaster：负责整个程序的过程调度及状态协调

2、MapTask：负责map阶段的整个数据处理流程

3、ReduceTask：负责reduce阶段的整个数据处理流程

二、MapReduce 编程规范及示例编写

2.1 编程规范

1、写一个类（MyMapper），继承hadoop框架的Mapper类，这个类就是map任务。我们只需要重写这个类的map方法（目的就是定义怎么检查每个组的作业）

2、写一个类（MyReducer），继承hadoop框架的Reducer类,这个类就是reduce任务。我们只需要重写这个类的reduce方法（目的就是定义怎么汇总那么多map任务的输出）

3、写一个普通的类（例如Demo），在Dem类中，创建一个Job对象，这个对象用于对MyMapper类和MyReducer类的配对，还用于配置MyMapper类的输出结果类型、输入数据来源。还用于配置MyReducer类的输出结果类型，输出结果目的地等等。

2.2 WordCount示例编写

需求：在一堆给定的文本文件中统计输出每一个单词出现的总次数

  1. public class Demo1_wordcount {
  2.  
  3.     private static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
  4.         IntWritable v = new IntWritable(1);
  5.         protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {
  6.             String[] arr = value.toString().split("\\s+");
  7.             for (String item : arr) {
  8.                 value.set(item);
  9.                 context.write(value, v);
  10.             }
  11.         }
  12.     }
  13.  
  14.     private static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
  15.         IntWritable v = new IntWritable();
  16.         @Override
  17.         protected void reduce(Text key, Iterable<IntWritable> value, Context context)throws IOException, InterruptedException {
  18.             int count = 0;
  19.             Iterator<IntWritable> it = value.iterator();
  20.             while(it.hasNext()){
  21.                 IntWritable next = it.next();
  22.                 count += next.get();
  23.             }
  24.             v.set(count);
  25.             context.write(key, v);
  26.         }
  27.     }
  28.  
  29.     public static void main(String[] args) throws Exception {
  30.         Job job = Job.getInstance();
  31.         //指定当前程序所在的jar包
  32.         job.setJarByClass(WordCountDemo01.class);
  33.  
  34.         //指定使用哪个Mapper类
  35.         job.setMapperClass(MyMapper.class);
  36.         //指定使用哪个Reducer类
  37.         job.setReducerClass(MyReducer.class);
  38.  
  39.         //指定mapper的输出类型
  40.         job.setMapOutputKeyClass(Text.class);
  41.         job.setMapOutputValueClass(IntWritable.class);
  42.  
  43.         //指定reducer的输出类型
  44.         job.setOutputKeyClass(Text.class);
  45.         job.setOutputValueClass(IntWritable.class);
  46.  
  47.         //指定数据来源
  48.         FileInputFormat.setInputPaths(job, new Path(args[0]));
  49.         //FileInputFormat.setInputPaths(job, new Path("adata/mr/wordcount/in"));
  50.         //指定目的地
  51.         FileOutputFormat.setOutputPath(job, new Path(args[1]));
  52.         //FileOutputFormat.setOutputPath(job, new Path("adata/mr/wordcount/out"));
  53.  
  54.         //提交任务
  55.         //job.submit();看不到日志的，不用它
  56.         boolean b = job.waitForCompletion(true);
  57.         System.exit(b ? 0 : 1);
  58.     }
  59. }

三、MapReduce 程序运行模式

3.1 本地运行模式

什么也不写，直接在工具上运行，默认就是本地运行模式。

前提：在windows中，解压好windows版本的hadoop，并且要配置好环境变量

  1. FileInputFormat.setInputPaths(job, new Path("adata/mr/wordcount/in"));
  2. FileOutputFormat.setOutputPath(job, new Path("adata/mr/wordcount/out"));

3.2 集群运行模式

1、注意修改数据来源是hdfs集群的路径、目的地也是集群中的路径

  1. FileInputFormat.setInputPaths(job, new Path(args[0]));
  2. FileOutputFormat.setOutputPath(job, new Path(args[1]));

2、对包含mr程序的项目打包，并且指定mainclass是谁（主方法所在的类就是mainclass）

  1. <plugin>
  2.     <groupId>org.apache.maven.plugins</groupId>
  3.     <artifactId>maven-jar-plugin</artifactId>
  4.     <version>2.4</version>
  5.     <configuration>
  6.         <archive>
  7.             <manifest>
  8.                 <addClasspath>true</addClasspath>
  9.                 <classpathPrefix>lib/</classpathPrefix>
  10.                 <mainClass>com.hadoop.Demo1_wordcount</mainClass>
  11.             </manifest>
  12.         </archive>
  13.     </configuration>
  14. </plugin>

3、把打好的jar包，上传到linux系统中

4、hdfs集群中要提前创建好数据来源，并且hdfs、yarn集群先启动

start-all.sh

5、hadoop jar hadoop-1.0-SNAPSHOT.jar /wc/in /wc/out

这里的/wc/in会传输到args[0]，/wc/out会传输到args[1]中

6、查看yarn集群界面：http://node1:8088

四、序列化

4.1  Writable 序列化接口  

  1. public class FlowBean implements Writable {
  2.  
  3.     private long upflow;	//上行流量
  4.     private long downflow;	//下行流量
  5.  
  6.     public long getUpflow() {
  7.         return upflow;
  8.     }
  9.  
  10.     public void setUpflow(long upflow) {
  11.         this.upflow = upflow;
  12.     }
  13.  
  14.     public long getDownflow() {
  15.         return downflow;
  16.     }
  17.  
  18.     public void setDownflow(long downflow) {
  19.         this.downflow = downflow;
  20.     }
  21.  
  22.     @Override
  23.     public void write(DataOutput out) throws IOException {
  24.         //先序列化upflow
  25.         out.writeLong(this.upflow);
  26.         //再序列化downflow
  27.         out.writeLong(this.downflow);
  28.     }
  29.  
  30.     @Override
  31.     public void readFields(DataInput in) throws IOException {
  32.         //第一次调用readLong方法，其实是read到了upflow，因为序列化的时候，是先序列化upflow的
  33.         this.upflow = in.readLong();
  34.         this.downflow = in.readLong();
  35.     }
  36.  
  37.     @Override
  38.     public String toString() {
  39.         return this.upflow + "\t" + this.downflow + "\t" + (this.upflow + this.downflow);
  40.     }
  41. }

4.2 流量汇总

  1. public class Demo02_FlowSum {
  2.  
  3.     public static void main(String[] args) throws Exception {
  4.         Job job = Job.getInstance();
  5.         job.setJarByClass(Demo02_FlowSum.class);
  6.  
  7.         job.setMapOutputKeyClass(Text.class);
  8.         job.setMapOutputValueClass(FlowBean.class);
  9.  
  10.         job.setOutputKeyClass(Text.class);
  11.         job.setOutputValueClass(FlowBean.class);
  12.  
  13.         job.setMapperClass(MyMapper.class);
  14.         job.setReducerClass(MyReducer.class);
  15.  
  16.         FileInputFormat.setInputPaths(job, new Path("adata/mr/wordcount/in"));
  17.         FileOutputFormat.setOutputPath(job, new Path("adata/mr/wordcount/out1"));
  18.  
  19.         System.exit(job.waitForCompletion(true) ? 0 : 1);
  20.     }
  21.  
  22.     public static class MyMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
  23.  
  24.         private Text k = new Text();
  25.         private FlowBean v = new FlowBean();
  26.  
  27.         @Override
  28.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  29.             String[] arr = value.toString().split("\\s+");
  30.             v.setUpflow(Integer.parseInt(arr[arr.length - 3]));
  31.             v.setDownflow(Integer.parseInt(arr[arr.length - 2]));
  32.             k.set(arr[1]);
  33.             context.write(k, v);
  34.         }
  35.     }
  36.  
  37.     public static class MyReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
  38.         private FlowBean flowBean = new FlowBean();
  39.         @Override
  40.         protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
  41.             flowBean.setUpflow(0);
  42.             flowBean.setDownflow(0);
  43.             for (FlowBean value : values) {
  44.                 flowBean.setUpflow(value.getUpflow() + flowBean.getUpflow());
  45.                 flowBean.setDownflow(value.getDownflow() + flowBean.getDownflow());
  46.             }
  47.             context.write(key, flowBean);
  48.         }
  49.     }
  50. }

五、自定义排序

5.1 需求

在得出统计每一个用户（手机号）所耗费的总上行流量、下行流量，总流量结果的基础之上再加一个需求：将统计结果按照总流量倒序排序。

5.2 排序代码实现

  1. public class FlowBean1 implements WritableComparable<FlowBean1> {
  2.  
  3.     private String phone;
  4.     private int upflow;	//上行流量
  5.     private int downflow;	//下行流量
  6.  
  7.     public String getPhone() {
  8.         return phone;
  9.     }
  10.  
  11.     public void setPhone(String phone) {
  12.         this.phone = phone;
  13.     }
  14.  
  15.     public int getUpflow() {
  16.         return upflow;
  17.     }
  18.  
  19.     public void setUpflow(int upflow) {
  20.         this.upflow = upflow;
  21.     }
  22.  
  23.     public int getDownflow() {
  24.         return downflow;
  25.     }
  26.  
  27.     public void setDownflow(int downflow) {
  28.         this.downflow = downflow;
  29.     }
  30.  
  31.     @Override
  32.     public void write(DataOutput out) throws IOException {
  33.         out.writeUTF(this.phone);
  34.         //先序列化upflow
  35.         out.writeInt(this.upflow);
  36.         //再序列化downflow
  37.         out.writeInt(this.downflow);
  38.     }
  39.  
  40.     @Override
  41.     public void readFields(DataInput in) throws IOException {
  42.         //第一次调用readLong方法，其实是read到了upflow，因为序列化的时候，是先序列化upflow的
  43.         this.phone = in.readUTF();
  44.         this.upflow = in.readInt();
  45.         this.downflow = in.readInt();
  46.     }
  47.  
  48.     @Override
  49.     public String toString() {
  50.         return this.phone + "\t" + this.upflow + "\t" + this.downflow + "\t" + (this.upflow + this.downflow);
  51.     }
  52.  
  53.     /*
  54.         1、按照总流量从大到小的顺序排序
  55.         2、总流量一样的情况，按照手机号的字典顺序来排序
  56.     */
  57.     @Override
  58.     public int compareTo(FlowBean1 o) {
  59.         int sum = this.upflow + this.downflow - o.upflow - o.downflow;
  60.         if (sum != 0) {
  61.             return -sum;
  62.         }
  63.         int num = this.phone.compareTo(o.phone);
  64.         if (num != 0) {
  65.             return -num;
  66.         }
  67.         return 111;
  68.     }
  69. }
  1. public class Demo03_FlowSumSort {
  2.  
  3.     public static void main(String[] args) throws Exception {
  4.         Job job = Job.getInstance();
  5.         job.setJarByClass(Demo03_FlowSumSort.class);
  6.  
  7.         job.setMapOutputKeyClass(FlowBean1.class);
  8.         job.setMapOutputValueClass(NullWritable.class);
  9.  
  10.         job.setOutputKeyClass(FlowBean1.class);
  11.         job.setOutputValueClass(NullWritable.class);
  12.  
  13.         job.setMapperClass(MyMapper.class);
  14.         job.setReducerClass(MyReducer.class);
  15.  
  16.         FileInputFormat.setInputPaths(job, new Path("adata/mr/wordcount/out1"));
  17.         FileOutputFormat.setOutputPath(job, new Path("adata/mr/wordcount/out2"));
  18.  
  19.         System.exit(job.waitForCompletion(true) ? 0 : 1);
  20.     }
  21.  
  22.     public static class MyMapper extends Mapper<LongWritable, Text, FlowBean1, NullWritable> {
  23.         private FlowBean1 bean = new FlowBean1();
  24.         private NullWritable v = NullWritable.get();
  25.         @Override
  26.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  27.             String[] arr = value.toString().split("\\s+");
  28.             bean.setPhone(arr[0]);
  29.             bean.setUpflow(Integer.parseInt(arr[1]));
  30.             bean.setDownflow(Integer.parseInt(arr[2]));
  31.             context.write(bean, v);
  32.         }
  33.     }
  34.  
  35.     public static class MyReducer extends Reducer<FlowBean1, NullWritable, FlowBean1, NullWritable> {
  36.         private NullWritable v = NullWritable.get();
  37.         @Override
  38.         protected void reduce(FlowBean1 key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
  39.             context.write(key, v);
  40.         }
  41.     }
  42. }

六、自定义分区

6.1 需求

将流量汇总统计结果，按照手机归属地不同省份输出到不同文件中。

6.2 代码实现

  1. public class Demo04_Partitioner {
  2.  
  3.     public static void main(String[] args) throws Exception {
  4.         Job job = Job.getInstance();
  5.         job.setJarByClass(Demo04_Partitioner.class);
  6.  
  7.         job.setMapOutputKeyClass(Text.class);
  8.         job.setMapOutputValueClass(NullWritable.class);
  9.  
  10.         job.setOutputKeyClass(Text.class);
  11.         job.setOutputValueClass(NullWritable.class);
  12.  
  13.         job.setMapperClass(MyMapper.class);
  14.         job.setReducerClass(MyReducer.class);
  15.         job.setPartitionerClass(MyPartitioner.class);//设置使用自己的分区类
  16.         job.setNumReduceTasks(3);//有几个分区，就设置几个reducetask
  17.  
  18.         FileInputFormat.setInputPaths(job, new Path("adata/mr/wordcount/out2"));
  19.         FileOutputFormat.setOutputPath(job, new Path("adata/mr/wordcount/out3"));
  20.  
  21.         System.exit(job.waitForCompletion(true) ? 0 : 1);
  22.     }
  23.  
  24.  
  25.     public static class MyMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
  26.         @Override
  27.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  28.             context.write(value, NullWritable.get());
  29.         }
  30.     }
  31.  
  32.     public static class MyPartitioner extends Partitioner<Text, NullWritable> {
  33.         @Override
  34.         public int getPartition(Text text, NullWritable nullWritable, int numPartitions) {
  35.             String phone = text.toString().substring(0, 3);
  36.             switch (phone) {
  37.                 case "134":
  38.                 case "135":
  39.                 case "136":
  40.                     return 0;
  41.                 case "137":
  42.                 case "138":
  43.                     return 1;
  44.             }
  45.             return 2;
  46.         }
  47.     }
  48.  
  49.     public static class MyReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
  50.  
  51.         private NullWritable v = NullWritable.get();
  52.         @Override
  53.         protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
  54.             context.write(key, v);
  55.         }
  56.     }
  57. }

七、combiner的使用

每一个 map 都可能会产生大量的本地输出，Combiner 的作用就是对 map 端的输出先做一次合并，以减少在 map 和 reduce 节点之间的数据传输量，以提高网络 IO 性能，是 MapReduce 的一种优化手段之一。

Combiner的父类，也是Reducer。使用时，可以公用MyReducer类，作为Combinner类 。注意：Combiner的输出类型，必须和Mapper的输出类型一致。

代码实现：

  1. public class Demo05_WordCount {
  2.  
  3.     public static void main(String[] args) throws Exception {
  4.         Job job = Job.getInstance();
  5.  
  6.         job.setCombinerClass(MyCombiner.class);
  7.  
  8.         //指定当前程序所在的jar包
  9.         job.setJarByClass(Demo05_WordCount.class);
  10.  
  11.         //指定数据来源
  12.         FileInputFormat.setInputPaths(job,new Path("adata/mr/wordcount/in"));
  13.  
  14.         //指定目的地
  15.         FileOutputFormat.setOutputPath(job,new Path("adata/mr/wordcount/out4"));
  16.  
  17.         //指定使用哪个mapper类
  18.         job.setMapperClass(MyMapper.class);
  19.  
  20.         //指定使用哪个Reducer类
  21.         job.setReducerClass(MyReducer.class);
  22.  
  23.         //指定map方法的输出类型
  24.         job.setMapOutputKeyClass(Text.class);
  25.         job.setMapOutputValueClass(IntWritable.class);
  26.  
  27.         //指定reduce方法的输出类型
  28.         job.setOutputKeyClass(Text.class);
  29.         job.setOutputValueClass(IntWritable.class);
  30.  
  31.         //提交任务
  32.         //job.submit();看不到日志的，不用
  33.         boolean b = job.waitForCompletion(true);
  34.  
  35.         //如果是0，就代表程序运行正常，正常退出虚拟机。非0就代表异常
  36.         System.exit(b ? 0 : 443);
  37.     }
  38.  
  39.  
  40.     public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
  41.         private IntWritable v = new IntWritable(1);
  42.         private Text k = new Text();
  43.         @Override
  44.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  45.             String[] arr = value.toString().split("\\s+");
  46.             for (String str : arr) {
  47.                 k.set(str);
  48.                 context.write(k, v);
  49.             }
  50.         }
  51.     }
  52.  
  53.     public static class MyCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
  54.         private IntWritable v = new IntWritable();
  55.         @Override
  56.         protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
  57.             int count = 0;
  58.             for (IntWritable value : values) {
  59.                 count += value.get();
  60.             }
  61.             v.set(count);
  62.             context.write(key, v);
  63.         }
  64.     }
  65.  
  66.     public static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  67.         @Override
  68.         protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
  69.             for (IntWritable value : values) {
  70.                 context.write(key, value);
  71.             }
  72.         }
  73.     }
  74. }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Cpp五条/article/detail/567765
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号