我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。 在文章《MapReduce V1：TaskTracker设计要点概要分析》中我们已经了解了org.apache.hadoop.mapred.Child启动的基本流程，在Child VM启动的过程中会运行MapTask，实际是运行用户编写的MapReduce程序中的map方法中的处理逻辑，我们首先看一下，在Child类中，Child基于TaskUmbilicalProtocol协议与TaskTracker通信，获取到该Child VM需要加载的Task相关数据，包括Task本身，代码如下所示： final TaskUmbilicalProtocol umbilical = taskOwner.doAs(new PrivilegedExceptionAction<TaskUmbilicalProtocol>() { @Override public TaskUmbilicalProtocol run() throws Exception { // 建立Child到TaskTracker的RPC连接 return (TaskUmbilicalProtocol)RPC.getProxy(TaskUmbilicalProtocol.class, TaskUmbilicalProtocol.versionID, address, defaultConf); } }); ... ... JvmContext context
按标签浏览文章: MapReduce
MapReduce V1：TaskTracker端启动Task流程分析
我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。 TaskTracker周期性地向JobTracker发送心跳报告，在RPC调用返回结果后，解析结果得到JobTracker下发的运行Task的指令，即LaunchTaskAction，就会在TaskTracker节点上准备运行这个Task。Task的运行是在一个与TaskTracker进程隔离的JVM实例中执行，该JVM实例是通过org.apache.hadoop.mapred.Child来创建的，所以在创建Child VM实例之前，需要做大量的准备工作来启动Task运行。一个Task的启动过程，如下序列图所示： 通过上图，结合源码，我们将一个Task启动的过程，分为下面3个主要的步骤： 初始化跟踪Task运行的相关数据结构 准备Task运行所共享的Job资源 启动Task 下面，我们详细分析上面3个步骤的流程： 初始化跟踪Task运行的相关数据结构 如果是LaunchTaskAction，则TaskTracker会将该指令加入到一个启动Task的队列中，进行一步加载处理，如下所示： private void addToTaskQueue(LaunchTaskAction action) { if (action.getTask().isMapTask()) { mapLauncher.addToTaskQueue(action);
MapReduce V1：TaskTracker设计要点概要分析
我们基于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中各个组件都负责处理哪些工作，我们通过下表来简要地说明各
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
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集合类型的高层抽象就包含
Hadoop MapReduce处理海量小文件：压缩文件
在HDFS上存储文件，大量的小文件是非常消耗NameNode内存的，因为每个文件都会分配一个文件描述符，NameNode需要在启动的时候加载全部文件的描述信息，所以文件越多，对 NameNode来说开销越大。 我们可以考虑，将小文件压缩以后，再上传到HDFS中，这时只需要一个文件描述符信息，自然大大减轻了NameNode对内存使用的开销。MapReduce计算中，Hadoop内置提供了如下几 种压缩格式： DEFLATE gzip bzip2 LZO 使用压缩文件进行MapReduce计算，它的开销在于解压缩所消耗的时间，在特定的应用场景中这个也是应该考虑的问题。不过对于海量小文件的应用场景，我们压缩了小文件，却换 来的Locality特性。 假如成百上千的小文件压缩后只有一个Block，那么这个Block必然存在一个DataNode节点上，在计算的时候输入一个InputSplit，没有网络间传输数据的开销，而且是在本地进行 运算。倘若直接将小文件上传到HDFS上，成百上千的小Block分布在不同DataNode节点上，为了计算可能需要“移动数据”之后才能进行计算。文件很少的情况下，除了NameNode内 存使用开销以外，可能感觉不到网
Hadoop MapReduce处理海量小文件：基于CombineFileInputFormat
在使用Hadoop处理海量小文件的应用场景中，如果你选择使用CombineFileInputFormat，而且你是第一次使用，可能你会感到有点迷惑。虽然，从这个处理方案的思想上很容易理解，但是可能会遇到这样那样的问题。 使用CombineFileInputFormat作为Map任务的输入规格描述，首先需要实现一个自定义的RecordReader。 CombineFileInputFormat的大致原理是，他会将输入多个数据文件（小文件）的元数据全部包装到CombineFileSplit类里面。也就是说，因为小文件的情况下，在HDFS中都是单Block的文件，即一个文件一个Block，一个CombineFileSplit包含了一组文件Block，包括每个文件的起始偏移（offset），长度（length），Block位置（localtions）等元数据。如果想要处理一个CombineFileSplit，很容易想到，对其包含的每个InputSplit（实际上这里面没有这个，你需要读取一个小文件块的时候，需要构造一个FileInputSplit对象）。 在执行MapReduce任务的时候，需要读取文件的文本行（简单一点是文本行，也可能是其他格式数据）。那么对于CombineFileSplit来说，你需要处理其包含的小文
Hadoop MapReduce处理海量小文件：自定义InputFormat和RecordReader
一般来说，基于Hadoop的MapReduce框架来处理数据，主要是面向海量大数据，对于这类数据，Hadoop能够使其真正发挥其能力。对于海量小文件，不是说不能使用Hadoop来处理，只不过直接进行处理效率不会高，而且海量的小文件对于HDFS的架构设计来说，会占用NameNode大量的内存来保存文件的元数据（Bookkeeping）。另外，由于文件比较小，我们是指远远小于HDFS默认Block大小（64M），比如1k~2M，都很小了，在进行运算的时候，可能无法最大限度地充分Locality特性带来的优势，导致大量的数据在集群中传输，开销很大。 但是，实际应用中，也存在类似的场景，海量的小文件的处理需求也大量存在。那么，我们在使用Hadoop进行计算的时候，需要考虑将小数据转换成大数据，比如通过合并压缩等方法，可以使其在一定程度上，能够提高使用Hadoop集群计算方式的适应性。Hadoop也内置了一些解决方法，而且提供的API，可以很方便地实现。 下面，我们通过自定义InputFormat和RecordReader来实现对海量小文件的并行处理。 基本思路描述如下： 在Mapper中将小文件合并，输出结果的文件中每
Hadoop MapReduce编程：计算极值
现在，我们不是计算一个最大值了（想了解如何计算最大值，可以参考Hadoop MapReduce编程：计算最大值），而是计算一个最大值和一个最小值。实际上，实现Mapper和Reducer也是非常简单的，但是我们要输出结果，需要能够区分出最大值和最小值，并同时输出结果，这就需要自定义自己的输出类型，以及需要定义输出格式。 测试数据 数据格式，如下所示： SG 253654006139495 253654006164392 619850464 KG 253654006225166 253654006252433 743485698 UZ 253654006248058 253654006271941 570409379 TT 253654006282019 253654006286839 23236775 BE 253654006276984 253654006301435 597874033 BO 253654006293624 253654006315946 498265375 SR 253654006308428 253654006330442 484613339 SV 253654006320312 253654006345405 629640166 LV 253654006330384 253654006359891 870680704 FJ 253654006351709 253654006374468 517965666 上面文本数据一行一行存储，一行包含4部分，分别表示： 国家代码 起始时间 截止时间 随机成本/权重估值 各个字段之间以