hadoop作业调度和mapreduce过程详解_在hodoop中mapruduce是如何完成调度和协调工作的

一直想写一个关于hadoopMR和作业调度,还有存储过程(hdfs)等的详细总结,因为这一段时间巩固并且学到了很多,所以借此来写一个好一点的详细一点的,以后忘了好再看着回忆一下;

 先从作业提交开始 ,首先来一个简略的,作为动作级别的;

首先你有了一个作业,比如MR,然后你作为客户端,向服务器提交作业,首先提交至ResourceManager,获取一个作业ID,如果没有,则生成,如果有的话,就抛出异常;

现在假设生成了ID,得到了ID,ResourceManager就开始保存提交的作业配置,函数等,并且通过调度器在某个节点通过NodeManager生成有一个ApplicationMaster,这玩意是占有一定资源,是一个子进程;然后AM(ApplicationMaster)来初始化多个map,reduce任务,

然后监控任务执行的程度,其中map reduce中经过shuffle,首先是map函数,然后将map的输出通过环形缓冲区,排序,分区,溢写成文件,多个溢写文件合并,reduce端fetch数据,归并,排序,执行reduce函数,结束,输出到hdfs;

详细版的:

首先你写了一个mapreduce的自定义函数;

首先你有两个类,一个实现Mapper接口的类,一个实现了Reducer接口的类,其中分别重写了map函数和reduce函数;当然这两个函数中不光有这两个函数,还有setup,cleanup等函数,setup,cleanup都是只执行一个(在一个map循环),一个reduce执行;

其实这两个函数的作用都是对数据进行处理的,MR基于key,value的,其中的value就是你所要处理的数据,而key一般是行的偏移量(看你的输入格式了,或者你要套用什么数据结构,也可以自定义);

然后这就是你主要使用的计算函数了,也是原始hadoop]中的最重要,最基础的计算部分,这就是你要提交的作业;

你要做什么作业,作为大数据处理,简化一下,无非就是获得输入流,处理,获得输出流,然后写到某个地方;这也是基础流程,而大数据,当然解决的是大量数据的处理,输入输出的问题,和一般量的数据处理,输入输出是不一样的,它考虑的更多,比如可靠性,可用性,高性能,易用性等;在这里,你作为一个客户端进行了提交,比如入门的WordCount,你要提交给一个集群,然后让它在一个集群的各个节点上运行,这就涉及了通信,信息异步,分布式存储等;先不扯这些,先进行提交;

一般都是在IDE上运行那个main方法,或者打个带main的jar包进行提交运行;官方使用的是用jar包来进行提交,所以我们就将这个方式的;

在提交之前,我们要知道提交这个动作,比如,提交给谁,为什么要提交,提交之后怎么识别;

你作为一个分布式作业,你要提交到每个节点上,你作为不同的业务,当然有着不同的计算,而提交之后怎么识别,当然是通过你重写的map reduce类来识别;

而识别,你就要知道一个东西,叫做configuration对象,还有job对象(虽然现在是Job.getInstance来创建的),在你写的那个驱动(对,就是那个带main方法的那个里面都是job.set什么的那个)中的job对象中有个new Configuration对象,这个对象,你觉得你只是通过set,设置了输入输出格式,map reduce类等,再多一点还有自定义分区,分组类的指定,存储路径指定等,但其实在你new那个configuration对象的时候,那个对象就会自动加载hadoop目录下的etc/hadoop下的xml文件

 

public Configuration() {
  this(true);
}
 
/** A new configuration where the behavior of reading from the default 
 * resources can be turned off.
 * 
 * If the parameter {@code loadDefaults} is false, the new instance
 * will not load resources from the default files. 
 * @param loadDefaults specifies whether to load from the default files
 */
public Configuration(boolean loadDefaults) {
  this.loadDefaults = loadDefaults;
  updatingResource = new HashMap<String, String[]>();
  synchronized(Configuration.class) {
    REGISTRY.put(this, null);
  }
}

无参调用有参,然后有参最后会调用loadDefaults,后面过程很复杂,也就是说你写配置里的会加载进来代码如下

 

static{
  //print deprecation warning if hadoop-site.xml is found in classpath
  ClassLoader cL = Thread.currentThread().getContextClassLoader();
  if (cL == null) {
    cL = Configuration.class.getClassLoader();
  }
  if(cL.getResource("hadoop-site.xml")!=null) {
    LOG.warn("DEPRECATED: hadoop-site.xml found in the classpath. " +
        "Usage of hadoop-site.xml is deprecated. Instead use core-site.xml, "
        + "mapred-site.xml and hdfs-site.xml to override properties of " +
        "core-default.xml, mapred-default.xml and hdfs-default.xml " +
        "respectively");
  }
  addDefaultResource("core-default.xml");
  addDefaultResource("core-site.xml");
  //Add code for managing deprecated key mapping
  //for example
  //addDeprecation("oldKey1",new String[]{"newkey1","newkey2"});
  //adds deprecation for oldKey1 to two new keys(newkey1, newkey2).
  //so get or set of oldKey1 will correctly populate/access values of 
  //newkey1 and newkey2
  addDeprecatedKeys();
}

当然你配置的非默认项会覆盖成你配置的,你没配置的默认为defualt.xml中的配置;

回到任务提交,现在我们提交了一个jar,,然后在设置的job.set中,结合configuration对象,创建JobConf对象,最为参数配额配置,

然后在一个jobclient下使用job.submit,然后使用submit方法,在此方法中创建cluster对象(集群对象),调用connect方法,连接集群,cluster中的ConfigUtil.loadResources静态工具方法加载mapred-default.xml,,mapred-site.xml,yarn-site.xml等

 

static {
    ConfigUtil.loadResources();
}

就是通过这个方法,这个是从cluster类中找到的;

然后获取节点信息,获取用户,获取文件系统,获取客户端;然后通过ClientProtocolProvider类的ServiceLoader来进行服务的提供,

然后进行初始化,获得YARNRunner,以及YARNRunner文件系统URL然后将这个Job和Cluster对象作为参数传给submitinternal方法,这个方法获得本节点的IP,然后通过内置的submitClinet.getNewJobID生成一个作业ID,将id写入Job对象,准备好http接口,证书等,然后将可执行文件之类的拷贝到hdfs,生成切片,决定map数量;这些信息提交到stagin directory下,提交这些信息之后根据切片信息 以及inputformat之类的生成切片,以及切片位置,然后使用YARNRunner进行提交,提交的是个ASC(ApplicationSubmissionContext,其中包括了个CLC, 随后,RM受理了这个作业之后,生成创建AM的命令shell,将CLC转发给某个NM节点,另起一个java虚拟机,然后执行MRApplicationMaster.class,执行了这个类之后，创建MRAM，通过CLC来识别和存初始化作业，然后进行TaskAttempt，进行尝试初始化其他作业的Container，如果发现通过心跳和反馈得到了足够的资源，就进行初始化Container，然后进行MapTask，在初始化任务的时候也有个，就是当初始化map超过%5的时候才能初始化ReducerTask，初始化完成以后，状态切换，触发状态机，进行下一步动作，开始mapTask，加载资源，加载切片规则，加载RecordReader，然后执行setup，map的循环调用，cleanup等函数，最后进行context.write,(之前要经过一次排序)也就是collector,也是RecordWriter,在outputBuffer(环形缓冲区)中进行分区,排序,(对元数据),其中缓冲区的数据包括元数据和数据两部分,元数据由k,v偏移量,长度;数据就是k,v,然后进行排序,然后进行溢写操作,进行spill操作,多次spill然后进行合并,最后形成一个大的spill;