在执行MAP任务之前,先了解一下它的容器和它容器的领导:container和nodemanager
NodeManager
NodeManager(NM)是YARN中每个节点上的代理,它管理Hadoop集群中的单个计算节点,包括与ResourceManager保持通信,监督Container的生命周期管理,监控每个Container的资源使用(内存、CPU等)情况,追踪节点健康状况,管理日志和不同应用程序用到的附属服务(auxiliary service)
它包含以下几大组件:
1.NodeStatusUpdater
当NM启动时,该组件向RM注册,并发送节点上的可用资源。接下来,NM与RM通信,汇报Container的状态更新,包括节点上正在运行的Container、已完成的Container等,此外,RM可能向NodeStatusUpdater发信号,杀死处于运行中的Container
NodeStatusUpdater是NM和RM通信的唯一通道,周期性地调用RPC函数nodeHeartbeat()向RM汇报节点上的各种信息
2.ContainerManager
它是NM中的核心组件,实现类是ContainerManagerImpl。它有几个组件组成,各自负责一部分功能,意管理运行在该节点上的所有Container
2.1RPC Server
它负责从AM上接收RPC请求以启动Container或者停止。
供AM使用的接口分别是:startContainer()、stopContainer()、getContainerStatus()
2.2ResourceLocalizationService
负责(从HDFS)安全下载(采用多线程)和组织Container需要的各种文件资源。
2.3ContainerLauncher
维护了一个线程池,随时准备并在必要时尽快启动Container。同时会接收来自RM或者AM的清理Container请求,清理相应进程
2.4AuxServices
NM提供了一个框架以通过配置附属服务扩展自己的功能,这些服务是与NM其他服务隔离开的。
2.5ContainerMonitor
当一个Container启动以后,该组件便开始观察它在运行过程中的资源利用。NM启动一个Container后,ContainerMonitor会将改Container进程对应的一个pid添加到监控列表中
2.6LogHandler
一个可插拔组件,用户通过它可以宣传将Container日志写到本地磁盘或者打包上传到一个文件系统中。
3.ContainerExecutor
与地产操作系统交互,安全存放Container需要的文件和目录,进而以以一种安全的方式启动或者清除Container进程
4.NodeHealthCheckerService
周期性地运行一个配置好的脚步检查节点的健康状况,任何系统健康方面的改变都会通知NodeStatusUpdater传递给RM
5.Security
5.1ApplicationACLsManager 为所有面向用户的API提供安全检查
5.2ContainerTokenSecretManager 检查收到各种请求的合法性,确保这些请求已被RM授权
6.WebServer
web展示
Container
Container的概念
首先它和Linux的Container完全不同。它的使用是启动AM的时候和运行Task的时候,但是它的涉及则是RM向资源调度器申请启动AM的资源时和AM向RM的资源调度器申请启动Task资源时;
当向资源调度器申请资源时,需向它发送一个ResourceRequest列表,其中描述了一个资源单元的详细请求,而资源调度器为之返回分配的资源来描述Container。每个ResourceRequest可看做一个可序列化的Java对象
message ResourceRequestProto { optional PriorityProto priority = 1; // 资源优先级 optional string resource_name = 2; // 资源名称(期望资源所在的host、rack名称等) optional ResourceProto capability = 3; // 资源量(仅支持CPU和内存两种资源) optional int32 num_containers = 4; // 满足以上条件的资源个数 optional bool relax_locality = 5 [default = true]; //是否支持本地性松弛 }
这些资源默认是本地松弛的,即申请优先级为10,资源名称为“node11”,资源量为<2GB,1CPU>的5份资源时,如果node11上没有满足要求的资源,则优先找node11同机架上其他资源,继而找其他机架
AM收到一个或者多个Container后,再次将改Container进一步分配给内部的某个任务,一旦确定任务后,AM需将任务运行环境(包括运行命令、环境变量、依赖的外部文件等)连同Container中的资源封装到ContainerLaunchContext对象中,进而与对应的NM通信,以启动该任务
message ContainerLaunchContextProto { repeated StringLocalResourceMapProto localResources = 1; //Container启动以来的外部资源 optional bytes tokens = 2; repeated StringBytesMapProto service_data = 3; repeated StringStringMapProto environment = 4; //Container启动所需的环境变量 repeated string command = 5; //Container内部运行的任务启动命令,如果是MapReduce的话,Map/Reduce Task启动命令就在该字段中 repeated ApplicationACLMapProto application_ACLs = 6; }
Container启动步骤:
1.资源本地化
在本地拷贝一份运行Container所需的所有资源(通过Distributed Cache实现);
为Container创建经隔离的工作目录,并在这些目录中准备好所有资源;
YARN将资源分为两类:一类是public级别的资源,放在公共目录下,由所有用户共享,另一类是private级别的资源,这类资源时用户私有的,只能在所属用户的各个作业间共享。
2.启动Container
启动Container是由ContainerLauncher完成的;
3.运行Container
由ContainerExecutor完成
4.资源回收
由ResourceLocalizationService服务完成,该过程与资源本地化正好相反,它负责撤销Container运行过程中使用的各种资源。
MAP
mapper就是在运行Container的时候执行的。主角上场。
Map任务是一类将输入记录转换为中间格式记录集的独立任务。Mapper类中的map方法将输入键值对映射到一组中间格式的键值对集合
Container启动以后会根据AM传过来的任务信息启动一个YarnChild进程来运行任务,YarnChild直接调用分给它的jvmTask,而jvmTask则判断是map任务还是reduce任务来分别执行MapTask和ReduceTask来运行Map过程和Reduce过程
每个task都会使用一个进程占用一个JVM来执行,org.apache.hadoop.mapred.Child方法是具体的JVM启动类
taskFinal.run(job, umbilical); // run the task
if (taskComing) { boolean isMap = in.readBoolean(); if (isMap) { t = new MapTask(); } else { t = new ReduceTask(); } t.readFields(in); }
这里的taskFinal就是jvmTask
自定义的Map类继承自Mapper,由MapTask的run()方法来运行
1 @Override 2 public void run(final JobConf job, final TaskUmbilicalProtocol umbilical) 3 throws IOException, ClassNotFoundException, InterruptedException { 4 this.umbilical = umbilical; 5 6 if (isMapTask()) { 7 // If there are no reducers then there won't be any sort. Hence the map 8 // phase will govern the entire attempt's progress. 9 if (conf.getNumReduceTasks() == 0) { 10 mapPhase = getProgress().addPhase("map", 1.0f); 11 } else { 12 // If there are reducers then the entire attempt's progress will be 13 // split between the map phase (67%) and the sort phase (33%). 14 mapPhase = getProgress().addPhase("map", 0.667f); 15 sortPhase = getProgress().addPhase("sort", 0.333f); 16 } 17 } 18 TaskReporter reporter = startReporter(umbilical); 19 20 boolean useNewApi = job.getUseNewMapper(); 21 initialize(job, getJobID(), reporter, useNewApi); 22 23 // check if it is a cleanupJobTask 24 if (jobCleanup) { 25 runJobCleanupTask(umbilical, reporter); 26 return; 27 } 28 if (jobSetup) { 29 runJobSetupTask(umbilical, reporter); 30 return; 31 } 32 if (taskCleanup) { 33 runTaskCleanupTask(umbilical, reporter); 34 return; 35 } 36 37 if (useNewApi) { 38 runNewMapper(job, splitMetaInfo, umbilical, reporter); 39 } else { 40 runOldMapper(job, splitMetaInfo, umbilical, reporter); 41 } 42 done(umbilical, reporter); 43 }
MapTask先判断是否有Reduce任务,如果没有的话,Map任务结束则整个提交的作业结束;如果有的话,当Map任务完成的时候设置当前进度为66.7%,Sort完成的时候设置进度为33.3%;
之后启动TaskReporter,用于更新当前状态;
之后初始化任务,设置当前任务的状态为RUNNING,设置输出目录等;
之后判断任务是不是jobCleanup任务、jobSetup任务、taskCleanup任务,并做相应的处理;
之后判断使用新旧哪套API,因为MapTask要兼容两套API;
确定以后调用runNewMapper方法,执行具体的map;
作业完成以后调用done方法,进行任务的清理、计数器的更新、任务状态更新等;
hadoop2的话是使用runNewMapper()
1 @SuppressWarnings("unchecked") 2 private <INKEY,INVALUE,OUTKEY,OUTVALUE> 3 void runNewMapper(final JobConf job, 4 final TaskSplitIndex splitIndex, 5 final TaskUmbilicalProtocol umbilical, 6 TaskReporter reporter 7 ) throws IOException, ClassNotFoundException, 8 InterruptedException { 9 // make a task context so we can get the classes 10 org.apache.hadoop.mapreduce.TaskAttemptContext taskContext = 11 new org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl(job, 12 getTaskID(), 13 reporter); 14 // make a mapper 15 org.apache.hadoop.mapreduce.Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE> mapper = 16 (org.apache.hadoop.mapreduce.Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE>) 17 ReflectionUtils.newInstance(taskContext.getMapperClass(), job); 18 // make the input format 19 org.apache.hadoop.mapreduce.InputFormat<INKEY,INVALUE> inputFormat = 20 (org.apache.hadoop.mapreduce.InputFormat<INKEY,INVALUE>) 21 ReflectionUtils.newInstance(taskContext.getInputFormatClass(), job); 22 // rebuild the input split 23 org.apache.hadoop.mapreduce.InputSplit split = null; 24 split = getSplitDetails(new Path(splitIndex.getSplitLocation()), 25 splitIndex.getStartOffset()); 26 LOG.info("Processing split: " + split); 27 28 org.apache.hadoop.mapreduce.RecordReader<INKEY,INVALUE> input = 29 new NewTrackingRecordReader<INKEY,INVALUE> 30 (split, inputFormat, reporter, taskContext); 31 32 job.setBoolean(JobContext.SKIP_RECORDS, isSkipping()); 33 org.apache.hadoop.mapreduce.RecordWriter output = null; 34 35 // get an output object 36 if (job.getNumReduceTasks() == 0) { 37 output = 38 new NewDirectOutputCollector(taskContext, job, umbilical, reporter); 39 } else { 40 output = new NewOutputCollector(taskContext, job, umbilical, reporter); 41 } 42 43 org.apache.hadoop.mapreduce.MapContext<INKEY, INVALUE, OUTKEY, OUTVALUE> 44 mapContext = 45 new MapContextImpl<INKEY, INVALUE, OUTKEY, OUTVALUE>(job, getTaskID(), 46 input, output, 47 committer, 48 reporter, split); 49 50 org.apache.hadoop.mapreduce.Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE>.Context 51 mapperContext = 52 new WrappedMapper<INKEY, INVALUE, OUTKEY, OUTVALUE>().getMapContext( 53 mapContext); 54 55 try { 56 input.initialize(split, mapperContext); 57 mapper.run(mapperContext); 58 mapPhase.complete(); 59 setPhase(TaskStatus.Phase.SORT); 60 statusUpdate(umbilical); 61 input.close(); 62 input = null; 63 output.close(mapperContext); 64 output = null; 65 } finally { 66 closeQuietly(input); 67 closeQuietly(output, mapperContext); 68 } 69 }
它的执行过程是:
1.获取配置信息类对象TaskAttemptContextImplement、自己开发的Mapper实例、用户指定的InputFormat对象(默认是TextInputFormat)、任务对应的分片信息split;
2.根据inputFormat构建一个NewTrackingRecordReader对象,这个对象中的RecordReader<K,V> real是LineRecordReader,用于读取分片中的内容,传递给Mapper的map方法处理;
3.执行Mapper中的setup方法;
4.循环执行map方法;
5.执行cleanup方法;
6.最后是输出流的关闭output.close(mapperContext),该方法会执行MapOutputBuffer.flush()操作,将剩余数据也通过sortAndSpill()方法写入本地文件,并在最后调用mergeParts()方法合并所有的spill文件。
关于spill,spill是map中比较重要的设计
spill过程包括输出、排序、溢写、合并等步骤;
每个Map任务不断的以<key,value>对的形式把数据输出到内存中构造一个环形的数据结构。这个数据结构其实是一个字节数组,叫kvbuffer,这里面不仅有<Key,Value>数据,还有索引数据,并且给放置索引数据的区域起了一个kvmeta的别名
当这个缓冲区满足一定条件后就会对缓冲区kvbuffer中的数据进行排序,先按分区编号partition进行升序,然后按照key进行升序。这样快速排序后数据以分区为单位聚集在一起,且同一分区的所有数据按照key有序。然后通过sortAndSpill方法写到本地文件和索引文件;如果有combiner,spill之前也会做一次聚集操作,等数据跑完通过归并合并所有spill文件和索引文件。
Map阶段的结果都会存储在本地种(如果有reducer的话),非HDFS。
转自:https://www.cnblogs.com/admln/p/hadoop2-work-excute-map.html