Hadoop学习之深入MapReduce

• 什么是MapReduce
• MapReduce执行原理
• • Map阶段
  • Reduce阶段
• MapReduce查看日志
• • 方法一：标准输出
  • 方法二：logger输出
  • 命令三：命令行查询
• 停止Hadoop集群中的任务
• 代码
• • Java代码
  • pom文件
• 参考文献

MapReduce是Google提出的一个软件架构，用于大规模数据集（大于1TB）的并行运算。
MapReduce是分布式运行的，由两个阶段组成：Map（映射）”和“Reduce（归纳）
MapReduce的计算思想可以用四个字概括，即并行计算。将获得的局部聚合的临时结果统一汇总，快速得到答案，这就是MapReduce的计算思想。
分布式计算的步骤：
第一步：对每个节点上面的数据进行局部计算
第二步：对每个节点上面计算的局部结果进行最终全局汇总

左下角是一个文件，文件最下面是几个block块，说明这个文件被切分成了这几个block块，文件上面是一些split，注意，每一个split产生一个map任务。

map和reduce在代码层面对应的就是两个类，map对应的是mapper类，reduce对应的是reducer类，下面我们就来根据一个案例具体分析一下这两个步骤。

假设文件hello.txt内有如下两行内容：
hello you
hell me
现利用MapReduce计算各单词出现的次数

第一步
默认情况下，每一行数据，都会被解析成一个 ，这里的k1是指每一行的起始偏移量，v1代表的是那一行内容，
所以，针对文件中的数据，经过map处理之后的结果是这样的

<0，hello you>
<10，hello me>

【注意】map每次只会读取一行数据，所以第一次执行会产生<0，hello you>，第二次执行会产生<10，hello me>，并不是执行一次就获取到这两行结果了。
第二步
框架调用Mapper类中的map(…)函数，map函数的输入是，输出是。
一个InputSplit对应一个map task。程序员需要自己覆盖Mapper类中的map函数，实现具体的业务逻辑。
因为我们需要统计文件中每个单词出现的总次数，所以需要先把每一行内容中的单词切开，然后记录出现次数为1。

针对<0，hello you>，进行切割记录得到：


针对<10，hello you>，进行切割记录得到：

第三步
框架对map函数输出的进行分区。不同分区中的由不同的reduce task处理，默认只有1个分区，所以所有的数据都在一个分区，最后只会产生一个reduce task。
经过这个步骤之后，数据没什么变化，如果有多个分区的话，需要把这些数据根据分区规则分开，在这里默认只有1个分区。

第四步
框架对每个分区中的数据，都会按照k2进行排序、分组。分组指的是相同k2的v2分成一个组，
先按照k2排序

然后按照k2进行分组，把相同k2的v2分成一个组

第五步
在map阶段，框架可以选择执行Combiner过程，可先对在map端对单词出现的次数进行局部求和操作，这样就可以减少map端到reduce端数据传输的大小，这就是Combiner的好处，但针对求平均值之类的操作就不能使用规约了，否则最终计算的结果就不准确了。
【Combiner一个可选步骤，默认这个步骤是不执行的】

第六步
框架会把map task输出的写入到linux 的磁盘文件中

【注意】如果没有Reduce阶段，其实map阶段只需要执行到第二步就可以，第二步执行完成以后，结果就可以直接输出到HDFS了。

第一步
框架对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。这个过程称作shuffle
针对我们这个需求，只有一个分区，所以把数据拷贝到reduce端之后还是老样子

第二步
框架对reduce端接收的相同分区的数据进行合并、排序、分组。在Reduce中再次排序、分组是因为Map端是局部的操作，而Reduce端是全局的操作。
不过针对我们这个需求只有一个map任务一个分区，所以最终的结果还是老样子

第三步
框架调用Reducer类中的Reduce方法，reduce方法的输入是，输出是。一个调用一次reduce函数。程序员需要覆盖reduce函数，实现具体的业务逻辑。
在这里就需要在Reduce函数中实现最终的聚合计算操作了，将相同k2的{v2}累加求和，然后再转化为k3,v3写出去，在这里最终会调用三次reduce函数。

第四步
框架把reduce的输出结果保存到HDFS中。

hello 2
me 1
you 1

单文件的单词计数执行流程：

两个文件的单词计数执行流程：

在自定义mapper类或reducer类的map函数或reduce函数中增加一个输出，将或,的值打印出来

protected void map(LongWritable k1, Text v1, Context context)
        throws IOException, InterruptedException {
    //输出k1,v1的值
    System.out.println("=<"+k1.get()+","+v1.toString()+">");
    ......
    }
}
protected void reduce(Text k2, Iterable v2s, Context context)
        throws IOException, InterruptedException {
	  ......
    for (LongWritable v2 : v2s) {
       ......
          //输出k2,v2的值
        System.out.println("=<"+k2.toString()+","+v2.get()+">");
       ......
    }
 	  ......
    //输出k3,v3的值
    System.out.println("=<"+k3.toString()+","+v3.get()+">");
	   ......
}

在Windows机器上打jar包，并把新的jar包上传到机器的/data/soft/hadoop-3.2.0目录中
重新向集群提交任务。
示例：

[root@bigdata01 hadoop-3.2.0]#  hadoop jar db_hadoop-1.0-SNAPSHOT-jar-with-dependencies.jar com.bigdata.mr.WordCountJob /test/hello.txt /out1

【注意】针对输出目录，要么换一个新的不存在的目录，要么把之前的目录删掉。

先进入到yarn的web界面(http://[ ip地址 ]:8088/)，点击对应任务的history链接，注意此方法需要开启日志聚合功能才可成功查看，因为默认情况下任务的日志是散落在NodeManager节点上的，通过日志聚合功能我们可以把之前本来散落在NodeManager节点上的日志统一收集到hdfs上的指定目录中，这样就可以在yarn的web界面中直接查看了。
开启日志聚合功能需要修改yarn-site.xml的配置，增加yarn.log-aggregation-enable和yarn.log.server.url这两个参数。

     
        yarn.log-aggregation-enable  
        true
    
    
        yarn.log.server.url
        http://bigdata01:19888/jobhistory/logs/
    

【注意】修改这个配置想要生效，需要重启集群。

启动historyserver进程，需要在集群的所有节点上都启动这个进程

开启historyserver进程命令：bin/mapred --daemon start historyserver

重新再提交MapReduce任务。

使用sout输出的日志信息需要到Log Type: stdout这里来查看

public static class MyMapper extends Mapper{
    Logger logger = LoggerFactory.getLogger(MyMapper.class);   
    @Override
    protected void map(LongWritable k1, Text v1, Context context)
            throws IOException, InterruptedException {
        //输出k1,v1的值      
        logger.info("=<"+k1.get()+","+v1.toString()+">");
     ......
        }
    }
}

public static class MyReducer extends Reducer{

    Logger logger = LoggerFactory.getLogger(MyReducer.class);   
    
    protected void reduce(Text k2, Iterable v2s, Context context)
            throws IOException, InterruptedException {
                    ......
        for (LongWritable v2 : v2s) {
            ......
            //输出k2,v2的值      
            logger.info("=<"+k2.toString()+","+v2.get()+">");
           ......
        }        
        //输出k3,v3的值      
        logger.info("=<"+k3.toString()+","+v3.get()+">");
       
    }
}

yarn logs -applicationId application_1587713567839_0001

【注意】后面指定的是任务Id，任务Id可以到yarn的web界面上查看。
结合grep进行筛选日志信息

[root@bigdata01 hadoop-3.2.0]# yarn logs -applicationId application_1587713567839_0001 | grep k1,v1
=<0,hello you>
=<10,hello me>

针对某一些艰难的场景下，无法使用yarn的web界面查看日志，就需要使用yarn logs命令了。

使用yarn application -kill命令，后面指定任务id即可

 yarn application -kill application_1587713567839_0001

package com.bigdata.mr;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;


public class WordCountJobNoReduce {
    
    public static class MyMapper extends Mapper{
        Logger logger = LoggerFactory.getLogger(MyMapper.class);
        
        @Override
        protected void map(LongWritable k1, Text v1, Context context)
                throws IOException, InterruptedException {
            //输出k1,v1的值
            //System.out.println("=<"+k1.get()+","+v1.toString()+">");
            logger.info("=<"+k1.get()+","+v1.toString()+">");
            //  k1代表的是每一行的行首偏移量，v1代表的是每一行内容
            //  对获取到的每一行数据进行切割，把单词切割出来
            String[] words = v1.toString().split(" ");
            //  迭代切割出来的单词数据
            for (String word:words) {
                //  把迭代出来的单词封装成的形式
                Text k2 = new Text(word);
                LongWritable v2 = new LongWritable(1L);
                //  把写出去
                context.write(k2,v2);
            }
        }
    }

    
    public static void main(String[] args) {
        try {
            if(args.length!=2){
                //  如果传递的参数不够，程序直接退出
                System.exit(100);
            }
            //  job需要的配置参数
            Configuration conf = new Configuration();
            //  创建一个job
            Job job = Job.getInstance(conf);
            //  注意：这一行必须设置，否则在集群中执行的是找不到WordCountJob这个类
            job.setJarByClass(WordCountJobNoReduce.class);
            //  指定输入路径(可以是文件，也可以是目录)
            FileInputFormat.setInputPaths(job,new Path(args[0]));
            //  指定输出路径(只能指定一个不存在的目录)
            FileOutputFormat.setOutputPath(job,new Path(args[1]));
            //  指定map相关的代码
            job.setMapperClass(MyMapper.class);
            //  指定k2的类型
            job.setMapOutputKeyClass(Text.class);
            //  指定v2的类型
            job.setMapOutputValueClass(LongWritable.class);
            //禁用reduce阶段
            job.setNumReduceTasks(0);
            //  提交job
            job.waitForCompletion(true);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

本实例使用的依赖如下

    4.0.0
    db_hadoop
    db_hadoop
    1.0-SNAPSHOT
    
        
            org.apache.hadoop
            hadoop-core
            1.1.0
        

    
        org.apache.hadoop
        hadoop-client
        3.2.0
        provided
    

    
        org.slf4j
        slf4j-api
        1.7.10
        provided
    
    
        org.slf4j
        slf4j-log4j12
        1.7.10
        provided
    

 
        
            
            
                org.apache.maven.plugins
                maven-compiler-plugin
                2.3.2
                
                    UTF-8
                    1.8
                    1.8
                    true
                
            
            
                maven-assembly-plugin
                
                    
                        jar-with-dependencies
                    
                    
                        
                            
                        
                    
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                
            
        
    

https://www.imooc.com/wiki/BigData：慕课网《大数据开发工程师体系课程》