在执行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