Hadoop：MapReduce概述和编程案例

偏执的太偏执、 2023-02-12 13:24 58阅读 0赞

### 文章目录 ###

*  一、MapReduce概述
 *   *  1.1 MapReduce优点
     *  1.2 MapReduce缺点
 *  二、MapReduce核心思想
 *  三、MapReduce编程规范
 *   *  3.1 常用的数据类型对应的Hadoop数据序列化类型
     *  3.2 MapReduce编程规范
 *  四、WordCount案例实操
 *  五、序列化
 *   *  5.1 序列化概述
     *  5.2 自定义bean对象实现序列化接口（Writable）
 *  六、序列化案例实操

# 一、MapReduce概述 #

`MapReduce`是一个**分布式运算程序的编程框架**，是用户开发“基于`Hadoop`的数据分析应用”的核心框架；

`MapReduce`核心功能是将用**户编写的业务逻辑代码**和**自带默认组件**整合成一个完整的分布式运算程序，并发运行在一个`Hadoop`集群上。

## 1.1 MapReduce优点 ##

**1.MapReduce 易于编程**  
它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的 `PC`机器运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。 就是因为这个特点使得`MapReduce`编程变得非常流行。

**2.良好的扩展性**  
当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。

**3.高容错性**  
`MapReduce`设计的初衷就是使程序能够部署在廉价的`PC`机器上，这就要求它具有很高的容错性。比如其中一台机器挂了，它可以把上面的计算任务转移到另外一个节点上面上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由`Hadoop`内部完成的。

**4.适合PB级以上海量数据的离线处理**  
可以实现上干台服务器集群并发工作，提供数据处理能力。

## 1.2 MapReduce缺点 ##

**1.实时计算**  
`MapReduce`无法像`Mysql`一样，在毫秒或者秒级内返回结果。

**2.流式计算**  
流式计算的输入数据时动态的，而`MapReduce`的输入数据集是静态的，不能动态变化。这是因为`MapReduce`自身的设计特点决定了数据源必须是静态的。

**3.DAG（有向图）计算**  
多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，`MapReduce`并不是不能做，而是使用后，每个`MapReduce`作业的输出结果都会写入到磁盘，会造成大量的磁盘`IO`，导致性能非常的低下。

# 二、MapReduce核心思想 #

![在这里插入图片描述][20200526143007579.png]

1.  分布式的运算程序往往需要分成至少2个阶段
2.  第一个阶段的`MapTask`并发实例，完全并行运行，互不相干
3.  第二个阶段的`ReduceTask`并发实例互不相干，但是他们的数据依赖于上一个阶段的所有`MapTask`并发实例的输出
4.  `MapReduce`编程模型只能包含一个`Map`阶段和一个`Reduce`阶段，如果用户的业务逻辑非常复杂，那就只能多个`MapReduce`程序，串行运行

# 三、MapReduce编程规范 #

## 3.1 常用的数据类型对应的Hadoop数据序列化类型 ##

<table> 
 <thead> 
  <tr> 
   <th>Java类型</th> 
   <th>Hadoop Writable类型</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Boolean</td> 
   <td>BooleanWritable</td> 
  </tr> 
  <tr> 
   <td>Byte</td> 
   <td>ByteWritable</td> 
  </tr> 
  <tr> 
   <td>Int</td> 
   <td>IntWritable</td> 
  </tr> 
  <tr> 
   <td>Float</td> 
   <td>FloatWritable</td> 
  </tr> 
  <tr> 
   <td>Long</td> 
   <td>LongWritable</td> 
  </tr> 
  <tr> 
   <td>Double</td> 
   <td>DoubleWritable</td> 
  </tr> 
  <tr> 
   <td>String</td> 
   <td>Text</td> 
  </tr> 
  <tr> 
   <td>Map</td> 
   <td>MapWritable</td> 
  </tr> 
  <tr> 
   <td>Array</td> 
   <td>ArrayWritable</td> 
  </tr> 
 </tbody> 
</table>

## 3.2 MapReduce编程规范 ##

用户编写的程序分成三个部分：`Mapper`、`Reducer`和`Driver`。

1.  `Mapper`阶段  
    ①用户自定义的`Mapper`要继承自己的父类  
    ②`Mapper`的输入数据是`KV`对的形式（`KV`的类型可自定义）  
    ③`Mapper`中的业务逻辑写在`map()`方法中  
    ④`Mapper`的输出数据是`KV`对的形式（`KV`的类型可自定义）  
    ⑤maptask进程对每一个`<K,V>`调用一次`map()`方法
2.  `Reducer`阶段  
    ①用户自定义的`Reducer`要继承自己的父类  
    ②`Reducer`的输入数据类型对应`Mapper`的输出数据类型，也是`KV`  
    ③`Reducer`的业务逻辑写在`reduce()`方法中  
    ④`Reducetask`进程对每一组相同`K`的`<K,V>`组调用一次`reduce()`方法
3.  `Driver`阶段  
    相当于`YARN`集群的客户端，用于提交我们整个程序到`YARN`集群,提交的是封装了`MapReduce`程序相关运行参数的`Job`对象

# 四、WordCount案例实操 #

**①添加依赖**

<dependencies>
    	<dependency>
    		<groupId>junit</groupId>
    		<artifactId>junit</artifactId>
    		<version>RELEASE</version>
    	</dependency>
    	<dependency>
    		<groupId>org.apache.logging.log4j</groupId>
    		<artifactId>log4j-core</artifactId>
    		<version>2.8.2</version>
    	</dependency>
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-common</artifactId>
    		<version>2.7.2</version>
    	</dependency>
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-client</artifactId>
    		<version>2.7.2</version>
    	</dependency>
    	<dependency>
    		<groupId>org.apache.hadoop</groupId>
    		<artifactId>hadoop-hdfs</artifactId>
    		<version>2.7.2</version>
    	</dependency>
    </dependencies>

②`src/main/resources`目录下新建`log4j.properties`，内容如下：

log4j.rootLogger=INFO, stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
    log4j.appender.logfile=org.apache.log4j.FileAppender
    log4j.appender.logfile.File=target/spring.log
    log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
    log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

**③编写程序**  
**1.编写Mapper类**

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    /** * KEYIN：输入数据首行的偏移量（固定） * VALUEIN：输入数据的类型 * KEYOUT：输出数据的K * VALUEOUT：输出数据的V */
    public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { 
        Text k = new Text();
        IntWritable v = new IntWritable(1);
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            //1.获取一行
            String line = value.toString();
            //2.根据空格切割
            String[] words = line.split(" ");
            for (String word : words) { 
                k.set(word);
                context.write(k, v);
            }
        }
    }

**2.编写Reducer类**

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /** * KEYIN：输入数据的K * VALUEIN：输入数据的V * KEYOUT：输出数据的K * VALUEOUT：输出数据的V */
    public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { 
        int sum;
        IntWritable v = new IntWritable();
    
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
            sum = 0;
            values.forEach(value -> sum += value.get());
            v.set(sum);
            context.write(key, v);
        }
    }

**3.编写Driver驱动类**

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /** * @author HuChan */
    public class WordCountDriver { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            //1.获取配置信息以及封装任务
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration);
    
            //2.设置jar的加载路径
            job.setJarByClass(WordCountDriver.class);
    
            //3.设置map和reduce类
            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCountReducer.class);
    
            //4.设置map输出类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            //5.设置最终输出kv类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            //6.设置输入输出路径
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
            //7.提交
            boolean result = job.waitForCompletion(true);
    
            System.exit(result ? 0 : 1);
        }
    }

**4. 添加启动配置**  
![在这里插入图片描述][20200526155733274.png]  
**5.win10可能出现的异常**

`Exception in thread "main" java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Z`

**解决方案：将`hadoop.dll`放置到`C:\Windows\System32`下**

**6.打包运行在集群上，执行命令即可**

[root@hadoop100 software]$ hadoop jar  wc.jar
     com.hucheng.mapreduce.WordCountDriver /user/input /user/output

# 五、序列化 #

## 5.1 序列化概述 ##

**序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储（持久化）和网络传输。**

**反序列化就是将收到字节序列（或其他数据传输协议）或者是硬盘的持久化数据，转换成内存中的对象。**

`Java`的序列化是一个重量级序列化框架（`Serializable`），一个对象被序列化后，会附带很多额外的信息（各种校验信息，`Header`，继承体系等），不便于在网络中高效传输。所以`Hadoop`自己开发了一套序列化机制（`Writable`），精简、高效。

`Hadoop`序列化特点：

1.  **紧凑**：高效使用存储空间
2.  **快速**：读写数据的额外开销小
3.  **可扩展**：随着通信协议的升级而可升级
4.  **互操作**：支持多语言的交互

## 5.2 自定义bean对象实现序列化接口（Writable） ##

在企业开发中往往常用的基本序列化类型不能满足所有需求，比如在`Hadoop`框架内部传递一个`Bean`对象，那么该对象就需要实现序列化接口。

具体实现`Bean`对象序列化步骤如下7步:

1.  必须实现`Writable`接口
2.  反序列化时，需要反射调用空参构造函数，所以必须有空参构造

public FlowBean() { 
    	super();
    }

1.  重写序列化方法

@Override
    public void write(DataOutput out) throws IOException { 
    	out.writeLong(upFlow);
    	out.writeLong(downFlow);
    	out.writeLong(sumFlow);
    }

1.  重写反序列化方法

@Override
    public void readFields(DataInput in) throws IOException { 
    	upFlow = in.readLong();
    	downFlow = in.readLong();
    	sumFlow = in.readLong();
    }

1.  注意反序列化的顺序和序列化的顺序完全一致
2.  注意反序列化的顺序和序列化的顺序完全一致
3.  如果需要将自定义的`Bean`放在`key`中传输，则还需要实现`Comparable`接口，因为`MapReduce`框中的`Shuffle`过程要求对`key`必须能排序。

@Override
    public int compareTo(FlowBean o) { 
    	// 倒序排列，从大到小
    	return this.sumFlow > o.getSumFlow() ? -1 : 1;
    }

# 六、序列化案例实操 #

**需求:**  
统计每一个手机号耗费的总上行流量、下行流量、总流量

**输入数据：**  
![在这里插入图片描述][20200526195547326.png]  
**期望输出数据格式：**  
![在这里插入图片描述][20200526195700274.png]  
**编写MapReduce程序：**

**①编写流量统计的Bean对象**

public class FlowBean implements Writable { 
        private long upFlow;
        private long downFlow;
        private long sumFlow;
    
        public FlowBean() { 
        }
    
        public void set(long upFlow, long downFlow) { 
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.sumFlow = upFlow + downFlow;
        }
    
        public long getUpFlow() { 
            return upFlow;
        }
    
        public void setUpFlow(long upFlow) { 
            this.upFlow = upFlow;
        }
    
        public long getDownFlow() { 
            return downFlow;
        }
    
        public void setDownFlow(long downFlow) { 
            this.downFlow = downFlow;
        }
    
        public long getSumFlow() { 
            return sumFlow;
        }
    
        public void setSumFlow(long sumFlow) { 
            this.sumFlow = sumFlow;
        }
    
        @Override
        public void write(DataOutput dataOutput) throws IOException { 
            dataOutput.writeLong(upFlow);
            dataOutput.writeLong(downFlow);
            dataOutput.writeLong(sumFlow);
        }
    
        @Override
        public void readFields(DataInput dataInput) throws IOException { 
            upFlow = dataInput.readLong();
            downFlow = dataInput.readLong();
            sumFlow = dataInput.readLong();
        }
    
        @Override
        public String toString() { 
            return "FlowBean{" +
                    "upFlow=" + upFlow +
                    ", downFlow=" + downFlow +
                    ", sumFlow=" + sumFlow +
                    '}';
        }
    }

**②编写Mapper类**

public class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean> { 
    
        Text k = new Text();
        FlowBean v = new FlowBean();
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            //1.获取行
            String line = value.toString();
            //2.根据空格切分
            String[] fields = line.split("\t");
            //3.取出想要对象
            String phone = fields[1];
            long upFlow = Long.parseLong(fields[fields.length - 3]);
            long downFlow = Long.parseLong(fields[fields.length - 2]);
            k.set(phone);
            v.set(upFlow, downFlow);
            context.write(k, v);
        }
    }

**③编写Reducer类**

public class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean> { 
        long sum_upFlow;
        long sum_downFlow;
        FlowBean v = new FlowBean();
    
        @Override
        protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { 
            sum_upFlow = 0;
            sum_downFlow = 0;
            values.forEach(value -> { 
                sum_upFlow += value.getUpFlow();
                sum_downFlow += value.getDownFlow();
            });
            v.set(sum_upFlow, sum_downFlow);
            context.write(key, v);
        }
    }

**④编写Driver驱动类**

public class FlowCountDriver { 
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
            //1.获取配置信息以及封装任务
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration);
    
            //2.设置jar的加载路径
            job.setJarByClass(FlowCountDriver.class);
    
            //3.设置map和reduce类
            job.setMapperClass(FlowCountMapper.class);
            job.setReducerClass(FlowCountReducer.class);
    
            //4.设置map输出类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(FlowBean.class);
    
            //5.设置最终输出kv类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
    
            //6.设置输入输出路径
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
            //7.提交
            boolean result = job.waitForCompletion(true);
    
            System.exit(result ? 0 : 1);
        }
    }

**⑤最终效果**  
![在这里插入图片描述][20200526203409965.png]

[20200526143007579.png]: https://img-blog.csdnimg.cn/20200526143007579.png
[20200526155733274.png]: https://img-blog.csdnimg.cn/20200526155733274.png
[20200526195547326.png]: https://img-blog.csdnimg.cn/20200526195547326.png
[20200526195700274.png]: https://img-blog.csdnimg.cn/20200526195700274.png
[20200526203409965.png]: https://img-blog.csdnimg.cn/20200526203409965.png