我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。 本文不打算深入地详细分析TaskTracker某个具体的处理流程，而是概要地分析TaskTracker在MapReduce框架中的主要负责处理那些事情，是我们能够在宏观上了解TaskTracker端都做了哪些工作。我尽量将TaskTracker端的全部要点内容提出来，但是涉及到详细的分析，只是点到为止，后续会对相应模块的处理流程结合代码进行分析。 TaskTracker主要负责MapReduce计算集群中Task运行的管理，所以TaskTracker要管理的事情比较多。一个MapReduce Job由很多的Task组成，而一个Job的所有Task被分成几个相斥的子集，每个子集被分配到某一个TaskTracker上去运行，所以一个TaskTracker管理运行了一个Job的所有Task的一个子集，也就是说TaskTracker不仅要维护每个Job对应的一个Task的子集，还要维护这些Task所属的Job的运行状态，对于Job/Task的状态的管理都是与JobTracker通过RPC通信保持状态的同步。 下面是TaskTracker端的主要组件，如下图所示： 为了了解TaskTracker中各个组件都负责处理哪些工作，我们通过下表来简要地说明各
按标签浏览文章: Hadoop/Hive/ZooKeeper
MapReduce V1：JobTracker处理Heartbeat流程分析
我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。这篇文章的内容，更多地主要是描述处理/交互流程性的东西，大部分流程图都是经过我梳理后画出来的（开始我打算使用序列图来描述流程，但是发现很多流程在单个对象内部都已经非常复杂，想要通过序列图表达有点担心描述不清，所以选择最基本的程序流程图），可能看起来比较枯燥，重点还是关注主要的处理流程要点，特别的地方我会刻意标示出来，便于理解。 JobTracker与TaskTracker之间通过org.apache.hadoop.mapred.InterTrackerProtocol协议来进行通信，TaskTracker通过该接口进行远程调用实现Heartbeat消息的发送，协议方法定义如下所示： HeartbeatResponse heartbeat(TaskTrackerStatus status, boolean restarted, boolean initialContact, boolean acceptNewTasks, short responseId) throws IOException; 通过该方法可以看出，最核心的Heartbeat报告数据都封装在Ta
MapReduce V1：JobTracker端Job/Task数据结构
我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。在MapReduce程序运行的过程中，JobTracker端会在内存中维护一些与Job/Task运行相关的信息，了解这些内容对分析MapReduce程序执行流程的源码会非常有帮助。 在编写MapReduce程序时，我们是以Job为单位进行编程处理，一个应用程序可能由一组Job组成，而MapReduce框架给我们暴露的只是一些Map和Reduce的函数接口，在运行期它会构建对应MapTask和ReduceTask，所以我们知道一个Job是由一个或多个MapTask，以及0个或1个ReduceTask组成。而对于MapTask，它是根据输入的数据文件的的逻辑分片（InputSplit）而定的，通常有多少个分片就会有多少个MapTask；而对于ReduceTask，它会根据我们编写的MapReduce程序配置的个数来运行。 有了这些信息，我们能够预想到，在Job运行过程中，无非也需要维护与这些Job/Task相关的一些状态信息，通过一定的调度策略来管理Job/Task的运行。这里，我们主要关注JobTracker端的一些非常有用的数据结构：JobTracker、JobInProgress、TaskInProgress，来熟悉各种数据结构的定义及作用。 数据
MapReduce V1：Job提交流程之JobTracker端分析
我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。MapReduce V1实现中，主要存在3个主要的分布式进程（角色）：JobClient、JobTracker和TaskTracker，我们主要是以这三个角色的实际处理活动为主线，并结合源码，分析实际处理流程。 上一篇我们分析了Job提交过程中JobClient端的处理流程（详见文章 MapReduce V1：Job提交流程之JobClient端分析），这里我们继续详细分析Job提交在JobTracker端的具体流程。通过阅读源码可以发现，这部分的处理逻辑还是有点复杂，经过梳理，更加细化清晰的流程，如下图所示： 上图中主要分为两大部分：一部分是JobClient基于RPC调用提交Job到JobTracker后，在JobTracker端触发TaskScheduler所注册的一系列Listener进行Job信息初始化；另一部分是JobTracker端监听Job队列的线程，监听到Job状态发生变更触发一系列Listener更新状态。我们从这两个方面展开分析： JobTracker接收Job提交 JobTracker接收到JobClient提交的Job，在JobTracker端具体执行流程，描述如下： JobClient基于JobSubmissionProtocol协议远程调用JobTracker的s
MapReduce V1：Job提交流程之JobClient端分析
我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。 MapReduce V1实现中，主要存在3个主要的分布式进程（角色）：JobClient、JobTracker和TaskTracker，我们主要是以这三个角色的实际处理活动为主线，并结合源码，分析实际处理流程。下图是《Hadoop权威指南》一书给出的MapReduce V1处理Job的抽象流程图： 如上图，我们展开阴影部分的处理逻辑，详细分析Job提交在JobClient端的具体流程。 在编写好MapReduce程序以后，需要将Job提交给JobTracker，那么我们就需要了解在提交Job的过程中，在JobClient端都做了哪些工作，或者说执行了哪些处理。在JobClient端提交Job的处理流程，如下图所示： 上图所描述的Job的提交流程，说明如下所示： 在MR程序中创建一个Job实例，设置Job状态 创建一个JobClient实例，准备将创建的Job实例提交到JobTracker 在创建JobClient的过程中，首先必须保证建立到JobTracker的RPC连接 基于JobSubmissionProtocol协议远程调用JobTracker获取一个新的Job ID 根据MR程序中配置的Job，在HDFS上创建Job相关目录，并将配置的tmpfiles、tmpja
Hadoop YARN架构设计要点
YARN是开源项目Hadoop的一个资源管理系统，最初设计是为了解决Hadoop中MapReduce计算框架中的资源管理问题，但是现在它已经是一个更加通用的资源管理系统，可以把MapReduce计算框架作为一个应用程序运行在YARN系统之上，通过YARN来管理资源。如果你的应用程序也需要借助YARN的资源管理功能，你也可以实现YARN提供的编程API，将你的应用程序运行于YARN之上，将资源的分配与回收统一交给YARN去管理，可以大大简化资源管理功能的开发。当前，也有很多应用程序已经可以构建于YARN之上，如Storm、Spark等计算框架。 YARN整体架构 YARN是基于Master/Slave模式的分布式架构，我们先看一下，YARN的架构设计，如图所示（来自官网文档）： 上图，从逻辑上定义了YARN系统的核心组件和主要交互流程，各个组件说明如下： YARN Client YARN Client提交Application到RM，它会首先创建一个Application上下文件对象，并设置AM必需的资源请求信息，然后提交到RM。YARN Client也可以与RM通信，获取到一个已经提交并运行的Application的状态信息等，具体详见后面ApplicationClientPro
Apache Crunch：简化编写MapReduce Pipeline程序
Apache Crunch提供了一套Java API，能够简化编写、测试、运行MapReduce Pipeline程序。Crunch的基本思想是隐藏编写MapReduce程序的细节，基于函数式编程的思想，定义了一套函数式编程接口，因为Java并不支持函数式编程，只能通过回调的方式来实现，虽然写起来代码不够美观简洁，但是编写MapReduce程序的思路是非常清晰的，而且比编写原生的MapReduce程序要容易地多。如果直接使用MapReduce API编写一个复杂的Pipeline程序，可能需要考虑好每个Job的细节（Map和Reduce的实现内容），而使用Crunch变成库来编写，只需要清晰地控制好要实现的业务逻辑处理的操作流程，调用Crunch提供的接口（类似函数操作的算子、如union、join、filter、groupBy、sort等等）。 下面，我们简单说明一下Crunch提供的一些功能或内容： Crunch集合及操作 我们看一下Crunch提供的用来在处理分布式数据集的集合类型的抽象定义，如下面类图所示： 上面，我给出了集合类对应的方法签名，其中具有相同名称签名的方法还具有重载的其他方法签名（参数列表不同），Crunch集合类型的高层抽象就包含
Hue安装配置实践
Hue是一个开源的Apache Hadoop UI系统，最早是由Cloudera Desktop演化而来，由Cloudera贡献给开源社区，它是基于Python Web框架Django实现的。通过使用Hue我们可以在浏览器端的Web控制台上与Hadoop集群进行交互来分析处理数据，例如操作HDFS上的数据，运行MapReduce Job等等。很早以前就听说过Hue的便利与强大，一直没能亲自尝试使用，下面先通过官网给出的特性，通过翻译原文简单了解一下Hue所支持的功能特性集合： 默认基于轻量级sqlite数据库管理会话数据，用户认证和授权，可以自定义为MySQL、Postgresql，以及Oracle 基于文件浏览器（File Browser）访问HDFS 基于Hive编辑器来开发和运行Hive查询 支持基于Solr进行搜索的应用，并提供可视化的数据视图，以及仪表板（Dashboard） 支持基于Impala的应用进行交互式查询 支持Spark编辑器和仪表板（Dashboard） 支持Pig编辑器，并能够提交脚本任务 支持Oozie编辑器，可以通过仪表板提交和监控Workflow、Coordinator和Bundle 支持HBase浏览器，能够可视化数据、查询数据、修改HBase表 支持Metastore浏览器，可以访问
基于Hadoop SLA认证机制实现权限控制
Hadoop集群上存储数据，同时基于MapReduce计算框架可以实现计算任务，那么无论是从数据保护的角度，还是从提交计算任务占用资源的角度来看，都需要存在一种权限管理与分配机制，能够很好地限制哪些人可以在HDFS上存储数据，哪些人可以利用集群的资源来处理特定的计算任务。当然，如果能够非常完美地解决这些问题是最好的。当前Hadoop本身提供的权限管理功能还不能满足普遍的需要，或者我们从Hadoop已有的一些简单或复杂的认证机制选择适合自己所在组织机构需要的，或者我们在外围开发一些权限管理系统与Hadoop整合作为补充。 对比Kerberos认证（Authentication）配置方式与SLA授权（Service Level Authorization）方式，Kerberos配置相当复杂，而且还要依赖于外部的密钥分发中心KDC（Key Distribution Center）服务器，如果KDC出现问题，那么就会导致依赖于KDC认证的整个Hadoop集群无法使用，鉴于此，对于一些相对小的开发团队来说还是更倾向于粗粒度的Hadoop SLA授权机制。 Hadoop SLA基于Hadoop的各种服务（基于协议来划分）与Linux系统的用户、用户组来实现。Had
HDFS读文件过程分析：读取文件的Block数据
我们可以从java.io.InputStream类中看到，抽象出一个read方法，用来读取已经打开的InputStream实例中的字节，每次调用read方法，会读取一个字节数据，该方法抽象定义，如下所示： public abstract int read() throws IOException; Hadoop的DFSClient.DFSInputStream类实现了该抽象逻辑，如果我们清楚了如何从HDFS中读取一个文件的一个block的一个字节的原理，更加抽象的顶层只需要迭代即可获取到该文件的全部数据。 从HDFS读文件过程分析：获取文件对应的Block列表（http://shiyanjun.cn/archives/925.html）中，我们已经获取到一个文件对应的Block列表信息，打开一个文件，接下来就要读取实际的物理块数据，我们从下面的几个方面来详细说明读取数据的过程。 Client从Datanode读取文件的一个字节 下面，我们通过分析DFSClient.DFSInputStream中实现的代码，读取HDFS上文件的内容。首先从下面的方法开始： @Override public synchronized int read() throws IOException { int ret = read( oneByteBuf, 0, 1 ); return ( ret <= 0 ) ? -
HDFS 写文件过程分析
HDFS 是一个分布式文件系统，在 HDFS 上写文件的过程与我们平时使用的单机文件系统非常不同，从宏观上来看，在 HDFS 文件系统上创建并写一个文件，流程如下图（来自《Hadoop：The Definitive Guide》一书）所示： 具体过程描述如下： Client 调用 DistributedFileSystem 对象的 create 方法，创建一个文件输出流（FSDataOutputStream）对象 通过 DistributedFileSystem 对象与 Hadoop 集群的 NameNode 进行一次 RPC 远程调用，在 HDFS 的 Namespace 中创建一个文件条目（Entry），该条目没有任何的 Block 通过 FSDataOutputStream 对象，向 DataNode 写入数据，数据首先被写入 FSDataOutputStream 对象内部的 Buffer 中，然后数据被分割成一个个 Packet 数据包 以 Packet 最小单位，基于 Socket 连接发送到按特定算法选择的 HDFS 集群中一组 DataNode（正常是 3 个，可能大于等于 1）中的一个节点上，在这组 DataNode 组成的 Pipeline 上依次传输 Packet 这组 DataNode 组成的 Pipeline 反方向上，发送 ack，最终由 Pipeline 中第一个 DataNode 节点将 Pipeline a
HDFS读文件过程分析：获取文件对应的Block列表
在使用Java读取一个文件系统中的一个文件时，我们会首先构造一个DataInputStream对象，然后就能够从文件中读取数据。对于存储在HDFS上的文件，也对应着类似的工具类，但是底层的实现逻辑却是非常不同的。我们先从使用DFSClient.DFSDataInputStream类来读取HDFS上一个文件的一段代码来看，如下所示： package org.shirdrn.hadoop.hdfs; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; public class HdfsFileReader { public static void main(String[] args) { String file = "hdfs://hadoop-cluster-m:8020/data/logs/basis_user_behavior/201405071237_10_10_1_73.log"; Path path = new Path(file); Configuration conf = new Configuration();