【Hadoop】Hadoop生态系列之MapReduce概述及MapReduce任务开发与发布

我就是我 2023-10-05 10:47 20阅读 0赞

上一篇：[Hadoop生态系列之HDFS常用Shell命令实践及Java API操作HDFS][Hadoop_HDFS_Shell_Java API_HDFS]

#### 这里写目录标题 ####

*   *  MapReduce
     *   *  概述
         *  流程分析
         *  环境搭建
         *  MapReduce任务开发
         *   *  背景
             *  实现
             *  任务发布
             *   *  远程部署
                 *  本地仿真
                 *  跨平台提交
                 *  本地仿真
                 *  跨平台提交

### MapReduce ###

#### 概述 ####

MapReduce是一个 Hadoop 的并行计算框架，借鉴了函数式编程思想和矢量编程。Hadoop 中是充分利用了存储节点/Data Node运行所在主机的计算资源（CPU、内存、网络、少许磁盘）完成对任务的并行计算。Map Reduce框架会在所有的DataNode所在的物理主机启动一个计算资源管理者Node Manager用于管理本地的计算资源，默认系统会将计算资源均分8个等份，每个等份抽象成一个Container，该Container主要作为资源隔离。还会再找一些其他的主机启动一个资源管理中心Resource Manager，用于管理集群的计算资源。

#### 流程分析 ####

当用户提交一个计算任务给MapReduce框架，框架会将任务拆分成Map阶段和Reduce阶段（矢量编程思想将任务拆分成两个阶段），框架会根据Map/Reduce阶段的任务并行度.在任务提交初期会启动一个任务管理者（每个任务都有自己的任务管理者）-MRAppMaster（该进程会浪费掉1个计算资源）用于管理Map阶段和Reduce阶段任务执行。在任务执行时期，每个阶段会根据阶段任务的并行度分配计算资源（每个计算资源启动一个Yarn Child），由MRAppMaster完成对阶段任务的检测管理。

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bGfu5Sbp-1601292732842)(assets\\1601263950874.png)\]][img-bGfu5Sbp-1601292732842_assets_1601263950874.png]  
上图可概括为以下11步：

> 1.  首先需要编写并运行程序；
> 2.  客户端client联系资源管理器ResourceManager获取计算任务ID；
> 3.  客户端向HDFS提交任务资源
> 4.  客户端向ResourceMans提交计算任务
> 5.  ResourceManager在DataNode上随机选择一个NodeManager服务，启动一个Container --> MRAppMaster
> 6.  MRAppMaster初始化job，用于监测任务的执行对象
> 7.  MRAppMaster从HDFS查询输入的切片
> 8.  MRAppMaster连接资源管理器申请Container
> 9.  MRAppMaster启动Container，执行YarnChild进程
> 10. YarnChild到HDFS上下载资源
> 11. 运行Map任务，即运行代码片段

`ResourceManager`:负责任务资源的统一调度，管理NodeManager资源，启动MRAppMaster

`NodeManager`:用于管理本机上的计算资源，默认会将本机的计算资源拆分为8个等份，每个等份抽象成Container

`MRAppMaster`:任何一个执行的任务都会有一个MRAppMaster负责YarnChild任务的执行和监测。

`YarnChild`:是具体执行的MapTask或者是ReduceTask的统称。

任务执行期间系统会启动MRAppmaster和YarnChild负责任务的执行，一旦任务执行结束MRAppMaster和YarnChild会自动退出。

#### 环境搭建 ####

①配置资源管理器

[root@CentOS ~]# vi /usr/hadoop-2.9.2/etc/hadoop/yarn-site.xml

<property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
    </property>
    
    <property>
        <name>yarn.resourcemanager.hostname</name>
        <value>CentOS</value>
    </property>
    
    <property>
        <name>yarn.nodemanager.pmem-check-enabled</name>
        <value>false</value>
    </property>
    
    <property>
        <name>yarn.nodemanager.vmem-check-enabled</name>
        <value>false</value>
    </property>

②配置MapReduce计算框架

[root@CentOS ~]# mv /usr/hadoop-2.9.2/etc/hadoop/mapred-site.xml.template  /usr/hadoop-2.9.2/etc/hadoop/mapred-site.xml
    [root@CentOS ~]# vi /usr/hadoop-2.9.2/etc/hadoop/mapred-site.xml

<property>
            <name>mapreduce.framework.name</name>
            <value>yarn</value>
    </property>

③启动计算服务

[root@CentOS ~]# start-yarn.sh
    starting yarn daemons
    starting resourcemanager, logging to /usr/hadoop-2.9.2/logs/yarn-root-resourcemanager-CentOS.out
    CentOS: starting nodemanager, logging to /usr/hadoop-2.9.2/logs/yarn-root-nodemanager-CentOS.out
    [root@CentOS ~]# jps
    13078 SecondaryNameNode
    12824 DataNode
    1080 ResourceManager
    12681 NameNode
    1195 NodeManager
    1262 Jps

④可以访问ResourManager内嵌WebUI页面:http://CentOS:8088

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Y3PWPSuq-1601292732844)(assets\\1601264990143.png)\]

#### MapReduce任务开发 ####

##### 背景 #####

假设我们有如下的一张表，需要统计出每个版块被点击次数。

<table> 
 <thead> 
  <tr> 
   <th>日志级别</th> 
   <th>类别</th> 
   <th>点击日期</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>INFO</td> 
   <td>/product/xxxx1</td> 
   <td>2020-09-28 10:10:00</td> 
  </tr> 
  <tr> 
   <td>INFO</td> 
   <td>/product/xxxx2</td> 
   <td>2020-09-28 12:10:00</td> 
  </tr> 
  <tr> 
   <td>INFO</td> 
   <td>/cart/xxxx2</td> 
   <td>2020-09-28 12:10:00</td> 
  </tr> 
  <tr> 
   <td>INFO</td> 
   <td>/order/xxxx</td> 
   <td>2020-09-28 12:10:00</td> 
  </tr> 
 </tbody> 
</table>

如果我们可以将以上的日志看做成是数据库中的一张表，这个问题就可以使用以下SQL解决：

select category,sum(1) from t_click group by category

如果使用上面提到的MapReduce计算模型，我们可以使用Map完成group的功能，使用Reduce完成sum的功能。有如下数据格式

INFO /product/xxx/1?name=zhangsan 2020-09-28 10:10:00
    INFO /product/xxx/1?name=zhangsan 2020-09-28 10:10:00
    INFO /cart/xxx/1?name=lisi 2020-09-28 10:10:00
    INFO /order/xxx/1?name=zhangsan 2020-09-28 10:10:00
    INFO /product/xxx/1?name=zhaoliu 2020-09-28 10:10:00
    INFO /cart/xxx/1?name=win7 2020-09-28 10:10:00

##### 实现 #####

①写Mapper逻辑

package com.Dora.click;
    
    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;
    
    /**
     * 1、用户必须首先清楚 数据格式、存储位置  -- 读取数据方式   --  Mapper写法?
     *    TextInputFormat<LongWritable,Text>  : 读取文件系统中文件 本地系统、HDFS
     *                    字节偏移量     文本行
     * 2、必须清楚的知道自己想干嘛？ 按 __类别__ 统计  __点击次数__ 值
     *                               key           value
     */
    public class URLMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
        
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        
             String line=value.toString();
             String url=line.split(" ")[1];
             //获取类别
             int endIndex=url.indexOf("/",1);
             String category=url.substring(0,endIndex);
    
             //将转换的结果输出
            context.write(new Text(category),new IntWritable(1));
        }
    
    }

②Reducer逻辑

package com.Dora.click;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    import java.util.Iterator;
    
    /**
     * 1、你要对哪种Mapper的输出结果汇总  --  决定Reducer输入的Key和Value类型
     *
     * 2、你要知道最终输出结果以哪种格式写出去,输出Key/Value格式用户只需要关注他的toString即可
     *
     *   TextOutputFormat<key,value>  将结果写出到文件系统：本地、HDFS
     */
    public class URLReducer extends Reducer<Text, IntWritable,Text,IntWritable> {
        
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        
            int total=0;
            for (IntWritable value : values) {
        
                total+=value.get();
            }
            context.write(key,new IntWritable(total));
        }
    }

③封装Job对象

package com.Dora.click;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    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.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    public class URLCountApplication extends Configured implements Tool {
        
        public int run(String[] strings) throws Exception {
        
            //1.创建一个Job对象
            Configuration conf = getConf();
            Job job= Job.getInstance(conf,"URLCountApplication");
    
            //2.告诉job数据格式
            job.setInputFormatClass(TextInputFormat.class);
            job.setOutputFormatClass(TextOutputFormat.class);
    
            //3.设置数据路径
            TextInputFormat.addInputPath(job,new Path("/demo/click"));
            //系统自动创建，如果在执行前存在，则放弃执行
            TextOutputFormat.setOutputPath(job,new Path("/demo/result"));
    
            //4.设置处理逻辑
            job.setMapperClass(URLMapper.class);
            job.setReducerClass(URLReducer.class);
    
            //5.设置输出的Key,value
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            //6.提交job
            return job.waitForCompletion(true)?1:0;
        }
        
        public static void main(String[] args) throws Exception {
        
            ToolRunner.run(new URLCountApplication(),args);
        }
    }

##### 任务发布 #####

###### 远程部署 ######

*  需要在job任务当中添加如下代码

job.setJarByClass(URLCountApplication.class);

设置程序的类加载路径，因为任务是打好jar包以后使用`hadoop jar`命令提交的.

[root@CentOS ~]# yarn jar MapReduce-1.0-SNAPSHOT.jar com.Dora.click.URLCountApplication
    或者
    [root@CentOS ~]# hadoop jar MapReduce-1.0-SNAPSHOT.jar com.Dora.click.URLCountApplication

**tips**:如果大家感觉这种打包，然后提交比较复杂，我们可以使用maven提供的ssh远程登录插件，先登录系统后自动执行后续提交任务。

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>org.example</groupId>
        <artifactId>MapReduce</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>2.9.2</version>
            </dependency>
        </dependencies>
        <build>
            <extensions>
                <extension>
                    <groupId>org.apache.maven.wagon</groupId>
                    <artifactId>wagon-ssh</artifactId>
                    <version>2.10</version>
                </extension>
            </extensions>
            <plugins>
                <plugin>
                    <groupId>org.codehaus.mojo</groupId>
                    <artifactId>wagon-maven-plugin</artifactId>
                    <version>1.0</version>
                    <executions>
                        <execution>
                            <id>upload-deploy</id>
                            
                            <phase>package</phase>
                            <goals>
                                <goal>upload-single</goal>
                                <goal>sshexec</goal>
                            </goals>
                            <configuration>
                                
                                <fromFile>target/${project.artifactId}-${project.version}.jar</fromFile>
                                
                                <url>
                                    <![CDATA[ scp://root:123456@CentOS/root/ ]]>
                                </url>
                                
                                <commands>
                                    <command> hadoop fs -rm -r -f /demo/result </command>
                                    <command> hadoop jar MapReduce-1.0-SNAPSHOT.jar com.Dora.click.URLCountApplication </command>
                                </commands>
                                <displayCommandOutputs>true</displayCommandOutputs>
                            </configuration>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    </project>

###### 本地仿真 ######

无需任何yarn环境，直接通过本地仿真的方式实现.一般需要更改NativeIO源码。这里面由于下载不到2.9.2的源码包，大家可以尝试使用2.6.0代替，将源码的557行代码修改如下：

public static boolean access(String path, AccessRight desiredAccess)
        throws IOException {
        
        return true;
    }

在resource资源目录下添加log4j.proeprties

log4j.rootLogger=INFO,CONSOLE
    log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender 
    log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout 
    log4j.appender.CONSOLE.layout.ConversionPattern=%p %d{yyyy-MM-dd HH:mm:ss,SSS} %C -%m%n

###### 跨平台提交 ######

①将core|hdfs|yarn|mapred-site.xml拷贝到项目的resources目录下

②在mapred-site.xml添加如下配置

<property>
        <name>mapreduce.app-submission.cross-platform	</name>
        <value>true</value>
    </property>

③修改job代码

conf.addResource("core-site.xml");
    conf.addResource("hdfs-site.xml");
    conf.addResource("yarn-site.xml");
    conf.addResource("mapred-site.xml");
    conf.set("mapreduce.job.jar","file:///xxx.jar");

</executions>
            </plugin>
        </plugins>
    </build>

\`\`\`

###### 本地仿真 ######

public static boolean access(String path, AccessRight desiredAccess)
        throws IOException {
        
        return true;
    }

在resource资源目录下添加log4j.proeprties

###### 跨平台提交 ######

①将core|hdfs|yarn|mapred-site.xml拷贝到项目的resources目录下

②在mapred-site.xml添加如下配置

<property>
        <name>mapreduce.app-submission.cross-platform	</name>
        <value>true</value>
    </property>

③修改job代码

下一篇：[Hadoop生态系列之InputForamt.class与OutputFormat.class分析][Hadoop_InputForamt.class_OutputFormat.class]

[Hadoop_HDFS_Shell_Java API_HDFS]: https://blog.csdn.net/qq_44664231/article/details/108855628
[img-bGfu5Sbp-1601292732842_assets_1601263950874.png]: https://img-blog.csdnimg.cn/20200929163038779.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0NjY0MjMx,size_16,color_FFFFFF,t_70#pic_center
[Hadoop_InputForamt.class_OutputFormat.class]: https://blog.csdn.net/qq_44664231/article/details/108874308