【Hadoop】面试复习知识总结

Hadoop是Apache开发的分布式、高可靠、可扩展的分布式系统基础架构。它的特点是：高可靠、可扩展、高效性、高容错性、低成本。

Hadoop Common：支持其他Hadoop模块的通用程序；
HDFS：分布式文件系统；
MapReduce：处理大量半结构化数据集合的编程模型；
Yarn：作业调度和集群资源管理的框架；
Ozone:基于HDFS进行对象的存储。

HDFS

• Namenode启动过程

1. 开启安全模式
 不能执行数据修改操作
2. 加载fsimage
3. 逐个执行所有Edits文件中的每一条操作将操作合并到fsimage，完成后生成一个空的edits文件
4. 接收datanode发送来的心跳消息和块信息
5. 根据以上信息确定文件系统状态
6. 退出安全模式

• 安全模式

namenode在启动期间，NameNode从fsimage和edits日志文件加载文件系统状态。然后，它等待
DataNodes报告其块，以便尽管群集中已存在足够的副本，但它不会过早地开始复制块。在此期
间，NameNode保持在Safemode中。NameNode的安全模式本质上是HDFS集群的只读模式，它不允
许对文件系统或块进行任何修改。

• 退出安全模式的条件

系统达到安全标准
•dfs.namenode.safemode.min.datanodes: 最小可用datanode数量
•dfs.namenode.safemode.threshold-pct: 副本数达到最小要求的block占系统总block数的百分比
• dfs.namenode.safemode.extension: 默认稳定时间30000ms=30s；

• Checkpoint

1. Secondarynamenode提醒namenode该合并元数据了；
 - 多久checkpoint?
	  默认3600s=60分钟=1小时            
	  100万条记录
  瓶颈：网络IO
2. Namenode日志滚动，在新的edits_inprogress中记录编辑操作；
3. 拷贝相关文件给Secondarynamenode；
   - 拷贝的什么文件？
	 正常情况下，拷贝edits文件;
	 初始化集群时拷贝fsimage;
	 手动checkpoint,拷贝namenode的fsimage,SN的edits会缺失。
4. 把相关文件加载进内存，生产新的fsimage;
5. 把新的fsimage写入磁盘；
6. 把新的fsimage传给namenode.

• HA Checkpoint

SBNN:standby namenode, ANN:active namenode
1.SBNN检查条件；
	>checkpoint时间间隔、100万条数据？
2.SBNN将内存中的状态保存为fsimage_txid,生成一个MD5文件,txid是edits最后一条事务的id；
3.SBNN向ANN发送Http get请求，包含SBNN域名和端口号以及txid;
4.ANN通过http请求获取最新的fsimage.

• 文件上传

1.客户端发送RPC请求上传某文件到某目录；
2.namenode检查，如有异常则IOException：
  -目录结构
  -Client权限
  -集群状态
  -租约检查
3.客户端文件切块，上传block01;
4.namenode分配相应的datanodes;
5.客户端与datanode建立管线，block上传到第一个datanode再传第二个、第三个；
  *传输过程中，其中一个datanode宕机了，怎么办？
  a.首先关闭管线；
  b.把所有数据包都添加回数据队列最前端；
  c.为储存正常的datanode制定新的标识，并将标识传送给namenode，以便日后恢复；
  d.namenode注意到复本不足时，会创建新复本。
6.datanodes全部储存完成后，namenode告诉client已经存储完了block01；
7.客户端继续上传block......

• 文件读取

1.客户端发送RPC请求给namenode读取文件；
2.namenode验证，文件是否存在，文件权限；
3.namenode将datanodes放在队列返回；
4.客户端选取最近的datanode读取数据；
5.对block进行checksum，如果数据损坏则报告给namenode，然后到其他datanode读取；
6.读取完后向namenode要后一块。都读取完后，客户端发送完毕信号，namenode关闭文件。

• 删除文件

1.客户端发送RPC请求删除文件；
2.namenode将这个操作记录在edits里，将数据从内存中删除，然后返回删除成功的消息给客户端；（实际数据还在datanode上）。
3.datanode向namenode发送心跳信息（节点状态、节点数据），namenode发现有的数据在元数据中没有记录，则命令datanode将其删除。

• 小文件问题

•定义:大量大小小于块大小的文件 
•实际场景:网页，Hive动态分区插入数据等 
•背景:每个文件的元数据对象约占150byte，所以如果有1千万个小文件，每个文件占用一个block，则NameNode大约需要2G空间。如果存储1亿个文件，则NameNode需要20G空间;数据以块为单位进行处理。 
•影响:占用资源，降低处理效率
•解决方案:
 •从源头减少小文件
 •使用archive打包
 •使用其他存储方式，如Hbase，ES等

• 常用命令

hadoop fs -put xxx.txt /xxx  上传文件
hadoop fs -mkdir /xxx  创建目录
hadoop fs -ls /xxx  查看
hadoop fs -cat a.txt 查看文件
hadoop fs -tail a.txt 查看文件最后1000字节

MapReduce

• 简述mapreduce全过程

1. 输入数据块;
2. 读取数据,切片（默认128mb，模糊值1.1）,InputFormat,LineRecorderReader一行一行读取给对应的maptask；
3. Maptask根据我们写的逻辑把数据进行映射；
4. 处理完数据：
1. 分区：
 根据Reduce数量进行分区
2. 排序：
 区内排序，按照key.compareTo()排序
3. 如果spill文件>=3,就combiner
5.将map结果写入磁盘；
6. reduce任务复制map输出，复制完后归并排序，combiner,再运行reduce处理逻辑，reduce结果写到hdfs.

• shuffle

系统执行排序、将map输出作为输入传给reduce的过程称为shuffle.
每个map任务都有一个环形内存缓冲区，默认大小100MB,缓存内容达到阈值0.80就开始spill溢出数据到磁盘。在spill过程中，map输出继续写入环形缓冲区，如果期间缓冲区被填满，map会被阻塞直至写磁盘过程完成。
（元数据包含partition,写入缓冲区，spill的时候按key排序，combiner,默认不压缩）
 为什么要用环形内存缓冲区？
  	便于写入缓冲区和写出同时进行
 为什么不写满再spill?
  	会出现写入缓冲区阻塞
Reduce端复制map输出，进行merge归并排序，combiner，调用reduce()函数，将结果输出到hdfs.

数据清洗案例

package com.hive.wash;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class DataWash {
    public static class DMap extends Mapper<LongWritable, Text,Text,IntWritable>{
        private String jobname;
        private String money;
        private String company;
        private String area;
        private String exprience;
        private String edu;
        private String mans;
        private String time;
        private String skill;
        private String profession;
        private String welfare;
        private String comptype;
        private String peoples;
        private String trade;


   //  运维工程师（微服务/Linux）,0.7-1万/月,南京壹证通信息科技有限公司,南京-雨花台区 , 5-7年经验 , 大专 , 招1人 , 09-25发布 , 计算机网络 计算机信息管理,,五险一金|餐饮补贴|带薪年假|周末双休|节日福利|定期体检|年终奖金,民营公司,50-150人,计算机软件 计算机服务(系统、数据服务、维修)
        /**
         * map清洗数据
         * anthor:bb
         */
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //读一行数据
            String[] words = value.toString().split("`");
            if(words.length<14){
                return;
            }
            jobname=words[0];
            money=words[1];
            company=words[2];
            area=words[3].replace(" ","");
            // 将字符 C2A0、制表符、换行、回车"\t|\r|\n|" +去掉
            // 下面并没有把空格去掉
            byte bytes[] = {(byte) 0xC2,(byte) 0xA0};
            String UTFSpace = new String(bytes,"utf-8");
            area = area.replace(UTFSpace,"");

            exprience=words[4].replace(" ","");
            edu=words[5].replace(" ","");
            mans=words[6];
            time=words[7].replace(" ","");
            skill=words[8].replace(" ","");
            profession=words[9].replace(" ","");
            welfare=words[10];
            comptype=words[11];
            peoples=words[12];
            trade=words[13];
            if(jobname.equals("")){
                return;
            }

            if(edu.contains("区")||edu.contains("司")||edu.contains("验")||edu.contains("null")){
                return;
            }

            Double min;
            Double max;
            if(money.equals("")||money.contains("面")){
                return;
            }else {
                if(money.length()<4){
                    return;
                }
                String danw=money.substring(money.length()-3,money.length());
                if("万/月".equals(danw)){
                    String xxs=money.substring(0,money.length()-3);
                    String[] split = xxs.split("-");
                    min=Double.parseDouble(split[0])*10000;
                    max=Double.parseDouble(split[1])*10000;
                }else if("千/月".equals(danw)){
                    String xxs=money.substring(0,money.length()-3);
                    String[] split = xxs.split("-");
                    min=Double.parseDouble(split[0])*1000;
                    max=Double.parseDouble(split[1])*1000;
                }else if("元/天".equals(danw)){
                    max=min=Double.parseDouble(money)*30;
                }else if("万/年".equals(danw)){
                    String xxs=money.substring(0,money.length()-3);
                    String[] split = xxs.split("-");
                    min=Double.parseDouble(split[0])*10000/12;
                    max=Double.parseDouble(split[1])*10000/12;
                }else {
                    return;
                }
            }

            String[] split = area.split("-");
            int s=split.length;
            String city;
            String qu;
            if(s==1){
                city=split[0];
                qu="null";
            }else if(s>=2){
                city=split[0];
                qu=split[1];
            }else {
                city="";
                qu="";
            }

            String exp="";
            if(exprience.contains("无")||exprience.equals("")){
                exp="0";
            }else {
                String reg="(\\d)";
                Pattern pattern=Pattern.compile(reg);
                Matcher matcher=pattern.matcher(exprience);
                if(matcher.find()){
                    exp=matcher.group(1);
                }
            }
            String man="";
            String reg="(\\d)";
            Pattern pattern=Pattern.compile(reg);
            Matcher matcher=pattern.matcher(mans);
            if(mans.contains("若干")){
                man="10";
            }else if(matcher.find()){
                man=matcher.group(1);
            }else {
                man="0";
            }
            if(skill.contains("语")||skill.contains("话")){

            }else {
                profession=skill;
                skill="";
            }
            String str=jobname+"`"+min+"`"+max+"`"+company+"`"+city+"`"+qu+"`"+exp+"`"+
                    edu+"`"+man+"`"+time+"`"+skill+"`"+profession+"`"+welfare+"`"+comptype+"`"+
                    peoples+"`"+trade;
            context.write(new Text(str),new IntWritable(1));
        }
    }


    public static class DMapReduce extends Reducer<Text,IntWritable,Text,Text>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            context.write(key,new Text(""));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "wash data");
        job.setJarByClass(DataWash.class);
        job.setMapperClass(DMap.class);
        job.setReducerClass(DMapReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

YARN

• MapReduce on Yarn任务调度

1：MapReduce向client提交任务，client将任务提交给ResourceManager，ResourceManager分为两个组件：资源管理（ResourceScheduler）和作业调度（ApplicationManager）
2：ApplicationManager收到作业后，会去查看哪一个NodeManager相对空间，就会在该NodeManager上分配一个Container，在Container中启动该作业的ApplicationMaster（AM）
3：ApplicationMaster首先会向ApplicationManager注册，这样用户就可以直接通过ResourceManager查询作业的状态
4：AM会将文件切分和任务切分，切分之后需要多少的Map和Reduce资源，就会向ResourceScheduler申请资源，ResourceScheduler分析哪些节点能分配任务，之后就会告诉ApplicationMaster去哪些节点申请资源
5：AM就会向NodeManager申请Container
6：NodeManager会分出Container来启动对应的Map Task或Reduce Task
7：Task受AM角色的监控，AM会轮询检查每个Container的作业进度，如果Reduce Task执行完了，就会告知AM
8：AM会上报RS中的ApplicationManager任务执行完了，ResourceManager会向ApplicationMaster关闭Container的命令，ApplicationMaster就会向Container发出关闭的命令，AM就会向RS注销并关闭自己

• 调度算法

FIFO
Capacity
Fair