Hadoop新MapReduce 框架 Yarn 详解(PDF清晰版)

Hadoop新MapReduce 框架 Yarn 详解 第 1 页 共 22 页 Hadoop新 MapReduce 框架 Yarn 详解 简介： 本文介绍了 Hadoop 自 0.23.0 版本后新的 map-reduce 框架（Yarn) 原理，优势，运作机制和配 置方法等；着重介绍新的 yarn 框架相对于原框架的差异及改进；并通过 Demo 示例详细描述了在新的 yarn 框架下搭建和开发 hadoop 程序的方法。 读者通过本文中新旧 hadoop map-reduce 框架的对比，更 能深刻理解新的 yarn 框架的技术原理和 设计 领导形象设计圆作业设计ao工艺污水处理厂设计附属工程施工组织设计清扫机器人结构设计 思想，文中的 Demo 代码经过微小修改即可用于用户基于 hadoop 新框架的实际生产环境。 Hadoop MapReduceV2(Yarn) 框架简介 原 Hadoop MapReduce 框架的问 题 快递公司问题件快递公司问题件货款处理关于圆的周长面积重点题型关于解方程组的题及答案关于南海问题 对于业界的大数据存储及分布式处理系统来说，Hadoop 是耳熟能详的卓越开源分布式文件存储及处理框 架，对于 Hadoop 框架的介绍在此不再累述，读者可参考 Hadoop 官方简介。使用和学习过老 Hadoop 框 架（0.20.0 及之前版本）的同仁应该很熟悉如下的原 MapReduce 框架图： 图 1.Hadoop 原 MapReduce 架构 从上图中可以清楚的看出原 MapReduce 程序的 流程 快递问题件怎么处理流程河南自建厂房流程下载关于规范招聘需求审批流程制作流程表下载邮件下载流程设计 及设计思路： 1. 首先用户程序 (JobClient) 提交了一个 job，job 的信息会发送到 Job Tracker 中，Job Tracker 是 Map-reduce 框架的中心，他需要与集群中的机器定时通信 (heartbeat), 需要管理哪些程序应该跑在 哪些机器上，需要管理所有 job 失败、重启等操作。 2. TaskTracker 是 Map-reduce 集群中每台机器都有的一个部分，他做的事情主要是监视自己所在机 器的资源情况。 Hadoop新MapReduce 框架 Yarn 详解 第 2 页 共 22 页 3. TaskTracker 同时监视当前机器的 tasks 运行状况。TaskTracker 需要把这些信息通过 heartbeat 发送给 JobTracker，JobTracker 会搜集这些信息以给新提交的 job 分配运行在哪些机器上。上图 虚线箭头就是表示消息的发送 - 接收的过程。 可以看得出原来的 map-reduce 架构是简单明了的，在最初推出的几年，也得到了众多的成功案例，获得业 界广泛的支持和肯定，但随着分布式系统集群的规模和其工作负荷的增长，原框架的问题逐渐浮出水面，主 要的问题集中如下： 1. JobTracker 是 Map-reduce 的集中处理点，存在单点故障。 2. JobTracker 完成了太多的任务，造成了过多的资源消耗，当 map-reduce job 非常多的时候，会造 成很大的内存开销，潜在来说，也增加了 JobTracker fail 的风险，这也是业界普遍总结出老 Hadoop 的 Map-Reduce 只能支持 4000 节点主机的上限。 3. 在 TaskTracker 端，以 map/reduce task 的数目作为资源的表示过于简单，没有考虑到 cpu/ 内存 的占用情况，如果两个大内存消耗的 task 被调度到了一块，很容易出现 OOM。 4. 在 TaskTracker 端，把资源强制划分为 map task slot 和 reduce task slot, 如果当系统中只有 map task 或者只有 reduce task 的时候，会造成资源的浪费，也就是前面提过的集群资源利用的问 题。 5. 源代码层面分析的时候，会发现代码非常的难读，常常因为一个 class 做了太多的事情，代码量达 3000 多行，，造成 class 的任务不清晰，增加 bug 修复和版本维护的难度。 6. 从操作的角度来看，现在的 Hadoop MapReduce 框架在有任何重要的或者不重要的变化 ( 例如 bug 修复，性能提升和特性化 ) 时，都会强制进行系统级别的升级更新。更糟的是，它不管用户的 喜好，强制让分布式集群系统的每一个用户端同时更新。这些更新会让用户为了验证他们之前的应用 程序是不是适用新的 Hadoop 版本而浪费大量时间。 新 Hadoop Yarn 框架原理及运作机制 从业界使用分布式系统的变化趋势和 hadoop 框架的长远发展来看，MapReduce 的 JobTracker/TaskTracker 机制需要大规模的调整来修复它在可扩展性，内存消耗，线程模型，可靠性和性能 上的缺陷。在过去的几年中，hadoop 开发团队做了一些 bug 的修复，但是最近这些修复的成本越来越高， 这表明对原框架做出改变的难度越来越大。 为从根本上解决旧 MapReduce 框架的性能瓶颈，促进 Hadoop 框架的更长远发展，从 0.23.0 版本开始， Hadoop 的 MapReduce 框架完全重构，发生了根本的变化。新的 Hadoop MapReduce 框架命名为 MapReduceV2 或者叫 Yarn，其架构图如下图所示： Hadoop新MapReduce 框架 Yarn 详解 第 3 页 共 22 页 图 2. 新的 Hadoop MapReduce 框架（Yarn）架构 重构根本的思想是将 JobTracker 两个主要的功能分离成单独的组件，这两个功能是资源管理和任务调度 / 监控。新的资源管理器全局管理所有应用程序计算资源的分配，每一个应用的 ApplicationMaster 负责相应 的调度和协调。一个应用程序无非是一个单独的传统的 MapReduce 任务或者是一个 DAG( 有向无环图 ) 任务。ResourceManager 和每一台机器的节点管理服务器能够管理用户在那台机器上的进程并能对计算进 行组织。 事实上，每一个应用的 ApplicationMaster 是一个详细的框架库，它结合从 ResourceManager 获得的资源 和 NodeManager 协同工作来运行和监控任务。 上图中 ResourceManager 支持分层级的应用队列，这些队列享有集群一定比例的资源。从某种意义上讲它 就是一个纯粹的调度器，它在执行过程中不对应用进行监控和状态跟踪。同样，它也不能重启因应用失败或 者硬件错误而运行失败的任务。 ResourceManager 是基于应用程序对资源的需求进行调度的 ; 每一个应用程序需要不同类型的资源因此 就需要不同的容器。资源包括：内存，CPU，磁盘，网络等等。可以看出，这同现 Mapreduce 固定类型的 资源使用模型有显著区别，它给集群的使用带来负面的影响。资源管理器提供一个调度策略的插件，它负责 将集群资源分配给多个队列和应用程序。调度插件可以基于现有的能力调度和公平调度模型。 上图中 NodeManager 是每一台机器框架的代理，是执行应用程序的容器，监控应用程序的资源使用情况 (CPU，内存，硬盘，网络 ) 并且向调度器汇报。 每一个应用的 ApplicationMaster 的职责有：向调度器索要适当的资源容器，运行任务，跟踪应用程序的状 态和监控它们的进程，处理任务的失败原因。 新旧 Hadoop MapReduce 框架比对 让我们来对新旧 MapReduce 框架做详细的分析和对比，可以看到有以下几点显著变化： 首先客户端不变，其调用 API 及接口大部分保持兼容，这也是为了对开发使用者透明化，使其不必对原有 代码做大的改变 ( 详见 2.3 Demo 代码开发及详解)，但是原框架中核心的 JobTracker 和 TaskTracker 不 见了，取而代之的是 ResourceManager, ApplicationMaster 与 NodeManager 三个部分。 我们来详细解释这三个部分，首先 ResourceManager 是一个中心的服务，它做的事情是调度、启动每一个 Job 所属的 ApplicationMaster、另外监控 ApplicationMaster 的存在情况。细心的读者会发现：Job 里面 所在的 task 的监控、重启等等内容不见了。这就是 AppMst 存在的原因。ResourceManager 负责作业与 Hadoop新MapReduce 框架 Yarn 详解 第 4 页 共 22 页 资源的调度。接收 JobSubmitter 提交的作业，按照作业的上下文 (Context) 信息，以及从 NodeManager 收 集来的状态信息，启动调度过程，分配一个 Container 作为 App Mstr NodeManager 功能比较专一，就是负责 Container 状态的维护，并向 RM 保持心跳。 ApplicationMaster 负责一个 Job 生命周期内的所有工作，类似老的框架中 JobTracker。但注意每一个 Job （不是每一种）都有一个 ApplicationMaster，它可以运行在 ResourceManager 以外的机器上。 Yarn 框架相对于老的 MapReduce 框架什么优势呢？我们可以看到： 1. 这个设计大大减小了 JobTracker（也就是现在的 ResourceManager）的资源消耗，并且让监测每 一个 Job 子任务 (tasks) 状态的程序分布式化了，更安全、更优美。 2. 在新的 Yarn 中，ApplicationMaster 是一个可变更的部分，用户可以对不同的编程模型写自己的 AppMst，让更多类型的编程模型能够跑在 Hadoop 集群中，可以参考 hadoop Yarn 官方配置 模板 个人简介word模板免费下载关于员工迟到处罚通告模板康奈尔office模板下载康奈尔 笔记本 模板 下载软件方案模板免费下载 中的 mapred-site.xml 配置。 3. 对于资源的表示以内存为单位 ( 在目前版本的 Yarn 中，没有考虑 cpu 的占用 )，比之前以剩余 slot 数目更合理。 4. 老的框架中，JobTracker 一个很大的负担就是监控 job 下的 tasks 的运行状况，现在，这个部分 就扔给 ApplicationMaster 做了，而 ResourceManager 中有一个模块叫做 ApplicationsMasters( 注意不是 ApplicationMaster)，它是监测 ApplicationMaster 的运行状况，如 果出问题，会将其在其他机器上重启。 5. Container 是 Yarn 为了将来作资源隔离而提出的一个框架。这一点应该借鉴了 Mesos 的工作，目 前是一个框架，仅仅提供 java 虚拟机内存的隔离 ,hadoop 团队的设计思路应该后续能支持更多的 资源调度和控制 , 既然资源表示成内存量，那就没有了之前的 map slot/reduce slot 分开造成集群 资源闲置的尴尬情况。 新的 Yarn 框架相对旧 MapRduce 框架而言，其配置文件 , 启停脚本及全局变量等也发生了一些变化，主 要的改变如下： 表 1. 新旧 Hadoop 脚本 / 变量 / 位置变化表 改变项 原框架中 新框架中（Yarn） 备注 配置文件位置 ${hadoop_home_dir}/conf ${hadoop_home_dir}/etc/ha doop/ Yarn 框架也兼容老的 ${hadoop_home_dir}/conf 位置配置，启动 时会检测是否存在老的 conf 目录，如果存 在将加载 conf 目录下的配置，否则加载 etc 下配置 启停脚本 ${hadoop_home_dir}/bin/ start（stop）-all.sh ${hadoop_home_dir}/sbin/st art（stop）-dfs.sh ${hadoop_home_dir}/bin/st art(stop)-all.sh 新的 Yarn 框架中启动分布式文件系统和启 动 Yarn 分离，启动 / 停止分布式文件系统 的命令位于 ${hadoop_home_dir}/sbin 目 录下，启动 / 停止 Yarn 框架位于 ${hadoop_home_dir}/bin/ 目录下 JAVA_HOME 全局变量 ${hadoop_home_dir}/bin/ start-all.sh 中 ${hadoop_home_dir}/etc/ha doop/hadoop-env.sh ${hadoop_home_dir}/etc/ha doop/Yarn-env.sh Yarn 框架中由于启动 hdfs 分布式文件系 统和启动 MapReduce 框架分离， JAVA_HOME 需要在 hadoop-env.sh 和 Yarn-env.sh 中分别配置 HADOOP_LO G_DIR 全局变 不需要配置 ${hadoop_home_dir}/etc/ha doop/hadoop-env.sh 老框架在 LOG，conf，tmp 目录等均默认 为脚本启动的当前目录下的 log,conf，tmp Hadoop新MapReduce 框架 Yarn 详解 第 5 页 共 22 页 量 子目录 Yarn 新框架中 Log 默认创建在 Hadoop 用户的 home 目录下的 log 子目录，因此 最好在 ${hadoop_home_dir}/etc/hadoop/hadoop-e nv.sh 配置 HADOOP_LOG_DIR，否则有 可能会因为你启动 hadoop 的用户 的 .bashrc 或者 .bash_profile 中指定了其 他的 PATH 变量而造成日志位置混乱，而 该位置没有访问权限的话启动过程中会报错 由于新的 Yarn 框架与原 Hadoop MapReduce 框架相比变化较大，核心的配置文件中很多项在新框架中已 经废弃，而新框架中新增了很多其他配置项，看下表所示会更加清晰： 表 2. 新旧 Hadoop 框架配置项变化表 配置文件 配置项 Hadoop 0.20.X 配置 Hadoop 0.23.X 配置 说明 core-site.xml 系统默认分布 式文件 URI fs.default.name fs.defaultFS hdfs-site.xml DFS name node 存放 name table 的 目录 dfs.name.dir dfs.namenode.name.dir 新框架中 name node 分成 dfs.namenode.name.dir( 存 放 naname table 和 dfs.namenode.edits.dir（存 放 edit 文件），默认是同一 个目录 DFS data node 存放数据 block 的目录 dfs.data.dir dfs.datanode.data.dir 新框架中 DataNode 增加 更多细节配置，位于 dfs.datanode. 配置项下，如 dfs.datanode.data.dir.perm （datanode local 目录默认 权限）； dfs.datanode.address （datanode 节点监听端 口）；等 分布式文件系 统数据块复制 数 dfs.replication dfs.replication 新框架与老框架一致，值建 议配置为与分布式 cluster 中实际的 DataNode 主机 数一致 mapred-site.x ml Job 监控地址 及端口 mapred.job.tracker 无 新框架中已改为 Yarn-site.xml 中的 resouceManager 及 nodeManager 具体配置项， Hadoop新MapReduce 框架 Yarn 详解 第 6 页 共 22 页 新框架中历史 job 的查询已 从 Job tracker 剥离，归入 单独的 mapreduce.jobtracker.jobhi story 相关配置， 第三方 MapReduce 框 架 无 mapreduce.framework.name 新框架支持第三方 MapReduce 开发框架以支 持如 SmartTalk/DGSG 等 非 Yarn 架构，注意通常情 况下这个配置的值都设置为 Yarn，如果没有配置这项， 那么提交的 Yarn job 只会 运行在 locale 模式，而不是 分布式模式。 Yarn-site.xml The address of the applications manager interface in the RM 无 Yarn.resourcemanager.addr ess 新框架中 NodeManager 与 RM 通信的接口地址 The address of the scheduler interface 无 Yarn.resourcemanager.sche duler.address 同上，NodeManger 需要知 道 RM 主机的 scheduler 调度服务接口地址 The address of the RM web application 无 Yarn.resourcemanager.web app.address 新框架中各个 task 的资源 调度及运行状况通过通过该 web 界面访问 The address of the resource tracker interface 无 Yarn.resourcemanager.reso urce-tracker.address 新框架中 NodeManager 需 要向 RM 报告 软件系统测试报告下载sgs报告如何下载关于路面塌陷情况报告535n,sgs报告怎么下载竣工报告下载 任务运行状 态供 Resouce 跟踪，因此 NodeManager 节点主机需 要知道 RM 主机的 tracker 接口地址 回页首 Hadoop Yarn 框架 Demo 示例 Demo 场景介绍：Weblogic 应用服务器日志分析 了解了 hadoop 新的 Yarn 框架的架构和思路后，我们用一个 Demo 示例来检验新 Yarn 框架下 Map-Reduce 程序的开发部署。 Hadoop新MapReduce 框架 Yarn 详解 第 7 页 共 22 页 我们考虑如下应用场景：用户的生产系统由多台 Weblogic 应用服务器组成，每天需要每台对应用服务器的 日志内容进行检查，统计其日志级别和日志模块的总数。 WebLogic 的日志范例如下图所示： 图 3.Weblogic 日志示例 如上图所示， 为 weblogic 的日志级别，， 为 Weblogic 的日志模块， 我们主要分析 loglevel 和 logmodule 这两个维度分别在 WebLogic 日志中出现的次数，每天需要统计出 loglevel 和 logmodule 分别出现的次数总数。 Demo 测试环境 Yarn 框架搭建 由于 Weblogic 应用服务器分布于不同的主机，且日志数据量巨大，我们采用 hadoop 框架将 WebLogic 各 个应用服务器主机上建立分布式目录，每天将 WebLogic 日志装载进 hadoop 分布式文件系统，并且编写 基于 Yarn 框架的 MapReduce 程序对日志进行处理，分别统计出 LogLevel 和 Logmodule 在日志中出 现的次数并计算总量，然后输出到分布式文件系统中，输出目录命名精确到小时为后缀以便区分每次 Demo 程序运行的处理结果。 我们搭建一个 Demo 测试环境以验证 Yarn 框架下分布式程序处理该案例的功能，以两台虚拟机作为该 Demo 的运行平台，两机均为 Linux 操作系统，机器 hostname 为 OEL 和 Stephen，OEL 作为 NameNode 和 ResouceManager 节点主机，64 位，Stephen 作为 DataNode 和 NodeManager 节点主 机，32 位（Hadoop 支持异构性）, 具体如下： 表 3.Demo 测试环境表 主机名 角色 备注 OEL(192.168.137.8) NameNode 节点主机 ResourceManager 主机 linux 操作系统 32bit Stephen(192.168.l37.2) DataNode 节点主机 NodeManager 主机 linux 操作系统 64bit 我们把 hadoop 安装在两台测试机的 /hadoop 文件系统目录下，安装后的 hadoop 根目录为： /hadoop/hadoop-0.23.0，规划分布式文件系统存放于 /hadoop/dfs 的本地目录，对应分布式系统中的目录 为 /user/oracle/dfs 我们根据 Yarn 框架要求，分别在 core-site.xml 中配置分布式文件系统的 URL，详细如下： Hadoop新MapReduce 框架 Yarn 详解 第 8 页 共 22 页 清单 1.core-site.xml 配置 fs.defaultFS hdfs://192.168.137.8:9100 在 hdfs-site.xml 中配置 nameNode，dataNode 的本地目录信息，详细如下： 清单 2.hdfs-site.xml 配置 dfs.namenode.name.dir /hadoop/dfs/name dfs.datanode.data.dir /hadoop/dfs/data dfs.replication 2 在 mapred-site.xml 中配置其使用 Yarn 框架执行 map-reduce 处理程序，详细如下： 清单 3.mapred-site.xml 配置 mapreduce.framework.name Yarn Hadoop新MapReduce 框架 Yarn 详解 第 9 页 共 22 页 最后在 Yarn-site.xml 中配置 ResourceManager，NodeManager 的通信端口，web 监控端口等，详细如 下： 清单 4.Yarn-site.xml 配置 Yarn.nodemanager.aux-services mapreduce.shuffle The address of the applications manager interface in the RM. Yarn.resourcemanager.address 192.168.137.8:18040 The address of the scheduler interface. Yarn.resourcemanager.scheduler.address 192.168.137.8:18030 The address of the RM web application. Yarn.resourcemanager.webapp.address 192.168.137.8:18088 The address of the resource tracker interface. Yarn.resourcemanager.resource-tracker.address 192.168.137.8:8025 具体配置项的含义，在 hadoop 官方网站有详细的说明，读者可以参见 hadoop 0.23.0 官方配置模板。 Demo 代码开发及详解 以下我们详细介绍一下新的 Yarn 框架下针对该应用场景的 Demo 代码的开发， 在 Demo 程序的每个类 都有详细的注释和说明，Yarn 开发为了兼容老版本，API 变化不大，可以参考 官方 Hadoop Yarn 框架 API。 Hadoop新MapReduce 框架 Yarn 详解 第 10 页 共 22 页 在 Map 程序中，我们以行号为 key，行文本为 value 读取每一行 WebLogic 日志输入，将 loglevel 和 logmodule 的值读出作为 Map 处理后的新的 key 值，由于一行中 loglevel 和 logmodule 的出现次数应 该唯一，所以经 Map 程序处理后的新的 record 记录的 value 应该都为 1： 清单 5. Map 业务逻辑 public static class MapClass extends Mapper { private Text record = new Text(); private static final IntWritable recbytes = new IntWritable(1); public void map(Object key, Text value,Context context) throws IOException,InterruptedException { String line = value.toString(); // 没有配置 RecordReader，所以默认采用 line 的实现， //key 就是行号，value 就是行内容， // 按行 key-value 存放每行 loglevel 和 logmodule 内容 if (line == null || line.equals("")) return; String[] words = line.split("> <"); if (words == null || words.length < 2) return; String logLevel = words[1]; String moduleName = words[2]; record.clear(); record.set(new StringBuffer("logLevel::").append(logLevel).toString()); context.write(record, recbytes); // 输出日志级别统计结果，通过 logLevel:: 作为前缀来标示。 record.clear(); record.set(new StringBuffer("moduleName::").append(moduleName).toString()); context.write(record, recbytes); // 输出模块名的统计结果，通过 moduleName:: 作为前缀来标示 } } 由于有 loglevel 和 logmodule 两部分的分析工作，我们设定两个 Reduce 来分别处理这两部分，loglevel 的交给 reduce1，logmodule 交给 reduce2。因此我们编写 Patitioner 类，根据 Map 传过来的 Key 中包 含的 logLevel 和 moduleName 的前缀，来分配到不同的 Reduce： 清单 6.Partition 业务逻辑 public static class PartitionerClass extends Partitioner { public int getPartition(Text key, IntWritable value, int numPartitions) { Hadoop新MapReduce 框架 Yarn 详解 第 11 页 共 22 页 if (numPartitions >= 2)//Reduce 个数，判断 loglevel 还是 logmodule 的统计，分配到不同的 Reduce if (key.toString().startsWith("logLevel::")) return 0; else if(key.toString().startsWith("moduleName::")) return 1; else return 0; else return 0; } } 在 Reduce 程序中，累加并合并 loglevel 和 logmodule 的出现次数 清单 7. Reduce 业务逻辑 public static class ReduceClass extends Reducer { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable values, Context context)throws IOException, InterruptedException { int tmp = 0; for (IntWritable val : values) { tmp = tmp + val.get(); } result.set(tmp); context.write(key, result);// 输出最后的汇总结果 } } 以上完成了 MapReduce 的主要处理逻辑，对于程序入口，我们使用 Hadoop 提供的 Tools 工具包方便的 进行 May-Reduce 程序的启动和 Map/Reduce 对应处理 class 的配置。 清单 8. Main 执行类 import java.io.File; import java.io.IOException; import java.text.SimpleDateFormat; import java.util.Date; import java.util.Iterator; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; Hadoop新MapReduce 框架 Yarn 详解 第 12 页 共 22 页 import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Partitioner; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class LogAnalysiser extends Configured implements Tool { public static void main(String[] args) { try { int res; res = ToolRunner.run(new Configuration(),new LogAnalysiser(), args); System.exit(res); } catch (Exception e) { e.printStackTrace(); } } public int run(String[] args) throws Exception { if (args == null || args.length <2) { System.out.println("need inputpath and outputpath"); return 1; } String inputpath = args[0]; String outputpath = args[1]; String shortin = args[0]; String shortout = args[1]; if (shortin.indexOf(File.separator) >= 0) shortin = shortin.substring(shortin.lastIndexOf(File.separator)); if (shortout.indexOf(File.separator) >= 0) shortout = shortout.substring(shortout.lastIndexOf(File.separator)); SimpleDateFormat formater = new SimpleDateFormat("yyyy.MM.dd.HH.mm"); shortout = new StringBuffer(shortout).append("-") .append(formater.format(new Date())).toString(); if (!shortin.startsWith("/")) shortin = "/" + shortin; Hadoop新MapReduce 框架 Yarn 详解 第 13 页 共 22 页 if (!shortout.startsWith("/")) shortout = "/" + shortout; shortin = "/user/oracle/dfs/" + shortin; shortout = "/user/oracle/dfs/" + shortout; File inputdir = new File(inputpath); File outputdir = new File(outputpath); if (!inputdir.exists() || !inputdir.isDirectory()) { System.out.println("inputpath not exist or isn't dir!"); return 0; } if (!outputdir.exists()) { new File(outputpath).mkdirs(); } // 以下注释的是 hadoop 0.20.X 老版本的 Job 代码，在 hadoop0.23.X 新框架中已经大大简化 // Configuration conf = getConf(); // JobConf job = new JobConf(conf, LogAnalysiser.class); // JobConf conf = new JobConf(getConf(),LogAnalysiser.class);// 构建 Config // conf.setJarByClass(MapClass.class); // conf.setJarByClass(ReduceClass.class); // conf.setJarByClass(PartitionerClass.class); // conf.setJar("hadoopTest.jar"); // job.setJar("hadoopTest.jar"); // 以下是新的 hadoop 0.23.X Yarn 的 Job 代码 job job = new Job(new Configuration()); job.setJarByClass(LogAnalysiser.class); job.setJobName("analysisjob"); job.setOutputKeyClass(Text.class);// 输出的 key 类型，在 OutputFormat 会检查 job.setOutputValueClass(IntWritable.class); // 输出的 value 类型，在 OutputFormat 会检查 job.setJarByClass(LogAnalysiser.class); job.setMapperClass(MapClass.class); job.setCombinerClass(ReduceClass.class); job.setReducerClass(ReduceClass.class); job.setPartitionerClass(PartitionerClass.class); job.setNumReduceTasks(2);// 强制需要有两个 Reduce 来分别处理流量和次数的统计 FileInputFormat.setInputPaths(job, new Path(shortin));//hdfs 中的输入路径 FileOutputFormat.setOutputPath(job,new Path(shortout));//hdfs 中输出路径 Date startTime = new Date(); System.out.println("Job started: " + startTime); job.waitForCompletion(true); Date end_time = new Date(); Hadoop新MapReduce 框架 Yarn 详解 第 14 页 共 22 页 System.out.println("Job ended: " + end_time); System.out.println("The job took " + (end_time.getTime() - startTime.getTime()) /1000 + " seconds."); // 删除输入和输出的临时文件 // fileSys.copyToLocalFile(new Path(shortout),new Path(outputpath)); // fileSys.delete(new Path(shortin),true); // fileSys.delete(new Path(shortout),true); return 0; } } Demo 部署及运行 Demo 输入输出的控制 本 demo 中我们将从 Weblogic 日志目录中拷贝原始待处理日志文件作为 Yarn 程序的输入，使用 hadoop dfs 命令将其放入分布式目录的 input 目录，处理完后将生成以时间戳为文件目录后缀的输出目录 Weblogic 日志存放的原始目录位于： /u01/app/Oracle/Middleware/user_projects/domains/test_domain/AdminServer/logs 分布式文件系统中的输入目录：/user/oracle/dfs/input 分布式文件系统中的输出目录：/user/oracle/dfs/output_%YYYY-MM-DD-hh-mm% Demo 打包和部署 可以使用 JDeveloper 或者 Eclipse 等 IDE 工具将开发的 Hadoop Demo 代码打包为 jar，并指定 Main 类为 LoyAnalyze，本文中我们采用 JDeveloper 打包 Demo 代码，如下图示例： 图 4.Yarn Demo 程序打包示例 Demo 执行与跟踪 我们在 OEL 主机（NameNode&ResourceManager 主机，192.168.137.8）上启动 dfs 分布式文件系统： Hadoop新MapReduce 框架 Yarn 详解 第 15 页 共 22 页 图 5. 启动 Demo dfs 文件系统 从上图可以看出 dfs 分布式文件系统已经在 OEL 和 Stephen 主机上成功启动，我们通过默认的分布式文 件系统 Web 监控 端口 http://192.168.137.8:50070（也可以在上文中 core-site.xml 中配置 dfs.namenode.http-address 项指定其他端口 ) 来验证其文件系统情况： 图 6.hadoop 文件系统 web 监控页面 从上图中我们可以看到 /user/oracle/dfs 分布式文件系统已成功建立。 接下来我们在 NameNode 主机（OEL，192.168.137.8）上启动 Yarn 框架： Hadoop新MapReduce 框架 Yarn 详解 第 16 页 共 22 页 图 7. 启动 Demo Yarn 框架 从上图我们可以看到 ResouceManager 在 OEL 主机上成功启动，NodeManager 进程在 Stephen 节点主 机上也已经启动，至此整个新的 Hadoop Yarn 框架已经成功启动。 我们将打好的 testHadoop.jar 包上传至 NameNode 主机（OEL）的 /hadoop/hadoop-0.23.0/ 根目录下， 我们使用 Hadoop 自带的 hadoop 命令行