Hadoop之MapReduce入门

喜欢ヅ旅行 2022-01-21 11:55 381阅读 0赞

# Hadoop之MapReduce概述 #

--------------------

## 目录 ##

1.  MapReduce定义
2.  MapReduce优缺点
3.  MapReduce核心思想
4.  MapReduce进程
5.  MapReduce编程规范
6.  MapReduce案例实操

--------------------

### 1. MapReduce定义 ###

1.  Mapreduce是一个分布式运算程序的编程框架，是用户开发“基于hadoop的数据分析应用”的核心框架。
2.  Mapreduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个hadoop集群上。

--------------------

### 2. MapReduce优缺点 ###

优点

1.  MapReduce 易于编程  
    它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的PC机器上运行。也就是说你写一个分布式程序，跟写一个简单的串行程序是一模一样的。就是因为这个特点使得MapReduce编程变得非常流行。
2.  良好的扩展性  
    当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。
3.  高容错性  
    MapReduce设计的初衷就是使程序能够部署在廉价的PC机器上，这就要求它具有很高的容错性。比如其中一台机器挂了，它可以把上面的计算任务转移到另外一个节点上运行，不至于这个任务运行失败，而且这个过程不需要人工参与，而完全是由Hadoop内部完成的。
4.  适合PB级以上海量数据的离线处理  
    这里加红字体离线处理，说明它适合离线处理而不适合在线处理。比如像毫秒级别的返回一个结果，MapReduce很难做到。

缺点  
MapReduce不擅长做实时计算、流式计算、DAG（有向图）计算。

1.  实时计算  
    MapReduce无法像Mysql一样，在毫秒或者秒级内返回结果。
2.  流式计算  
    流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化。这是因为MapReduce自身的设计特点决定了数据源必须是静态的。
3.  DAG（有向图）计算  
    多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

--------------------

### 3. MapReduce核心思想 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70]

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

--------------------

### 4. MapReduce进程 ###

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

1.  MrAppMaster：负责整个程序的过程调度及状态协调。
2.  MapTask：负责map阶段的整个数据处理流程。
3.  ReduceTask：负责reduce阶段的整个数据处理流程。

--------------------

### 5. MapReduce编程规范 ###

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

1.  Mapper阶段（k-long首字母的偏移量，v-string 记录一行的数据）
    
    1.  用户自定义的Mapper要继承自己的父类
    2.  Mapper的输入数据是KV对的形式（KV的类型可自定义）
    3.  Mapper中的业务逻辑写在map()方法中
    4.  Mapper的输出数据是KV对的形式（KV的类型可自定义）
    5.  map()方法（maptask进程）对每一个<K,V>调用一次
2.  Reducer阶段
    
    1.  用户自定义的Reducer要继承自己的父类
    2.  Reducer的输入数据类型对应Mapper的输出数据类型，也是KV
    3.  Reducer的业务逻辑写在reduce()方法中
    4.  Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法
3.  Driver阶段  
    相当于yarn集群的客户端，用于提交我们整个程序到yarn集群，提交的是封装了mapreduce程序相关运行参数的job对象

--------------------

### 6. MapReduce案例实操 ###

WordCount案例实操

1.  需求  
    在给定的文本文件中统计输出每一个单词出现的总次数
2.  数据准备(hello.txt)

hello world
    atguigu atguigu
    hadoop 
    spark
    hello world
    atguigu atguigu
    hadoop 
    spark
    hello world
    atguigu atguigu
    hadoop 
    spark

1.  分析  
    按照mapreduce编程规范，分别编写Mapper，Reducer，Driver，如下图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70 1]  
4. 导入相应的依赖坐标+日志添加

<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.  编写程序
    
    1.  编写mapper类

import java.io.IOException;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    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(" ");
    		
    		// 3 输出
    		for (String word : words) {
    			
    			k.set(word);
    			context.write(k, v);
    		}
    
    	}
    }

1.  编写reducer类

import java.io.IOException;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
    
    int sum;
    IntWritable v = new IntWritable();
    
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> value,
    			Context context) throws IOException, InterruptedException {
    		
    		// 1 累加求和
    		sum = 0;
    		for (IntWritable count : value) {
    			sum += count.get();
    		}
    		
    		// 2 输出
            v.set(sum);
    		context.write(key,v);
    	}
    }

1.  编写驱动类

import java.io.IOException;
    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;
    
    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 设置Reduce输出
    		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);
    	}
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70]: /images/20220121/68e6c15fbc49426b855e89d3cd64dc90.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkxMDY5NA_size_16_color_FFFFFF_t_70 1]: /images/20220121/1b112396a7c849ec8df84355fd24a99e.png