Hadoop内存管理及cpu管理

remington_young

　　转自 :http://blog.csdn.net/flyingpig4/article/details/8754625
　　

　　Hadoop目前较新版本的内存管理(含cpu)分为三个层面，包括JobTracker对Job提交阶段， TaskTracker内存定期检查，JobTracker分配 task扩展点


<一>JobTracker 设置参数控制Job提交时的内存设置
1.1:initializeTaskMemoryRelatedConfig() |初始化TaskMemory
|-->memSizeForMapSlotOnJT
|-->memSizeForReduceSlotOnJT
|-->limitMaxMemForMapTasks
|-->limitMaxMemForReduceTasks
1.2:checkMemoryRequirements |job 提交阶段审核内存需求
|-->perTaskMemoryConfigurationSetOnJT |如果上述4值均未设置，则不用检查
|-->maxMemForMapTask，maxMemForReduceTask |从Job配置信息中获取
|-->JobConf.DISABLED_MEMORY_LIMIT |需在Job中强制配置，否则为无限制，会被分配为非法Job
|-->maxMemForMapTask > limitMaxMemForMapTasks |判断map是否超过最大限制
|-->maxMemForReduceTask > limitMaxMemForReduceTasks |判断reduce是否超过最大限制







<二>TaskTracker设置参数进行定期检查内存，删除task
2.1:initializeMemoryManagement(); |TaskTracker.initialize()初始化阶段初始化内存
|-->if (resourceCalculatorPlugin) |配置了resourceCalculatorPlugin做总控，memoryPlugin已不推荐，一般用Linux*plugin  
|-->获取配置值
|-->totalVirtualMemoryOnTT,totalPhysicalMemoryOnTT
|-->mapSlotMemorySizeOnTT,reduceSlotSizeMemoryOnTT
|-->totalMemoryAllottedForTasks |由maxMapnum*mapSlotMemory + maxReducenum*reduceSlotMemory构成
|-->reservedPhysicalMemoryOnTT,reservedPhysicalMemoryOnTT
|-->setTaskMemoryManagerEnabledFlag() |检查是否为Linux系统
|-->TaskMemoryManagerThread.start() |开始检查taskmemory内存状况
|-->taskRunner()
|-->addToMemoryManager(t.getTaskID(), t.isMapTask(), conf)
  |-->physicalMemoryLimit,virtualMemoryLimit |获取limit内存限制task执行
  |-->taskMemoryManager.addTask(attemptId, virtualMemoryLimit, physicalMemoryLimit)
|--> ProcessTreeInfo |生成一个Task的进程树信息，并添加至processTreeInfoMap当中


2.2:TaskMemoryManagerThread()进程
|-->construct构造函数
|-->maxRssMemoryAllowedForAllTasks,monitoringInterval ｜计算最大物理内存及间隔时间
|-->processTreeInfoMap |循环处理各task进程
|-->ProcfsBasedProcessTree pTree = ptInfo.getProcessTree();  |获取当前进程树信息
|-->currentMemUsage,currentRssMemUsage,curMemUsageOfAgedProcesses,curRssMemUsageOfAgedProcesses |获取当前内存使用状况
|-->limit,limitPhysical |Task内存限制信息获取
|-->check task memory limit |检查内存是否超过限制，此处的检查主要针对单个task而言，不针对整个TaskTracker
|-->doCheckVirtualMemory & isProcessTreeOverLimit() |检查虚拟内存
|-->doCheckPhysicalMemory & isProcessTreeOverLimit() |检查物理内存
|-->isProcessTreeOverLimit(tid.toString(), currentMemUsage, curMemUsageOfAgedProcesses, limit) |检查函数
|-->if (currentMemUsage > (2*limit)) |主要考虑 fork出子进程
|-->else if (curMemUsageOfAgedProcesses > limit) ｜curMemUsageOfAgedProcesses包含了子进程内存大小，即内存树大小
|-->return isOverLimit |超出内存限制标志
|-->do something |获取信息后处理 by isOverLimit
   |-->if (isMemoryOverLimit)
|-->TaskInProgress tip = taskTracker.getRunningTask(tid);
|-->taskTracker.cleanUpOverMemoryTask(tid, true, msg);
|-->else
|-->memoryStillInUsage += currentMemUsage;rssMemoryStillInUsage += currentRssMemUsage;
|-->check total TaskTracker limit大小
|-->doCheckVirtualMemory() & memoryStillInUsage > maxMemoryAllowedForAllTasks
|-->killTasksWithLeastProgress(memoryStillInUsage); |实际取的是LinkedHashMap里面的Task值进行清理
|-->doCheckPhysicalMemory() & rssMemoryStillInUsage > maxRssMemoryAllowedForAllTasks
|-->killTasksWithMaxRssMemory(rssMemoryStillInUsage); |实际取的是max memory使用者进行清理


2.3：此处主要考虑限制物理内存大小，保证一定量的物理内存
2.3.1 保证预留物理内存大小为2G,单位为MB
|-->mapreduce.tasktracker.reserved.physicalmemory.mb = 2 * 1024
2.3.2 另外设置单个task值，最大设为4G
|-->mapreduce.map.memory.physical.mb,mapreduce.reduce.memory.physical.mb
2.3.3 设置plugin插件mapreduce.tasktracker.resourcecalculatorplugin
|-->mapreduce.tasktracker.resourcecalculatorplugin=LinuxResourceCalculatorPlugin






<三>JobTracker 通过心跳分配task任务
3.1:传递TaskStatus给JobTracker
|-->if (askForNewTask)  
|-->freeDiskSpace,totVmem,totPmem,availableVmem,availablePmem,cumuCpuTime,cpuFreq,numCpu,cpuUsage |主要包括cpu及memory参数
|-->status.getResourceStatus().set() ｜一系列set方法设置相应值
|-->TaskTrackerHealthStatus healthStatus = status.getHealthStatus();
|-->healthChecker.setHealthStatus(healthStatus); |设置健康检查脚本，进行健康检查


3.2:heartbeat()方法
|-->taskScheduler.assignTasks() |交由调度器处理Task分配
|-->FairScheduler.assignTasks()
|-->loadMgr.canAssignMap(tts)|LoadManager 来控制是否可以分配task，此处为扩展点
  |--> 可利用task传递的tts的内存及cpu状态确定是否分配task
  |-->CapBasedLoadManager |目前默认|LoadManager为CapBasedLoadManager
|-->canAssignMap() |加入扩展，目前只适宜于cpu，内存控制需要额外扩展TasktrackerStatus中的resource内容(can do)







<四>其它小结

4.1.Java内存分析程序，可借鉴hadoop-cdh3u4版本中的ProcfsBaseprocessTree.java读取进程分析
|-->内存状态的更新也是在此体现
|-->文件分析跟linux思想有关，linux认为一切皆文件
|-->对于进程的子进程，可以以进程树ptree方式获取
|-->linux保存进程信息在/proc/pid*/smmap文件当中，详解见http://blog.csdn.net/insect27/article/details/6214286