7.20

一、座右铭

我的故事你说，我的文字我落，我值几两你定，我去何方我挑。

二、HDFS宕机之后的副本数的问题

HDFS中DataNode如果宕机之后，NameNode会根据当前宕机的DataNode上存储的Block块的副本去找一个存活的DataNode进行复制。 DataNode宕机恢复以后，此时整个集群中宕机节点存储的Block块的副本超过设置的副本数，会把这些副本都保留了，因此HDFS集群有个规则，block块的最大副本数512副本，除非执行start-balancer.sh命令

三、MapReduce的工作流程（简单版本）

1. 客户端在执行MR程序时，客户端先根据设置的InputFormat实现类去对输入的数据文件进行切片（getSplits），如果没有设置InputFormat实现类，MR程序会使用默认的实现类（TextInputFormat–>FileInputFormat的子类）进行切片规划，生成一个切片规划文件
2. 客户端的切片规划文件生成以后，客户端还会把整个MR程序的配置项（Configuration配置），会封装成为一个job.xml文件，同时还会把MR程序的代码包括job.xml文件、切片规划文件提交给资源调度器（YARN/windowsCPU）,资源调度器会先分配资源启动MRAPPMaster的进程
3. MRAPPMaster会根据切片规划的切片个数，向资源调度器申请资源启动对应个数的MapTask任务区运行Mapper阶段的计算逻辑
4. MapTask启动成功以后会根据切片规划，借助指定的InputFormat的实现类中createRecoder方法去对应的切片中读取k-v数据，然后交给map方法做处理
5. map方法将切片的k-v数据处理完成，会k-v数据写到一个内存缓冲区中(100M)，如果内存缓冲区超过容量的80%,会溢写磁盘，溢写磁盘的时候会根据map输出的key值进行排序，通过还会根据指定的Partitioner分区机制进行分区。溢写文件可能会存在多个，等map阶段执行完成，每一个MapTask对应的多个溢写文件以及缓冲区中还没有溢写的数据整体会进行一次合并，形成一个最终的大文件（分区排序）
6. 紧跟着MRAPPMaster会向资源管理器申请资源启动ReduceTask,ReduceTask启动成功会从不同的MapTask的合并的大的溢写文件中去复制对应的分区的数据，ReduceTask会对所有复制过来的数据在进行一次排序
7. ReduceTask会对排好序的数据按照key进行分组，分好组之后一组相同的key值调用一次reduce方法进行计算，计算完成的数据会借助指定的OutputFormat类(没有指定，默认使用TextOutputFormat类–FileOutputFormat实现子类)将key-value数据写出到最终的结果文件中part-r-xxxxx

四、Hadoop的序列化问题

1. MapReduce程序运行中，Mapper阶段和Reducer阶段的输入和输出都是以key-value的格式进行的。同时Mapper和Reducer阶段的任务中需要的数据可能会跨网络或者跨节点传输，因此我们就要求，MR程序运行过程中所有的输入和输出的数据必须都得是可以被序列化的。
2. Java本身是有序列化机制的，但是Java本身的序列化机制比较笨重的，因此Hadoop建立一套全新的比较轻量级的序列化机制Writable
3. Hadoop的提供的序列化接口有两个
   1. Writable：自定义的数据如果只想当作MR程序的value使用，只需要实现Writable即可
   2. WritableComparable：自定义的数据如果想当作MR程序的key值来使用，那么必须实现WritableComparable接口 接口除了序列化和反序列化以外，还多了一个比较器的接口。 因此MR程序在整体运行过程中，可能会进行很多次的排序，排序都是基于key值来进行排序
4. Hadoop的常用序列化类

五、MR程序运行中InputFormat类的作用

5.1 作用主要有两个

1. 负责进行输入数据的切片 InputFormat抽象类提供了一个抽象方法getSplits\
2. 负责MapTask读取数据时，如何从切片从读取key-value数据，key-value的含义包括key-value的类型 InputFormat抽象类提供另外一个抽象方法createRecordReader

5.2 有一个核心实现类–抽象类FileInputFormat 当输入的数据是文件的时候

5.2.1 TextInputFormat：默认实现类

1. 编写MR程序的时候，如果没有指定InputFormat实现类，MR程序默认会使用TextInputFormat进行切片和k-v数据的读取
2. 切片机制1、每一个输入文件都需要单独切片，如果输入文件有N个，那么切片数量最少有N个2、每一个文件切片之前，首先需要先获取三个值：maxSize minSize blockSize，然后基于这三个值计算一个SplitSize=Math.max(minSize,Math.min(maxSize,blocksize));3、切片的核心：先判断这个文件能否被切割（文件是不是压缩文件 gz zip）,不能被切割，文件单独成为一个切片，如果能被切割，判断文件的长度是否大于splitsize的1.1倍，如果不大于，文件单独成为一个切片，如果大于的话，先按照splitsize切一片，然后判断文件剩余的容量和splitsize的倍数关系。
3. k-v数据的读取机制TextInputFormat读取切片数据是按行读取，一行一行读取的，每一行数据以行的偏移量为key，以每一行的数据为value进行读取行的偏移量指的是每一行的首字符在文件中的位置，位置是一个正整数，因此key是用LongWritable表示的，value因为代表的是每一行的数据，是个字符串，因此使用Text类型来表示

5.2.2 KeyValueTextInputFormat

1. 切片机制：和TextInputFormat的切片机制是一模一样的

2. K-V数据的读取规则

   1. 按照一行一行的读取数据，每一行的数据以指定的分隔符分割这一行的数据，以分割之后的第一个字符串当作key值，剩余的字符串当作value值 因此在这种机制下 key和value都是Text类型的

   2. 如果使用KeyValueTextInputFormat，我们需要指定一个行的分隔符，如果没有指定，那么默认的分隔符的

      \t conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator","分隔符")

5.2.3 NLineInputFormat

1. 切片机制

   1. 切片不是按照文件的大小和splitsize进行切片的，而是根据所有输入文件的行数进行切片的，每一个文件单独切片

   2. 使用NLineInputFormat的时候，需要指定切片的行数 NLineInputFormat.setNumLinesPerSplit(job,3)

   3. 3个文件 3n

      a.txt 10行 4

      b.txt 12行 4

      c.txt 10行 4

2. kv读取数据的机制

和TextInputFormat一模一样 以LongWirtable 每一行的偏移量为key 以Text每一行的数据为value进行读取

5.2.4 CombineTextInputFormat：使用频率相对比较高

1. 切片机制

   1. 适用于大量的小文件的场景

   2. 诞生背景

      1. 不管是TextInputFormat还是KeyValueTextInputFormat、还是NLineInputFormat，在进行切片的时候都是每一个文件单独进行切片，也就意味着，如果输入文件有n个，切片数最小有n个。
      2. 如果输入的文件都是一堆小文件，每一个文件只有几百kb,如果使用上述的切片机制，会产生很多的小切片，每一个切片就撑死几百KB，然后我们还得需要启动N个maptask运行。这就浪费资源了。大数据中，资源可是非常宝贵的东西，浪费可耻。 MR程序一般情况MapTask处理的切片一般最好都在几百M左右，这样才不浪费资源。
      3. 这个ConbineTextInputFormat使用大量小文件的切片规划，进行切片的时候，不是一个文件单独切片，而是根据容量进行切片，可能在一个切片中包含很多个小文件

   3. 切片规则

      1. ConbineTextInputFormat进行切片之前，需要指定一个容量–虚拟的切片容量（可以理解为切片容量）

         CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);//b

      2. 切片的时候，每一个文件先按照虚拟的切片容量进行一次虚拟切片，虚拟切片机制如下：每一个文件判断，如果小于虚拟的切片容量，那么成为一个虚拟切片，如果文件大于虚拟切片容量但是小于虚拟切片的2倍，那么文件平均划分为两个虚拟切片，如果文件大于虚拟切片的2倍，那么按照虚拟切片的大小切一片，剩余的容量继续上述的判断 假如指定虚拟切片容量是100M

         	虚拟切片
         a.txt 4.1M	4.1M
         b.txt 3M	3M
         c.txt 10M	10M

      3. 将文件虚拟切片完成以后，我们将虚拟切片按照顺序累加起来，如果累加起来的容量大于设置的虚拟切片容量，单独成为一个物理切片，如果不大于的话，那么继续累加下一个切片，直到累加的结果大于等于设置的虚拟切片容量。

         虚拟切片：4.1M 3M 10M

         物理切片：17.1M

2. KV数据的读取机制

   1. 和TextInputFormat一模一样的
   2. key LongWritable类型的 每一行的偏移量
   3. value Text类型 每一行的数据

5.2.5 SequenceFileInputFormat——不讲

5.3 MR程序的Job提交流程的源码阅读

1. MR程序在运行的时候，首先先计算MR程序的输入文件的切片（根据指定的InputFormat实现类），生成一个切片规划文件job.split,随后会在根据MR程序的相关Configuration的配置生成一个配置文件job.xml，然后将job.xml、job.split、job.jar(编写好的MR程序)提供给资源调度器
2. 我们MR程序运行的时候，我们目前是在windows上运行的，并不在大数据集群环境中，因此windows上的运行严格意义上不属于分布式运行，只是Windows模拟MR的运行环境，因此如果在windows上运行的话，MR程序运行的时候只需要提供job.split、job.xml文件给本地的一个目录
3. 同时MR程序一般在windows上去属于测试运行（看看代码有没有bug的），如果代码没有bug,我们一般在企业中都是这样运行MR程序的，将MR程序打成一个jar包，然后上传到Hadoop集群的节点上，然后通过hadoop jar xxx.jar xxx,xxDriver方式运行MR程序，这样运行的MR程序是在YARN之上运行的，此时MR程序才属于真正的分布式运行，同时如果是在YARN上运行的话，job提交job.split、job.xml、job.jar文件到HDFS上。

5.4 自定义InputFormat(自己定义切片机制以及KV数据的读取规则)

1. 自定义一个类继承InputFormat
2. 重写getSplits方法
3. 重写createRecordReader方法

六、MR项目创建使用的细节问题

6.1 创建时需要导入的依赖以及相关配置性问题

1. 导入的依赖
   1. hadoop-client
   2. hadoop-hdfs
   3. slf4j-log4j12：查看MR程序的运行日志
2. 还需要在resources目录下引入一个log4j.properties文件，文件查看日志
3. 同时还可以在resources目录引入Hadoop的相关配置文件：core-site.xml hdfs-site.xml yarn-site.xml mapred-site.xml 如果引入这些配置文件，那么MR程序在运行的时候，配置文件生效的范围： Configuration配置文件对象----->resources目录下引入配置文件----->大数据环境下配置的配置文件（MR程序必须运行在大数据集群中，而非windows上，如果是在windows上运行，那么使用的默认配置）

6.2 MR项目的打包在Hadoop集群运行

6.2.1 概念

在windows上只是测试运行的，使用的环境不是大数据环境，因此无法做到分布式运行，如果真的想让MR程序分布式运行，我们需要将本地编写好的MR程序打成一个jar包，上传到Hadoop集群的某个节点，然后使用 hadoop jar xxx.jar xxx.xxxDriver 运行MR程序，

6.2.2 windows的idea打jar包有两种方式:

1. 自己手动生成jar包

   file--->project structure---->artifacts--->+--->jar

2. 借助maven自动化构建工具生成jar包

   1. 原理：maven是一个自动化构建工具，maven工具除了可以帮助我们自动引入第三方编程依赖以外，他还有一个最核心最重要的功能：帮助进行项目的自动化构建管理。
   2. maven的生命周期：maven用来管理项目的编译、测试和打包的

      1. 

      2. 如果只运行后面的后面的周期，前面的生命周期也会自动触发

      3. 如果前面的生命周期运行失败，那么后面的运行周期就无法执行

   3. maven每一个生命周期之所以帮助我们做对应的操作，是因为maven底层有一些插件，点击对应的生命周期时，调用底层的默认插件帮助我们完成操作，如果插件打包出现的效果不是我们需要的，那么我们就可以把maven生命周期对应的插件给替换了即可。

6.2.3 【注意】如果我们需要在Hadoop集群上运行，那么必须启动YARN

七、MR程序运行中Mapper阶段的作用

1. mapper阶段是MR程序运行的一个核心阶段，提供一个map方法，这个map方法会借助Inputformat提供的createRecordReader方法进行对应切片的key-value数据的读取,map方法每读取一个k-v数据处理一次key-value,输出一个结果到MR的一个内存缓冲区当中。
2. Mapper阶段启动多个MapTask任务，MapTask的任务个数和切片个数是对应的

八、Map程序运行中Shuffle阶段的学习

1. shuffle是大数据分布式计算的一个核心，也是大数据分布式计算性能受影响的核心，shuffle又名重新洗牌（将数据重新打乱，然后不同节点上和网络中进行数据的传输）。

2. MapReduce程序中，MR的分布式计算程序的shuffle出现在map执行之后，reduce任务执行之前。

3. Shuffle阶段工作的详细流程

九、代码示例

package com.sxuek.demo;

import org.apache.hadoop.conf.Configuration;

public class DemoDriver {
    public static void main(String[] args) {
        /**
         * MR程序运行的时候，配置可以三个地方配置：
         * 1、Configuration配置
         * 2、在resources目录将Hadoop的配置文件放进去 配置文件也可以配置
         * 3、Hadoop集群上进行配置（必须保证代码是在Hadoop集群节点运行的，否则3使用的是默认配置）
         *
         * 1的配置会覆盖2的相同配置 2的相同配置会覆盖3的相同配置
         */
        Configuration configuration = new Configuration();
//        configuration.set("fs.defaultFS","hdfs://aaaaa:9000");
        String s = configuration.get("fs.defaultFS");
        System.out.println(s);
    }
}

package com.sxuek.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class WCDriver {
    public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException, ClassNotFoundException {
        //1、准备一个配置文件对象
        Configuration configuration = new Configuration();
        //在配置文件指定HDFS的地址，因此MR处理的数据一般都是HDFS的，但是我这里不指定了，因此在resources目录下已经通过core-site.xml文件指定了
//        configuration.set(KeyValueLineRecordReader.KEY_VALUE_SEPARATOR," ");

        //2、创建一个封装MR程序使用Job对象
        Job job = Job.getInstance(configuration);

        /**
         * 3、封装指定的InputFormat类 如果没有指定，默认使用TextInputFormat
         */
//        job.setInputFormatClass(KeyValueTextInputFormat.class);
//        job.setInputFormatClass(NLineInputFormat.class);
//        NLineInputFormat.setNumLinesPerSplit(job,3);//指定3行做一个切片

        job.setInputFormatClass(CombineTextInputFormat.class);
        CombineTextInputFormat.setMaxInputSplitSize(job,8);
        //指定输入文件路径  输入路径默认是本地的，如果你想要是HDFS上的 那么必须配置fs.defaultFS  指定HDFS的路径
        FileInputFormat.setInputPaths(job,new Path("/demo"));

        /**
         * 4、封装Mapper阶段
         */
        job.setMapperClass(WCMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        /**
         * 5、封装Partitioner分区类，分区类可以不用指定，默认分区机制
         */
//        job.setPartitionerClass();

        /**
         * 6、封装Reducer阶段
         */
        job.setReducerClass(WCReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        job.setNumReduceTasks(1);

        /**
         * 7、封装指定的OutputFormat,如果没有指定OutputFormat  默认使用TextOutputFormat
         */
//        job.setOutputFormatClass();
        //封装输出路径  输出路径不能提前存在，因此代码在中先判断是否存在，如果存在删除了
        Path path = new Path("/output");
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.68.101:9000"), configuration, "root");
        if (fs.exists(path)){
            fs.delete(path,true);
        }
        FileOutputFormat.setOutputPath(job,path);

        /**
         * 8、提交程序运行
         *    提交的时候先进行切片规划，然后将配置和代码提交给资源调度器
         */
        boolean b = job.waitForCompletion(true);
        System.exit(b?0:1);
    }
}


class WCMapper extends Mapper<LongWritable, Text,Text,LongWritable>{
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        System.out.println("map通过inoutformat机制读取的key值为"+key.get()+",读取的value值为"+line);
        String[] words = line.split(" ");
        for (String word : words) {
            context.write(new Text(word),new LongWritable(1L));
        }
    }
}
 class WCReducer extends Reducer<Text,LongWritable,Text,LongWritable>{
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        long sum =0l;
        for (LongWritable value : values) {
            sum += value.get();
        }
        context.write(key,new LongWritable(sum));
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!--
  Licensed under the Apache License, Version 2.0 (the "License");
  you may not use this file except in compliance with the License.
  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License. See accompanying LICENSE file.
-->

<!-- Put site-specific property overrides in this file. -->

<configuration>
   <!-- 指定HDFS中NameNode的地址 -->
	<property>
  		<name>fs.defaultFS</name>
  		<value>hdfs://192.168.68.101:9000</value>
	</property>
<!-- 指定hadoop运行时产生文件的存储目录  HDFS相关文件存放地址-->
	<property>
	 	 <name>hadoop.tmp.dir</name>
  		<value>/opt/app/hadoop-3.1.4/metaData</value>
	</property>
</configuration>

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