Hadoop源代码分析

Hadoop源代码分析（一） 关键字: 分布式 云计算 Google的核心竞争技术是它的计算平台。Google的大牛们用了下面5篇文章，介绍了它们的计算设施。 GoogleCluster： http://research.google.com/archive/googlecluster.html Chubby：http://labs.google.com/papers/chubby.html GFS：http://labs.google.com/papers/gfs.html BigTable：http://labs.google.com/papers/bigtable.html MapReduce：http://labs.google.com/papers/mapreduce.html 很快，Apache上就出现了一个类似的解决方案，目前它们都属于Apache的Hadoop项目，对应的分别是： Chubby-->ZooKeeper GFS-->HDFS BigTable-->HBase MapReduce-->Hadoop 目前，基于类似思想的Open Source项目还很多，如Facebook用于用户分析的Hive。 HDFS作为一个分布式文件系统，是所有这些项目的基础。分析好HDFS，有利于了解其他系统。由于Hadoop的HDFS和MapReduce是同一个项目，我们就把他们放在一块，进行分析。 下图是MapReduce整个项目的顶层包图和他们的依赖关系。Hadoop包之间的依赖关系比较复杂，原因是HDFS提供了一个分布式文件系统，该系统提供API，可以屏蔽本地文件系统和分布式文件系统，甚至象Amazon S3这样的在线存储系统。这就造成了分布式文件系统的实现，或者是分布式文件系统的底层的实现，依赖于某些貌似高层的功能。功能的相互引用，造成了蜘蛛网型的依赖关系。一个典型的例子就是包conf，conf用于读取系统配置，它依赖于fs，主要是读取配置文件的时候，需要使用文件系统，而部分的文件系统的功能，在包fs中被抽象了。 Hadoop的关键部分集中于图中蓝色部分，这也是我们考察的重点。 · · 大小: 78.3 KB Hadoop源代码分析（二） 下面给出了Hadoop的包的功能分析。 Package Dependences tool 提供一些命令行工具，如DistCp，archive mapreduce Hadoop的Map/Reduce实现 filecache 提供HDFS文件的本地缓存，用于加快Map/Reduce的数据访问速度 fs 文件系统的抽象，可以理解为支持多种文件系统实现的统一文件访问接口 hdfs HDFS，Hadoop的分布式文件系统实现 ipc 一个简单的IPC的实现，依赖于io提供的编解码功能 参考：http://zhangyu8374.javaeye.com/blog/86306 io 表示层。将各种数据编码/解码，方便于在网络上传输 net 封装部分网络功能，如DNS，socket security 用户和用户组信息 conf 系统的配置参数 metrics 系统统计数据的收集，属于网管范畴 util 工具类 record 根据DDL（数据描述语言）自动生成他们的编解码函数，目前可以提供C++和Java http 基于Jetty的HTTP Servlet，用户通过浏览器可以观察文件系统的一些状态信息和日志 log 提供HTTP访问日志的HTTP Servlet Hadoop源代码分析（三） 由于Hadoop的MapReduce和HDFS都有通信的需求，需要对通信的对象进行序列化。Hadoop并没有采用Java的序列化，而是引入了它自己的系统。 org.apache.hadoop.io中定义了大量的可序列化对象，他们都实现了Writable接口。实现了Writable接口的一个典型例子如下： Java代码 1. public class MyWritable implements Writable { 2. // Some data 3. private int counter; 4. private long timestamp; 5. 6. public void write(DataOutput out) throws IOException { 7. out.writeInt(counter); 8. out.writeLong(timestamp); 9. } 10. 11. public void readFields(DataInput in) throws IOException { 12. counter = in.readInt(); 13. timestamp = in.readLong(); 14. } 15. 16. public static MyWritable read(DataInput in) throws IOException { 17. MyWritable w = new MyWritable(); 18. w.readFields(in); 19. return w; 20. } 21. } public class MyWritable implements Writable { // Some data private int counter; private long timestamp; public void write(DataOutput out) throws IOException { out.writeInt(counter); out.writeLong(timestamp); } public void readFields(DataInput in) throws IOException { counter = in.readInt(); timestamp = in.readLong(); } public static MyWritable read(DataInput in) throws IOException { MyWritable w = new MyWritable(); w.readFields(in); return w; } } 其中的write和readFields分别实现了把对象序列化和反序列化的功能，是Writable接口定义的两个方法。下图给出了庞大的org.apache.hadoop.io中对象的关系。 这里，我把ObjectWritable标为红色，是因为相对于其他对象，它有不同的地位。当我们讨论Hadoop的RPC时，我们会提到RPC上交换的信息，必须是Java的基本类型，String和Writable接口的实现类，以及元素为以上类型的数组。ObjectWritable对象保存了一个可以在RPC上传输的对象和对象的类型信息。这样，我们就有了一个万能的，可以用于客户端/服务器间传输的Writable对象。例如，我们要把上面例子中的对象作为RPC请求，需要根据MyWritable创建一个ObjectWritable，ObjectWritable往流里会写如下信息 对象类名长度，对象类名，对象自己的串行化结果 这样，到了对端，ObjectWritable可以根据对象类名创建对应的对象，并解串行。应该注意到，ObjectWritable依赖于WritableFactories，那存储了Writable子类对应的工厂。我们需要把MyWritable的工厂，保存在WritableFactories中（通过WritableFactories.setFactory）。 Hadoop源代码分析（五） 介绍完org.apache.hadoop.io以后，我们开始来分析org.apache.hadoop.rpc。RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。当我们讨论HDFS的，通信可能发生在： · Client-NameNode之间，其中NameNode是服务器 · Client-DataNode之间，其中DataNode是服务器 · DataNode-NameNode之间，其中NameNode是服务器 · DataNode-DateNode之间，其中某一个DateNode是服务器，另一个是客户端 如果我们考虑Hadoop的Map/Reduce以后，这些系统间的通信就更复杂了。为了解决这些客户机/服务器之间的通信，Hadoop引入了一个RPC框架。该RPC框架利用的Java的反射能力，避免了某些RPC解决方案中需要根据某种接口语言（如CORBA的IDL）生成存根和框架的问题。但是，该RPC框架要求调用的参数和返回结果必须是Java的基本类型，String和Writable接口的实现类，以及元素为以上类型的数组。同时，接口方法应该只抛出IOException异常。（参考自http://zhangyu8374.javaeye.com/blog/86306） 既然是RPC，当然就有客户端和服务器，当然，org.apache.hadoop.rpc也就有了类Client和类Server。但是类Server是一个抽象类，类RPC封装了Server，利用反射，把某个对象的方法开放出来，变成RPC中的服务器。 下图是org.apache.hadoop.rpc的类图。 · · 大小: 130.3 KB Hadoop源代码分析（六） 既然是RPC，自然就有客户端和服务器，当然，org.apache.hadoop.rpc也就有了类Client和类Server。在这里我们来仔细考察org.apache.hadoop.rpc.Client。下面的图包含了org.apache.hadoop.rpc.Client中的关键类和关键方法。 由于Client可能和多个Server通信，典型的一次HDFS读，需要和NameNode打交道，也需要和某个/某些DataNode通信。这就意味着某一个Client需要维护多个连接。同时，为了减少不必要的连接，现在Client的做法是拿ConnectionId（图中最右侧）来做为Connection的ID。ConnectionId包括一个InetSocketAddress（IP地址+端口号或主机名+端口号）对象和一个用户信息对象。这就是说，同一个用户到同一个InetSocketAddress的通信将共享同一个连接。 连接被封装在类Client.Connection中，所有的RPC调用，都是通过Connection，进行通信。一个RPC调用，自然有输入参数，输出参数和可能的异常，同时，为了区分在同一个Connection上的不同调用，每个调用都有唯一的id。调用是否结束也需要一个标记，所有的这些都体现在对象Client.Call中。Connection对象通过一个Hash表，维护在这个连接上的所有Call： Java代码 1. private Hashtable calls = new Hashtable(); private Hashtable calls = new Hashtable(); 一个RPC调用通过addCall，把请求加到Connection里。为了能够在这个框架上传输Java的基本类型，String和Writable接口的实现类，以及元素为以上类型的数组，我们一般把Call需要的参数打包成为ObjectWritable对象。 Client.Connection会通过socket连接服务器，连接成功后回校验客户端/服务器的版本号（Client.ConnectionwriteHeader()方法），校验成功后就可以通过Writable对象来进行请求的发送/应答了。注意，每个Client.Connection会起一个线程，不断去读取socket，并将收到的结果解包，找出对应的Call，设置Call并通知结果已经获取。 Call使用Obejct的wait和notify，把RPC上的异步消息交互转成同步调用。 还有一点需要注意，一个Client会有多个Client.Connection，这是一个很自然的结果。 Hadoop源代码分析（七） 聊完了Client聊Server，按惯例，先把类图贴出来。 需要注意的是，这里的Server类是个抽象类，唯一抽象的地方，就是 Java代码 1. public abstract Writable call(Writable param, long receiveTime) throws IOException; public abstract Writable call(Writable param, long receiveTime) throws IOException; 这表明，Server提供了一个架子，Server的具体功能，需要具体类来完成。而具体类，当然就是实现call方法。 我们先来分析Server.Call，和Client.Call类似，Server.Call包含了一次请求，其中，id和param的含义和Client.Call是一致的。不同点在后面三个属性，connection是该Call来自的连接，当然，当请求处理结束时，相应的结果会通过相同的connection，发送给客户端。属性timestamp是请求到达的时间戳，如果请求很长时间没被处理，对应的连接会被关闭，客户端也就知道出错了。最后的response是请求处理的结果，可能是一个Writable的串行化结果，也可能一个异常的串行化结果。 Server.Connection维护了一个来之客户端的socket连接。它处理版本校验，读取请求并把请求发送到请求处理线程，接收处理结果并把结果发送给客户端。 Hadoop的Server采用了Java的NIO，这样的话就不需要为每一个socket连接建立一个线程，读取socket上的数据。在Server中，只需要一个线程，就可以accept新的连接请求和读取socket上的数据，这个线程，就是上面图里的Listener。 请求处理线程一般有多个，它们都是Server.Handle类的实例。它们的run方法循环地取出一个Server.Call，调用Server.call方法，搜集结果并串行化，然后将结果放入Responder队列中。 对于处理完的请求，需要将结果写回去，同样，利用NIO，只需要一个线程，相关的逻辑在Responder里。 Hadoop源代码分析（八） （注：本节需要用到一些Java反射的背景） 有了Client和Server，很自然就能RPC啦。下面轮到RPC.java啦。 一般来说，分布式对象一般都会要求根据接口生成存根和框架。如CORBA，可以通过IDL，生成存根和框架。但是，在org.apache.hadoop.rpc，我们就不需要这样的步骤了。上类图。 为了分析Invoker，我们需要介绍一些Java反射实现Dynamic Proxy的背景。 Dynamic Proxy是由两个class实现的：java.lang.reflect.Proxy 和 java.lang.reflect.InvocationHandler，后者是一个接口。所谓Dynamic Proxy是这样一种class：它是在运行时生成的class，在生成它时你必须提供一组interface给它，然后该class就宣称它实现了这些interface。 这个Dynamic Proxy其实就是一个典型的Proxy模式，它不会替你作实质性的工作，在生成它的实例时你必须提供一个handler，由它接管实际的工作。这个handler，在Hadoop的RPC中，就是Invoker对象。 我们可以简单地理解：就是你可以通过一个接口来生成一个类，这个类上的所有方法调用，都会传递到你生成类时传递的InvocationHandler实现中。 在Hadoop的RPC中，Invoker实现了InvocationHandler的invoke方法（invoke方法也是InvocationHandler的唯一方法）。Invoker会把所有跟这次调用相关的调用方法名，参数类型列表，参数列表打包，然后利用前面我们分析过的Client，通过socket传递到服务器端。就是说，你在proxy类上的任何调用，都通过Client发送到远方的服务器上。 Invoker使用Invocation。Invocation封装了一个远程调用的所有相关信息，它的主要属性有: methodName，调用方法名，parameterClasses，调用方法参数的类型列表和parameters，调用方法参数。注意，它实现了Writable接口，可以串行化。 RPC.Server实现了org.apache.hadoop.ipc.Server，你可以把一个对象，通过RPC，升级成为一个服务器。服务器接收到的请求（通过Invocation），解串行化以后，就变成了方法名，方法参数列表和参数列表。利用Java反射，我们就可以调用对应的对象的方法。调用的结果再通过socket，返回给客户端，客户端把结果解包后，就可以返回给Dynamic Proxy的使用者了。 Hadoop源代码分析（九） 一个典型的HDFS系统包括一个NameNode和多个DataNode。NameNode维护名字空间；而DataNode存储数据块。 DataNode负责存储数据，一个数据块在多个DataNode中有备份；而一个DataNode对于一个块最多只包含一个备份。所以我们可以简单地认为DataNode上存了数据块ID和数据块内容，以及他们的映射关系。 一个HDFS集群可能包含上千DataNode节点，这些DataNode定时和NameNode通信，接受NameNode的指令。为了减轻NameNode的负担，NameNode上并不永久保存那个DataNode上有那些数据块的信息，而是通过DataNode启动时的上报，来更新NameNode上的映射表。 DataNode和NameNode建立连接以后，就会不断地和NameNode保持心跳。心跳的返回其还也包含了NameNode对DataNode的一些命令，如删除数据库或者是把数据块复制到另一个DataNode。应该注意的是：NameNode不会发起到DataNode的请求，在这个通信过程中，它们是严格的客户端/服务器架构。 DataNode当然也作为服务器接受来自客户端的访问，处理数据块读/写请求。DataNode之间还会相互通信，执行数据块复制任务，同时，在客户端做写操作的时候，DataNode需要相互配合，保证写操作的一致性。 下面我们就来具体分析一下DataNode的实现。DataNode的实现包括两部分，一部分是对本地数据块的管理，另一部分，就是和其他的实体打交道。我们先来看本地数据块管理部分。 安装Hadoop的时候，我们会指定对应的数据块存放目录，当我们检查数据块存放目录目录时，我们回发现下面有个叫dfs的目录，所有的数据就存放在dfs/data里面。 其中有两个文件，storage里存的东西是一些出错信息，貌似是版本不对…云云。in_use.lock是一个空文件，它的作用是如果需要对整个系统做排斥操作，应用应该获取它上面的一个锁。 接下来是3个目录，current存的是当前有效的数据块，detach存的是快照（snapshot，目前没有实现），tmp保存的是一些操作需要的临时数据块。 但我们进入current目录以后，就会发现有一系列的数据块文件和数据块元数据文件。同时还有一些子目录，它们的名字是subdir0到subdir63，子目录下也有数据块文件和数据块元数据。这是因为HDFS限定了每个目录存放数据块文件的数量，多了以后会创建子目录来保存。 数据块文件显然保存了HDFS中的数据，数据块最大可以到64M。每个数据块文件都会有对应的数据块元数据文件。里面存放的是数据块的校验信息。下面是数据块文件名和它的元数据文件名的例子： blk_3148782637964391313 blk_3148782637964391313_242812.meta 上面的例子中，3148782637964391313是数据块的ID号，242812是数据块的版本号，用于一致性检查。 在current目录下还有下面几个文件： VERSION，保存了一些文件系统的元信息。 dncp_block_verification.log.curr和dncp_block_verification.log.prev，它记录了一些DataNode对文件系定时统做一致性检查需要的信息。 Hadoop源代码分析（一零） 在继续分析DataNode之前，我们有必要看一下系统的工作状态。启动HDFS的时候，我们可以选择以下启动参数： · FORMAT("-format")：格式化系统 · REGULAR("-regular")：正常启动 · UPGRADE("-upgrade")：升级 · ROLLBACK("-rollback")：回滚 · FINALIZE("-finalize")：提交 · IMPORT("-importCheckpoint")：从Checkpoint恢复。 作为一个大型的分布式系统，Hadoop内部实现了一套升级机制（http://wiki.apache.org/hadoop/Hadoop_Upgrade）。upgrade参数就是为了这个目的而存在的，当然，升级可能成功，也可能失败。如果失败了，那就用rollback进行回滚；如果过了一段时间，系统运行正常，那就可以通过finalize，正式提交这次升级(跟数据库有点像啊)。 importCheckpoint选项用于NameNode发生故障后，从某个检查点恢复。 有了上面的描述，我们得到下面左边的状态图： 大家应该注意到，上面的升级/回滚/提交都不可能一下就搞定，就是说，系统故障时，它可能处于上面右边状态中的某一个。特别是分布式的各个节点上，甚至可能出现某些节点已经升级成功，但有些节点可能处于中间状态的情况，所以Hadoop采用类似于数据库事务的升级机制也就不是很奇怪。 大家先理解一下上面的状态图，它是下面我们要介绍DataNode存储的基础。 Hadoop源代码分析（一一） 我们来看一下升级/回滚/提交时的DataNode上会发生什么（在类DataStorage中实现）。 前面我们提到过VERSION文件，它保存了一些文件系统的元信息，这个文件在系统升级时，会发生对应的变化。 升级时，NameNode会将新的版本号，通过DataNode的登录应答返回。DataNode收到以后，会将当前的数据块文件目录改名，从current改名为previous.tmp，建立一个snapshot，然后重建current目录。重建包括重建VERSION文件，重建对应的子目录，然后建立数据块文件和数据块元数据文件到previous.tmp的硬连接。建立硬连接意味着在系统中只保留一份数据块文件和数据块元数据文件，current和previous.tmp中的相应文件，在存储中，只保留一份。当所有的这些工作完成以后，会在current里写入新的VERSION文件，并将previous.tmp目录改名为previous，完成升级。 了解了升级的过程以后，回滚就相对简单。因为说有的旧版本信息都保存在previous目录里。回滚首先将current目录改名为removed.tmp，然后将previous目录改名为current，最后删除removed.tmp目录。 提交的过程，就是将上面的previous目录改名为finalized.tmp，然后启动一个线程，将该目录删除。 下图给出了上面的过程： 需要注意的是，HDFS的升级，往往只是支持从某一个特点的老版本升级到当前版本。回滚时能够恢复到的版本，也是previous中记录的版本。 下面我们继续分析DataNode。 文字分析完DataNode存储在文件上的数据以后，我们来看一下运行时对应的数据结构。从大到小，Hadoop中最大的结构是Storage，最小的结构，在DataNode上是block。 类Storage保存了和存储相关的信息，它继承了StorageInfo，应用于DataNode的DataStorage，则继承了Storage，总体类图如下： StorageInfo包含了3个字段，分别是layoutVersion：版本号，如果Hadoop调整文件结构布局，版本号就会修改，这样可以保证文件结构和应用一致。namespaceID是Storage的ID，cTime，creation time。 和StorageInfo相比，Storage就是个大家伙了。 Storage可以包含多个根（参考配置项dfs.data.dir的说明），这些根通过Storage的内部类StorageDirectory来表示。StorageDirectory中最重要的方法是analyzeStorage，它将根据系统启动时的参数和我们上面提到的一些判断条件，返回系统现在的状态。StorageDirectory可能处于以下的某一个状态（与系统的工作状态一定的对应）： NON_EXISTENT：指定的目录不存在； NOT_FORMATTED：指定的目录存在但未被格式化； COMPLETE_UPGRADE：previous.tmp存在，current也存在 RECOVER_UPGRADE：previous.tmp存在，current不存在 COMPLETE_FINALIZE：finalized.tmp存在，current也存在 COMPLETE_ROLLBACK：removed.tmp存在，current也存在，previous不存在 RECOVER_ROLLBACK：removed.tmp存在，current不存在，previous存在 COMPLETE_CHECKPOINT：lastcheckpoint.tmp存在，current也存在 RECOVER_CHECKPOINT：lastcheckpoint.tmp存在，current不存在 NORMAL：普通工作模式。 StorageDirectory处于某些状态是通过发生对应状态改变需要的工作文件夹和正常工作的current夹来进行判断。状态改变需要的工作文件夹包括： previous：用于升级后保存以前版本的文件 previous.tmp：用于升级过程中保存以前版本的文件 removed.tmp：用于回滚过程中保存文件 finalized.tmp：用于提交过程中保存文件 lastcheckpoint.tmp：应用于从NameNode中，导入一个检查点 previous.checkpoint：应用于从NameNode中，结束导入一个检查点 有了这些状态，就可以对系统进行恢复（通过方法doRecover）。恢复的动作如下（结合上面的状态转移图）： COMPLETE_UPGRADE：mv previous.tmp -> previous RECOVER_UPGRADE：mv previous.tmp -> current COMPLETE_FINALIZE：rm finalized.tmp COMPLETE_ROLLBACK：rm removed.tmp RECOVER_ROLLBACK：mv removed.tmp -> current COMPLETE_CHECKPOINT：mv lastcheckpoint.tmp -> previous.checkpoint RECOVER_CHECKPOINT：mv lastcheckpoint.tmp -> current 我们以RECOVER_UPGRADE为例，分析一下。根据升级的过程， 1. current->previous.tmp 2. 重建current 3. previous.tmp->previous 当我们发现previous.tmp存在，current不存在，我们知道只需要将previous.tmp改为current，就能恢复到未升级时的状态。 StorageDirectory还管理着文件系统的元信息，就是我们上面提过StorageInfo信息，当然，StorageDirectory还保存每个具体用途自己的信息。这些信息，其实都存储在VERSION文件中，StorageDirectory中的read/write方法，就是用于对这个文件进行读/写。下面是某一个DataNode的VERSION文件的例子： 配置文件代码 1. #Fri Nov 14 10:27:35 CST 2008 2. namespaceID=1950997968 3. storageID=DS-697414267-127.0.0.1-50010-1226629655026 4. cTime=0 5. storageType=DATA_NODE 6. layoutVersion=-16 #Fri Nov 14 10:27:35 CST 2008 namespaceID=1950997968 storageID=DS-697414267-127.0.0.1-50010-1226629655026 cTime=0 storageType=DATA_NODE layoutVersion=-16 对StorageDirectory的排他操作需要锁，还记得我们在分析系统目录时提到的in_use.lock文件吗？它就是用来给整个系统加/解锁用的。StorageDirectory提供了对应的lock和unlock方法。 分析完StorageDirectory以后，Storage类就很简单了。基本上都是对一系列StorageDirectory的操作，同时Storage提供一些辅助方法。 DataStorage是Storage的子类，专门应用于DataNode。上面我们对DataNode的升级/回滚/提交过程，就是对DataStorage的doUpgrade/doRollback/doFinalize分析得到的。 DataStorage提供了format方法，用于创建DataNode上的Storage，同时，利用StorageDirectory，DataStorage管理存储系统的状态。 Hadoop源代码分析（一二） 分析完Storage相关的类以后，我们来看下一个大家伙，FSDataset相关的类。 上面介绍Storage时，我们并没有涉及到数据块Block的操作，所有和数据块相关的操作，都在FSDataset相关的类中进行处理。下面是类图： Block是对一个数据块的抽象，通过前面的讨论我们知道一个Block对应着两个文件，其中一个存数据，一个存校验信息，如下： blk_3148782637964391313 blk_3148782637964391313_242812.meta 上面的信息中，blockId是3148782637964391313，242812是数据块的版本号，当然，系统还会保存数据块的大小，在类中是属性numBytes。Block提供了一系列的方法来操作对象的属性。 DatanodeBlockInfo存放的是Block在文件系统上的信息。它保存了Block存放的卷（FSVolume），文件名和detach状态。这里有必要解释一下detach状态：我们前面分析过，系统在升级时会创建一个snapshot，snapshot的文件和current里的数据块文件和数据块元文件是通过硬链接，指向了相同的内容。当我们需要改变current里的文件时，如果不进行detach操作，那么，修改的内容就会影响snapshot里的文件，这时，我们需要将对应的硬链接解除掉。方法很简单，就是在临时文件夹里，复制文件，然后将临时文件改名成为current里的对应文件，这样的话，current里的文件和snapshot里的文件就detach了。这样的技术，也叫copy-on-write，是一种有效提高系统性能的方法。DatanodeBlockInfo中的detachBlock，能够对Block对应的数据文件和元数据文件进行detach操作。 介绍完类Block和DatanodeBlockInfo后，我们来看FSVolumeSet，FSVolume和FSDir。我们知道在一个DataNode上可以指定多个Storage来存储数据块，由于HDFS规定了一个目录能存放Block的数目，所以一个Storage上存在多个目录。对应的，FSDataset中用FSVolume来对应一个Storage，FSDir对应一个目录，所有的FSVolume由FSVolumeSet管理，FSDataset中通过一个FSVolumeSet对象，就可以管理它的所有存储空间。 FSDir对应着HDFS中的一个目录，目录里存放着数据块文件和它的元文件。FSDir的一个重要的操作，就是在添加一个Block时，根据需要有时会扩展目录结构，上面提过，一个Storage上存在多个目录，所有的目录，都对应着一个FSDir，目录的关系，也由FSDir保存。FSDir的getBlockInfo方法分析目录下的所有数据块文件信息，生成Block对象，存放到一个集合中。getVolumeMap方法能，则会建立Block和DatanodeBlockInfo的关系。以上两个方法，用于系统启动时搜集所有的数据块信息，便于后面快速访问。 FSVolume对应着是某一个Storage。数据块文件，detach文件和临时文件都是通过FSVolume来管理的，这个其实很自然，在同一个存储系统上移动文件，往往只需要修改文件存储信息，不需要搬数据。FSVolume有一个recoverDetachedBlocks的方法，用于恢复detach文件。和Storage的状态管理一样，detach文件有可能在复制文件时系统崩溃，需要对detach的操作进行回复。FSVolume还会启动一个线程，不断更新FSVolume所在文件系统的剩余容量。创建Block的时候，系统会根据各个FSVolume的容量，来确认Block的存放位置。 FSVolumeSet就不讨论了，它管理着所有的FSVolume。 HDFS中，对一个chunk的写会使文件处于活跃状态，FSDataset中引入了类ActiveFile。ActiveFile对象保存了一个文件，和操作这个文件的线程。注意，线程有可能有多个。ActiveFile的构造函数会自动地把当前线程加入其中。 有了上面的基础，我们可以开始分析FSDataset。FSDataset实现了接口FSDatasetInterface。FSDatasetInterface是DataNode对底层存储的抽象。 下面给出了FSDataset的关键成员变量： FSVolumeSet volumes; private HashMap ongoingCreates = new HashMap(); private HashMap volumeMap = null; 其中，volumes就是FSDataset使用的所有Storage，ongoingCreates是Block到ActiveFile的映射，也就是说，说有正在创建的Block，都会记录在ongoingCreates里。 下面我们讨论FSDataset中的方法。 public long getMetaDataLength(Block b) throws IOException; 得到一个block的元数据长度。通过block的ID，找对应的元数据文件，返回文件长度。 public MetaDataInputStream getMetaDataInputStream(Block b) throws IOException; 得到一个block的元数据输入流。通过block的ID，找对应的元数据文件，在上面打开输入流。下面对于类似的简单方法，我们就不再仔细讨论了。 public boolean metaFileExists(Block b) throws IOException; 判断block的元数据的元数据文件是否存在。简单方法。 public long getLength(Block b) throws IOException; block的长度。简单方法。 public Block getStoredBlock(long blkid) throws IOException; 通过Block的ID，找到对应的Block。简单方法。 public InputStream getBlockInputStream(Block b) throws IOException; public InputStream getBlockInputStream(Block b, long seekOffset) throws IOException; 得到Block数据的输入流。简单方法。 public BlockInputStreams getTmpInputStreams(Block b, long blkoff, long ckoff) throws IOException; 得到Block的临时输入流。注意，临时输入流是指对应的文件处于tmp目录中。新创建块时，块数据应该写在tmp目录中，直到写操作成功，文件才会被移动到current目录中，如果失败，就不会影响current目录了。简单方法。 public BlockWriteStreams writeToBlock(Block b, boolean isRecovery) throws IOException; 得到一个block的输出流。BlockWriteStreams既包含了数据输出流，也包含了元数据（校验文件）输出流，这是一个相当复杂的方法。 参数isRecovery说明这次写是不是对以前失败的写的一次恢复操作。我们先看正常的写操作流程：首先，如果输入的block是个正常的数据块，或当前的block已经有线程在写，writeToBlock会抛出一个异常。否则，将创建相应的临时数据文件和临时元数据文件，并把相关信息，创建一个ActiveFile对象，记录到ongoingCreates中，并创建返回的BlockWriteStreams。前面我们已经提过，建立新的ActiveFile时，当前线程会自动保存在ActiveFile的threads中。 我们以blk_3148782637964391313为例，当DataNode需要为Block ID为3148782637964391313创建写流时，DataNode创建文件tmp/blk_3148782637964391313做为临时数据文件，对应的meta文件是tmp/blk_3148782637964391313_XXXXXX.meta。其中XXXXXX是版本号。 isRecovery为true时，表明我们需要从某一次不成功的写中恢复，流程相对于正常流程复杂。如果不成功的写是由于提交（参考finalizeBlock方法）后的确认信息没有收到，先创建一个detached文件（备份）。接着，writeToBlock检查是否有还有对文件写的线程，如果有，则通过线程的interrupt方法，强制结束线程。这就是说，如果有线程还在写对应的文件块，该线程将被终止。同时，从ongoingCreates中移除对应的信息。接下来将根据临时文件是否存在，创建/复用临时数据文件和临时数据元文件。后续操作就和正常流程一样，根据相关信息，创建一个ActiveFile对象，记录到ongoingCreates中…… 由于这块涉及了一些HDFS写文件时的策略，以后我们还会继续讨论这个话题。 public void updateBlock(Block oldblock, Block newblock) throws IOException; 更新一个block。这也是一个相当复杂的方法。 updateBlock的最外层是一个死循环，循环的结束条件，是没有任何和这个数据块相关的写线程。每次循环，updateBlock都会去调用一个叫tryUpdateBlock的内部方法。tryUpdateBlock发现已经没有线程在写这个块，就会跟新和这个数据块相关的信息，包括元文件和内存中的映射表volumeMap。如果tryUpdateBlock发现还有活跃的线程和该块关联，那么，updateBlock会试图结束该线程，并等在join上等待。 public void finalizeBlock(Block b) throws IOException; 提交（或叫：结束finalize）通过writeToBlock打开的block，这意味着写过程没有出错，可以正式把Block从tmp文件夹放到current文件夹。 在FSDataset中，finalizeBlock将从ongoingCreates中删除对应的block，同时将block对应的DatanodeBlockInfo，放入volumeMap中。我们还是以blk_3148782637964391313为例，当DataNode提交Block ID为3148782637964391313数据块文件时，DataNode将把tmp/blk_3148782637964391313移到current下某一个目录，以subdir12为例，这是tmp/blk_3148782637964391313将会挪到current/subdir12/blk_3148782637964391313。对应的meta文件也在目录current/subdir12下。 public void unfinalizeBlock(Block b) throws IOException; 取消通过writeToBlock打开的block，与finalizeBlock方法作用相反。简单方法。 public boolean isValidBlock(Block b); 该Block是否有效。简单方法。 public void invalidate(Block invalidBlks[]) throws IOException; 使block变为无效。简单方法。 public void validateBlockMetadata(Block b) throws IOException; 检查block的有效性。简单方法。 Hadoop源代码分析（一三） 通过上面的一系列介绍，我们知道了DataNode工作时的文件结构和文件结构在内存中的对应对象。下面我们可以来开始分析DataNode上的动态行为。首先我们来分析DataXceiverServer和DataXceiver。DataNode上数据块的接受/发送并没有采用我们前面介绍的RPC机制，原因很简单，RPC是一个命令式的接口，而DataNode处理数据部分，往往是一种流式机制。DataXceiverServer和DataXceiver就是这个机制的实现。其中，DataXceiver还依赖于两个辅助类：BlockSender和BlockReceiver。下面是类图： （为了简单起见，BlockSender和BlockReceiver的成员变量没有进入UML模型中） DataXceiverServer很简单，它打开一个端口，然后每接收到一个连接，就创建一个DataXceiver，服务于该连接，并记录该连接的socket，对应的实现在DataXceiverServer的run方法里。当系统关闭时，DataXceiverServer将关闭监听的socket和所有DataXceiver的socket，这样就导致了DataXceiver出错并结束线程。 DataXceiver才是真正干活的地方，目前，DataXceiver支持的操作总共有六条，分别是： OP_WRITE_BLOCK (80)：写数据块 OP_READ_BLOCK (81)：读数据块 OP_READ_METADATA (82)：读数据块元文件 OP_REPLACE_BLOCK (83)：替换一个数据块 OP_COPY_BLOCK (84)：拷贝一个数据块 OP_BLOCK_CHECKSUM (85)：读数据块检验码 DataXceiver首先读取客户端的版本号并检验，然后再读取一个字节的操作码，并转入相关的子程序进行处理。我们先看一下读数据块的过程吧。 首先看输入，下图是读数据块时，客户端发送过来的信息： 包括了要读取的Block的ID，时间戳，开始偏移和读取的长度，最后是客户端的名字（貌似只是在写日志的时候用到了）。根据上面的信息，我们可以创建一个BlockSender，如果BlockSender没有出错，返回客户端一个正确指示后，否则，返回错误码。成功创建BlockSender以后，就可以开始通过BlockSender.sendBlock发送数据。 下面我们就来分析BlockSender。BlockSender的构造函数看似很复杂，其实就是根据需求（特别是在处理checksum上，因为checksum是基于块的），打开相应的数据流。close()用于释放各种资源，如已经打开的数据流。sendBlock用于发送数据，数据发送包括应答头和后续的数据包。应答头如下（包含DataXceiver中发送的成功标识）： 然后后面的数据就组织成数据包来发送，包结构如下： 各个字段含义： packetLen：包长度，包括包头 offset：偏移量 seqno：包序列号 tail：是否是最后一个包 len：数据长度 checksum：检验数据 data：数据块数据 需要注意的，在写数据前，BlockSender会校验数据，保证数据包中的checksum和数据的一致性。同时，如果数据出错，将会有ChecksumException抛出。 数据传输结束的标志，是一个packetLen长度为0的包。客户端可以返回一个两字节的应答OP_STATUS_CHECKSUM_OK(5) Hadoop源代码分析（一四） 继续DataXceiver分析，下一块硬骨头是写数据块。HDFS的写数据操作，比读数据复杂N多倍。读数据的时候，只需要在多个数据块文件的选一个读，就可以了；但是，写数据需要同时写到多个数据块文件上，这就比较复杂了。HDFS实现了了Google写文件时的机制，如下图： 数据流从客户端开始，流经一系列的节点，到达最后一个DataNode。图中的所有DataNode只需要写一次硬盘，DataNode1和DataNode2会将从socket上接受到的数据，直接写到到下个节点的socket上。 我们来看一下写数据块的请求。 首先是客户端的版本号和一个字节的操作码，接下来是我们熟悉的blockId和generationStamp。参数pipelineSize是整个数据流链的长度，以上面为例，pipelineSize=3。isRecovery指示这次写是否是一次恢复操作，还记得我们在讨论FSDataset.writeToBlock时的那个参数吗？isRecovery来自客户端。client是客户端的名字，就是发起请求的节点名，需要特别注意的是，如果是从NameNode来的复制请求，client为空。hasSrcDataNode是一个标志位，如果被设置，表明源节点是个DataNode，接下来读取的数据就是DataNode的信息。numTargets是目标节点的数目，包括当前节点，以上面的图为例，DataNode1上这个参数值为3，到了DataNode3，就只有1了。targets包含了目标节点的相关信息，根据这些信息，就可以创建到它们上面的socket连接。targets后跟着的是校验头。 writeBlock最开始是处理上面提到的消息包，然后创建一个BlockReceiver。接下来就是创建一堆用于读写的流，如下图（图中除了in外，都是在writeBlock中创建，这个图还不涉及在BlockReceiver对本地文件读写的流）： 在进行实际的数据写之前，上面的这些流会被建立起来（也就是说，DataNode1到DataNode3都可写以后，才开始处理写数据）。如果其中某一个点出错了，那么，出错的节点名会通过mirrorIn发送回来，一直沿着这条链，传播到客户端。 如果一切正常，那么，BlockReceiver.receiveBlock就开始干活了。 BlockReceiver的构造函数会创建写数据块和校验数据的输出流。剩下的就交给receiveBlock这个大家伙了。首先receiveBlock会再启动一个线程（一般来说，BlockReceiver就跑在它自己的线程上），用于处理应答（内部类PacketResponder定义了该线程），然后就不断调用receivePacket读数据。 数据是以分块的形式传送，格式和读Block的时候是一样的。如下图（很奇怪，为啥不抽象为类）： 注意：如果当前DataNode处于数据流的中间，该数据包会发送到下一个节点。 接下来的处理，就是处理数据和校验，并分别写到数据块文件和数据块元数据文件。如果出错，抛出的异常会导致receiveBlock关闭相关的输出流，并终止传输。注意，数据校验出错还会上报到NameNode上。 PacketResponder用于处理应答。也就是上面讲的mirrorIn和replyOut。PacketResponder里有一个队列ackQueue，receivePacket每收到一个包，都会往队列里添加一项。PacketResponder的run方法，根据工作的DataNode所处的位置，行为不一样。 最后一个DataNode由于没有后续节点，PacketResponder的ackQueue每收到一项，表明对应的数据块已经处理完毕，那么就可以发送成功应答。如果该应答是最后一个包的，PacketResponder会关闭相关的输出流，并提交（前面讲FSDataset时后我们讨论过的finalizeBlock方法）。 如果DataNode有后续节点，那么，它必须等到后续节点的成功应答，才可以发送应答到它前面的节点。 PacketResponder的run方法还引入了心跳机制，用于检测连接是否还存在。 注意：所有改变DataNode的操作，需要把信息更新到NameNode上，这是通过DataNode.notifyNamenodeReceivedBlock方法，然后通过DataNode统一发送到NameNode上。 Hadoop源代码分析（一五） DataXceiver支持的的6条操作，我们已经分析完最重要的两条。剩下的分别是： OP_READ_METADATA (82)：读数据块元文件 OP_REPLACE_BLOCK (83)：替换一个数据块 OP_COPY_BLOCK (84)：拷贝一个数据块 OP_BLOCK_CHECKSUM (85)：读数据块检验码 我们逐个讨论。 读数据块元文件的请求如图（操作码82）： 应答很简单，应答码（如OP_STATUS_SUCCESS），文件长度（int），数据。 拷贝数据块和替换数据块是一对相对应操作。 替换数据块的请求如图（操作码83）。这个比起上面的读数据块元文件请求，有点复杂。替换一个数据块是系统平衡操作的一部分，用于接收一个数据块。它和普通的数据块写的差别是，它只发生在两个节点上，一个写，一个读，而不需要建立数据链。我们可以比较一下它们在创建BlockReceiver对象时的差别： Java代码 1. blockReceiver = new BlockReceiver(block, proxyReply, 2. proxySock.getRemoteSocketAddress().toString(), 3. proxySock.getLocalSocketAddress().toString(), 4. false, "", null, datanode); //OP_REPLACE_BLOCK 5. blockReceiver = new BlockReceiver(block, in, 6. s.getRemoteSocketAddress().toString(), 7. s.getLocalSocketAddress().toString(), 8. isRecovery, client, srcDataNode, datanode); //OP_WRITE_BLOCK blockReceiver = new BlockReceiver(block, proxyReply, proxySock.getRemoteSocketAddress().toString(), proxySock.getLocalSocketAddress().toString(), false, "", null, datanode); //OP_REPLACE_BLOCK blockReceiver = new BlockReceiver(block, in, s.getRemoteSocketAddress().toString(), s.getLocalSocketAddress().toString(), isRecovery, client, srcDataNode, datanode); //OP_WRITE_BLOCK 首先，proxyReply和in不一样，这是因为发起请求的节点和提供数据的节点并不是同一个。写数据块发起请求方也提供数据，替换数据块请求方不提供数据，而是提供了一个数据源（proxySource参数），由replaceBlock发起一个拷贝数据块的请求，建立数据源。对于拷贝数据块操作，isRecovery=false，client=””， srcDataNode=null。注意，我们在分析BlockReceiver是，讨论过client=””的情况，就是应用于这种场景。 在创建BlockReceiver对象前，需要利用下面介绍的拷贝数据块的请求建立到数据源的socket连接并发送拷贝数据块请求。然后通过BlockReceiver.receiveBlock接收数据。任务成功后将结果通知notifyNamenodeReceivedBlock。 拷贝数据块的请求如图（操作码84）。和读数据块操作请求类似，但是读取的是整个数据块，所以少了很多参数。 读数据块检验码的请求如图（操作码85）。它能够读取某个数据块的检验和的MD5结果，实现的方法很简单。 Hadoop源代码分析（一六） 通过上面的讨论，DataNode上的读/写流程已经基本清楚了。我们来看下一个非主流流程， DataBlockScanner用于定时对数据块文件进行校验。类图如下： DataBlockScanner拥有它单独的线程，能定时地从目前DataNode管理的数据块文件进行校验。其实最重要的方法就是verifyBlock，我们来看这个方法最关键的地方： Java代码 1. blockSender = new BlockSender(block, 0, -1, false, false, true, datanode); 2. DataOutputStream out = new DataOutputStream(new IOUtils.NullOutputStream()); 3. blockSender.sendBlock(out, null, throttler); blockSender = new BlockSender(block, 0, -1, false, false, true, datanode); DataOutputStream out = new DataOutputStream(new IOUtils.NullOutputStream()); blockSender.sendBlock(out, null, throttler); 校验利用了BlockSender，因为我们知道BlockSender中，发送数据的同时，会对数据进行校验。verifyBlock只需要读一个Block到一个空输出设备（NullOutputStream），如果有异常，那么校验失败，如果正常，校验成功。 DataBlockScanner其他的辅助方法用于对DataBlockScanner管理的数据块文件信息进行增加/删除，排序操作。同时，校验的信息还会保持在Storage上，保存在dncp_block_verification.log.curr和dncp_block_verification.log.prev中。 Hadoop源代码分析（一七） 周围的障碍扫清以后，我们可以开始分析类DataNode。类图如下： public class DataNode extends Configured implements InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable 上面给出了DataNode的继承关系，我们发现，DataNode实现了两个通信接口，其中ClientDatanodeProtocol是用于和Client交互的，InterDatanodeProtocol，就是我们前面提到的DataNode间的通信接口。ipcServer（类图的左下方）是DataNode的一个成员变量，它启动了一个IPC服务，这样，DataNode就能提供ClientDatanodeProtocol和InterDatanodeProtocol的能力了。 我们从main函数开始吧。这个函数很简单，调用了createDataNode的方法，然后就等着DataNode的线程结束。createDataNode首先调用instantiateDataNode初始化DataNode，然后执行runDatanodeDaemon。runDatanodeDaemon会向NameNode注册，如果成功，才启动DataNode线程，DataNode就开始干活了。 初始化DataNode的方法instantiateDataNode会读取DataNode需要的配置文件，同时读取配置的storage目录（可能有多个，看storage的讨论部分），然后把这两参数送到makeInstance中，makeInstance会先检查目录（存在，是目录，可读，可写），然后调用： new DataNode(conf, dirs); 接下来控制流就到了构造函数上。构造函数调用startDataNode，完成和DataNode相关的初始化工作（注意，DataNode工作线程不在这个函数里启动）。首先是初始化一堆的配置参数，什么NameNode地址，socket参数等等。然后，向NameNode请求配置信息（DatanodeProtocol.versionRequest），并检查返回的NamespaceInfo和本地的版本是否一致。 正常情况的下一步是检查文件系统的状态并做必要的恢复，初始化FSDataset（到这个时候，上面图中storage和data成员变量已经初始化）。 然后，找一个端口并创建DataXceiverServer（run方法里启动），创建DataBlockScanner（根据需要在offerService中启动，只启动一次），创建DataNode上的HttpServer，启动ipcServer。这样就结束了DataNode相关的初始化工作。 在启动DataNode工作线程前，DataNode需要向NameNode注册。注册信息在初始化的时候已经构造完毕，包括DataXceiverServer端口，ipcServer端口，文件布局版本号等重要信息。注册成功后就可以启动DataNode线程。 DataNode的run方法，循环里有两种选择，升级（暂时不讨论）/正常工作。我们来看正常工作的offerService方法。offerService也是个循环，在循环里，offerService会定时向NameNode发送心跳，报告系统中Block状态的变化，报告DataNode现在管理的Block状态。发送心跳和Block状态报告时，NameNode会返回一些命令，DataNode将执行这些命令。 心跳的处理比较简单，以heartBeatInterval间隔发送。 Block状态变化报告，会利用保存在receivedBlockList和delHints两个列表中的信息。receivedBlockList表明在这个DataNode成功创建的新的数据块，而delHints，是可以删除该数据块的节点。如在DataXceiver的replaceBlock中，有调用： datanode.notifyNamenodeReceivedBlock(block, sourceID) 这表明，DataNode已经从sourceID上接收了一个Block，sourceID上对应的Block可以删除了（这个场景出现在当系统需要做负载均衡时，Block在DataNode之间拷贝）。 Block状态变化报告通过NameNode.blockReceived来报告。 Block状态报告也比较简单，以blockReportInterval间隔发送。 心跳和Block状态报告可以返回命令，这也是NameNode先DataNode发起请求的唯一方法。我们来看一下都有那些命令： DNA_TRANSFER：拷贝数据块到其他DataNode DNA_INVALIDATE：删除数据块（简单方法） DNA_SHUTDOWN：关闭DataNode（简单方法） DNA_REGISTER：DataNode重新注册（简单方法） DNA_FINALIZE：提交升级（简单方法） DNA_RECOVERBLOCK：恢复数据块 拷贝数据块到其他DataNode由transferBlocks方法执行。注意，返回的命令可以包含多个数据块，每一个数据块可以包含多个目标地址。transferBlocks方法将为每一个Block启动一个DataTransfer线程，用于传输数据。 DataTransfer是一个DataNode的内部类，它利用我们前面介绍的OP_WRITE_BLOCK写数据块操作，发送数据到多个目标上面。 恢复数据块和NameNode的租约（lease）恢复有关，我们后面再讨论。 Hadoop源代码分析（一八） DataNode的介绍基本告一段落。我们开始来分析NameNode。相比于DataNode，NameNode比较复杂。系统中只有一个NameNode，作为系统文件目录的管理者和“inode表”（熟悉UNIX的同学们应该了解inode）。为了高可用性，系统中还存在着从NameNode。 先前我们分析DataNode的时候，关注的是数据块。NameNode作为HDFS中文件目录和文件分配的管理者，它保存的最重要信息，就是下面两个映射： 文件名à数据块 数据块àDataNode列表 其中，文件名à数据块保存在磁盘上（持久化）；但NameNode上不保存数据块àDataNode列表，该列表是通过DataNode上报建立起来的。 下图包含了NameNode和DataNode往外暴露的接口，其中，DataNode实现了InterDatanodeProtocol和ClientDatanodeProtocol，剩下的，由NameNode实现。 ClientProtocol提供给客户端，用于访问NameNode。它包含了文件角度上的HDFS功能。和GFS一样，HDFS不提供POSIX形式的接口，而是使用了一个私有接口。一般来说，程序员通过org.apache.hadoop.fs.FileSystem来和HDFS打交道，不需要直接使用该接口。 DatanodeProtocol：用于DataNode向NameNode通信，我们已经在DataNode的分析过程中，了解部分接口，包括：register，用于DataNode注册；sendHeartbeat/blockReport/blockReceived，用于DataNode的offerService方法中；errorReport我们没有讨论，它用于向NameNode报告一个错误的Block，用于BlockReceiver和DataBlockScanner；nextGenerationStamp和commitBlockSynchronization用于lease管理，我们在后面讨论到lease时，会统一说明。 NamenodeProtocol用于从NameNode到NameNode的通信。 下图补充了接口里使用的数据的关系。 · · 大小: 122 KB · · 大小: 115.5 KB Hadoop源代码分析（一九） 我们先分析INode*.java，类INode*抽象了文件层次结构。如果我们对文件系统进行面向对象的抽象，一定会得到和下面一样类似的结构图（类INode*）： INode是一个抽象类，它的两个字类，分别对应着目录（INodeDirectory）和文件（INodeFile）。INodeDirectoryWithQuota，如它的名字隐含的，是带了容量限制的目录。INodeFileUnderConstruction，抽象了正在构造的文件，当我们需要在HDFS中创建文件的时候，由于创建过程比较长，目录系统会维护对应的信息。 INode中的成员变量有：name，目录/文件名；modificationTime和accessTime是最后的修改时间和访问时间；parent指向了父目录；permission是访问权限。HDFS采用了和UNIX/Linux类似的访问控制机制。系统维护了一个类似于UNIX系统的组表（group）和用户表（user），并给每一个组和用户一个ID，permission在INode中是long型，它同时包含了组和用户信息。 INode中存在大量的get和set方法，当然是对上面提到的属性的操作。导出属性，比较重要的有：collectSubtreeBlocksAndClear，用于收集这个INode所有后继中的Block；computeContentSummary用于递归计算INode包含的一些相关信息，如文件数，目录数，占用磁盘空间。 INodeDirectory是HDFS管理的目录的抽象，它最重要的成员变量是： private List children; 就是这个目录下的所有目录/文件集合。INodeDirectory也是有大量的get和set方法，都很简单。INodeDirectoryWithQuota进一步加强了INodeDirectory，限制了INodeDirectory可以使用的空间（包括NameSpace和磁盘空间）。 INodeFile是HDFS中的文件，最重要的成员变量是： protected BlockInfo blocks[] = null; 这是这个文件对应的Block列表，BlockInfo增强了Block类。 INodeFileUnderConstruction保存了正在构造的文件的一些信息，包括clientName，这是目前拥有租约的节点名（创建文件时，只有一个节点拥有租约，其他节点配合这个节点工作）。clientMachine是构造该文件的客户端名称，如果构造请求由DataNode发起，clientNode会保持相应的信息，targets保存了配合构造文件的所有节点。 上面描述了INode*类的关系。下面我们顺便考察一下一些NameNode上的数据类。 BlocksMap保存了Block和它在NameNode上一些相关的信息。其核心是一个map：Map。BlockInfo扩展了Block，保存了该Block归属的INodeFile和DatanodeDescriptor，同时还包括了它的前继和后继Block。有了BlocksMap，就可以通过Block找对应的文件和这个Block存放的DataNode的相关信息。 接下来我们来分析类Datanode*。DatanodeInfo和DatanodeID都定义在包org.apache.hadoop.hdfs.protocol。DatanodeDescriptor是DatanodeInfo的子类，包含了NameNode需要的附加信息。DatanodeID只包含了一些配置信息，DatanodeInfo增加了一些动态信息，DatanodeDescriptor更进一步，包含了DataNode上一些Block的动态信息。DatanodeDescriptor包含了内部类BlockTargetPair，它保存Block和对应DatanodeDescriptor的关联，BlockQueue是BlockTargetPair队列。 DatanodeDescriptor包含了两个BlockQueue，分别记录了该DataNode上正在复制（replicateBlocks）和Lease恢复（recoverBlocks）的Block。同时还有一个Block集合，保存的是该DataNode上已经失效的Block。DatanodeDescriptor提供一系列方法，用于操作上面保存的队列和集合。也提供get*Command方法，用于生成发送到DataNode的命令。 当NameNode收到DataNode对现在管理的Block状态的汇报是，会调用reportDiff，找出和现在NameNode上的信息差别，以供后续处理用。 readFieldsFromFSEditLog方法用于从日志中恢复DatanodeDescriptor。 Hadoop源代码分析（二零） 前面我们提过关系：文件名à数据块持久化在磁盘上，所有对目录树的更新和文件名à数据块关系的修改，都必须能够持久化。为了保证每次修改不需要从新保存整个结构，HDFS使用操作日志，保存更新。 现在我们可以得到NameNode需要存储在Disk上的信息了，包括： [hadoop@localhost dfs]$ ls -R name name: current image in_use.lock name/current: edits fsimage fstime VERSION name/image: fsimage in_use.lock的功能和DataNode的一致。fsimage保存的是文件系统的目录树，edits则是文件树上的操作日志，fstime是上一次新打开一个操作日志的时间（long型）。 image/fsimage是一个保护文件，防止0.13以前的版本启动（0.13以前版本将fsimage存放在name/image目录下，如果用0.13版本启动，显然在读fsimage会出错J）。 我们可以开始讨论FSImage了，类FSImage如下图： 分析FSImage，不免要跟DataStorage去做比较（上图也保留了类DataStorage）。前面我们已经分析过DataStorage的状态变化，包括升级/回滚/提交，FSImage也有类似的升级/回滚/提交动作，而且这部分的行为和DataStorage是比较一致，如下状态转移图。图中update方法和DataStorage的差别比较大，是因为处理数据库和处理文件系统名字空间不一样，其他的地方都比较一致。FSImage也能够管理多个Storage，而且还能够区分Storage为IMAGE(目录结构)/EDITS（日志）/IMAGE_AND_EDITS（前面两种的组合）。 我们可以看到，FSImage和DataStorage都有recoverTransitionRead方法。FSImage的recoverTransitionRead方法主要步骤是检查系统一致性（analyzeStorage）并尝试恢复，初始化新的storage，然后根据启动NameNode的参数，做升级/回滚等操作。 FSImage需要支持参数-importCheckpoint，该参数用于在某一个checkpoint目录里加载HDFS的目录信息，并更新到当前系统，该参数的主要功能在方法doImportCheckpoint中。该方法很简单，通过读取配置的checkpoint目录来加载fsimage文件和日志文件，然后利用saveFSImage（下面讨论）保存到当前的工作目录，完成导入。 loadFSImage(File curFile)用于在fsimage中读入NameNode持久化的信息，是FSImage中最重要的方法之一，该文件的结构如下： 最开始是版本号（注意，各版本文件布局不一样，文中分析的样本是0.17的），然后是命名空间的ID号，文件个数和最高文件版本号（就是说，下一次产生文件版本号的初始值）。接下来就是文件的信息啦，首先是文件名，然后是该文件的副本数，接下来是修改时间/访问时间，数据块大小，数据块数目。数据块数目如果大于0，表明这是个文件，那么接下来就是numBlocks个数据块（浅蓝），如果数据块数目等于0，那该条目是目录，接下来是应用于该目录的quota。最后是访问控制的一些信息。文件信息一共有numFiles个，接下来是处于构造状态的文件的信息。（有些版本可能还会保留DataNode的信息，但0.17已经不保存这样的信息啦）。loadFSImage(File curFile)的对应方法是saveFSImage(File newFile)，FSImage中还有一系列的方法（大概7，8个）用于配合这两个方法工作，我们就不再深入讨论了。 loadFSEdits(StorageDirectory sd)用于加载日志文件，并把日志文件记录的内容应用到NameNode，loadFSEdits只是简单地调用FSEditLog中对应的方法。 loadFSImage()和saveFSImage()是另外一对重要的方法。 loadFSImage()会在所有的Storage中，读取最新的NameNode持久化信息，并应用相应的日志，当loadFSImage()调用返回以后，内存中的目录树就是最新的。loadFSImage()会返回一个标记，如果Storage中有任何和内存中最终目录树中不一致的Image（最常见的情况是日志文件不为空，那么，内存中的Image应该是Storage的Image加上日志，当然还有其它情况），那么，该标记为true。 saveFSImage()的功能正好相反，它将内存中的目录树持久化，很自然，目录树持久化后就可以把日志清空。saveFSImage()会创建edits.new，并把当前内存中的目录树持久化到fsimage.ckpt（fsimage现在还存在），然后重新打开日志文件edits和edits.new，这会导致日志文件edits和edits.new被清空。最后，saveFSImage()调用rollFSImage()方法。 rollFSImage()上来就把所有的edits.new都改为edits（经过了方法saveFSImage，它们都已经为空），然后再把fsimage.ckpt改为fsimage。如下图： 为了防止误调用rollFSImage()，系统引入了状态CheckpointStates.UPLOAD_DONE。 有了上面的状态转移图，我们就很好理解方法recoverInterruptedCheckpoint了。 图中存在另一条路径，应用于GetImageServlet中。GetImageServlet是和从NameNode进行文件通信的接口，这个场景留到我们分析从NameNode时再进行分析。 最后我们分析一下和检查点相关的一个类，rollFSImage()会返回这个类的一个实例。CheckpointSignature用于标识一个日志的检查点，它是StorageInfo的子类，同时实现了WritableComparable接口，出了StorageInfo的信息，它还包括了两个属性：editsTime和checkpointTime。editsTime是日志的最后修改时间，checkpointTime是日志建立时间。在和从NameNode节点的通信中，需要用CheckpointSignature，来保证从NameNode获得的日志是最新的。 Hadoop源代码分析（二一） 不好意思，突然间需要忙项目的其他事情了，更新有点慢下来，争取月底搞定HDFS吧。 我们来分析FSEditLog.java，该类提供了NameNode操作日志和日志文件的相关方法，相关类图如下： 首先是FSEditLog依赖的输入/输出流。输入流基本上没有新添加功能；输出流在打开的时候，会写入日志的版本号（最前面的4字节），同时，每次将内存刷到硬盘时，会为日志尾部写入一个特殊的标识（OP_INVALID）。 FSEditLog有打开/关闭的方法，它们都是很简单的方法，就是关闭的时候，要等待所有正在写日志的操作都完成写以后，才能关闭。processIOError用于处理IO出错，一般这会导致对于的Storage的日志文件被关闭（还记得loadFSImage要找出最后写的日志文件吧，这也是提高系统可靠性的一个方法），如果系统再也找不到可用的日志文件，NameNode将会退出。 loadFSEdits是个大家伙，它读取日志文件，并把日志应用到内存中的目录结构中。这家伙大是因为它需要处理所有类型的日志记录，其实就一大case语句。logEdit的作用和loadFSEdits相反，它向日志文件中写入日志记录。我们来分析一下什么操作需要写log，还有就是需要log那些参数： logOpenFile（OP_ADD）：申请lease path(路径)/replication（副本数，文本形式）/modificationTime（修改时间，文本形式）/accessTime（访问时间，文本形式）/preferredBlockSize（块大小，文本形式）/BlockInfo[]（增强的数据块信息，数组）/permissionStatus（访问控制信息）/clientName（客户名）/clientMachine（客户机器名） logCloseFile（OP_CLOSE）：归还lease path/replication/modificationTime/accessTime/preferredBlockSize/BlockInfo[]/permissionStatus logMkDir（OP_MKDIR）：创建目录 path/modificationTime/accessTime/permissionStatus logRename（OP_RENAME）：改文件名 src（原文件名）/dst（新文件名）/timestamp（时间戳） logSetReplication（OP_SET_REPLICATION）：更改副本数 src/replication logSetQuota（OP_SET_QUOTA）：设置空间额度 path/nsQuota（文件空间额度）/dsQuota（磁盘空间额度） logSetPermissions（OP_SET_PERMISSIONS）：设置文件权限位 src/permissionStatus logSetOwner（OP_SET_OWNER）：设置文件组和主 src/username（所有者）/groupname（所在组） logDelete（OP_DELETE）：删除文件 src/timestamp logGenerationStamp（OP_SET_GENSTAMP）：文件版本序列号 genstamp（序列号） logTimes（OP_TIMES）：更改文件更新/访问时间 src/modificationTime/accessTime 通过上面的分析，我们应该清楚日志文件里记录了那些信息。 rollEditLog()我们在前面已经提到过（配合saveFSImage和rollFSImage），它用于关闭edits，打开日志到edits.new。purgeEditLog()的作用正好相反，它删除老的edits文件，然后把edits.new改名为edits。这也是Hadoop在做更新修改时经常采用的策略。 Hadoop源代码分析（二二） 我们开始对租约Lease进行分析，下面是类图。Lease可以认为是一个文件写锁，当客户端需要写文件的时候，它需要申请一个Lease，NameNode负责记录那个文件上有Lease，Lease的客户是谁，超时时间（分布式处理的一种常用技术）等，所有这些工作由下面3个类完成。至于租约过期NameNode需要采取什么动作，并不是这部分code要完成的功能。 LeaseManager（左）管理着系统中的所有Lease（右），同时，LeaseManager有一个线程Monitor，用于检查是否有Lease到期。 一个租约由一个holder（客户端名），lastUpdate（上次更新时间）和paths（该客户端操作的文件集合）构成。了解了这些属性，相关的方法就很好理解了。LeaseManager的方法也就很好理解，就是对Lease进行操作。注意，LeaseManager的addLease并没有检查文件上是否已经有Lease，这个是由LeaseManager的调用者来保证的，这使LeaseManager跟简单。内部类Monitor通过对Lease的最后跟新时间来检测Lease是否过期，如果过期，简单调用FSNamesystem的internalReleaseLease方法。 这部分的代码比我想象的简单，主要是大部分的一致性逻辑都存在于LeaseManager的使用者。在开始分析FSNamesystem.java这个4.5k多行的庞然大物之前，我们继续来扫除外围的障碍。下面是关于访问控制的一些类： Hadoop文件保护采用的UNIX的机制，文件用户分文件属主、文件组和其他用户，权限读，写和执行（FsAction中抽象了所有组合）。 我们先分析包org.apache.hadoop.fs.permission的几个类吧。FsAction抽象了操作权限，FsPermission记录了某文件/路径的允许情况，分文件属主、文件组和其他用户，同时提供了一系列的转换方法，applyUMask用于去掉某些权限，如某些操作需要去掉文件的写权限，那么可以通过该方法，生成对应的去掉写权限的FsPermission对象。PermissionStatus用于描述一个文件的文件属主、文件组和它的FsPermission。 INode在保存PermissionStatus时，用了不同的方法，它用一个long变量，和SerialNumberManager配合，保存了PermissionStatus的所有信息。 SerialNumberManager保存了文件主和文件主号，用户组和用户组号的对应关系。注意，在持久化信息FSImage中，不保存文件主号和用户组号，它们只是SerialNumberManager分配的，只保存在内存的信息。通过SerialNumberManager得到某文件主的文件主号时，如果找不到文件主号，会往对应关系中添加一条记录。 INode的long变量作为一个位串，分组保存了FsPermission（MODE），文件主号（USER）和用户组号（GROUP）。 PermissionChecker用于权限检查。 Hadoop源代码分析（二三） 下面我们来分析FSDirectory。其实分析FSDirectory最好的地方，应该是介绍完INode*以后，FSDirectory在INode*的基础上，保存了HDFS的文件目录状态。系统加载FSImage时，FSImage会在FSDirectory对象上重建文件目录状态，HDFS文件目录状态的变化，也由FSDirectory写日志，同时，它保存了文件名à数据块的映射关系。 FSDirectory只有很少的成员变量，如下： final FSNamesystem namesystem; final INodeDirectoryWithQuota rootDir; FSImage fsImage; boolean ready = false; 其中，namesystem，fsImage是指向FSNamesystem对象和FSImage对象的引用，rootDir是文件系统的根，ready初值为false，当系统成功加载FSImage以后，ready会变成true，FSDirectory的使用者就可以调用其它FSDirectory功能了。 FSDirectory中剩下的，就是一堆的方法（我们不讨论和MBean相关的类，方法和过程）。 loadFSImage用于加载目录树结构，它会去调用FSImage的方法，完成持久化信息的导入以后，它会把成员变量ready置为true。系统调用loadFSImage是在FSNamesystem.java的initialize方法，那是系统初始化重要的一步。 addFile用于创建文件或追加数据时创建INodeFileUnderConstruction，下图是它的Call Hierachy图： addFile首先会试图在系统中创建到文件的路径，如果文件为/home/hadoop/Hadoop.tar，addFile会调用mkdirs（创建路径为/home/hadoop，这也会涉及到一系列方法），保证文件路径存在，然后创建INodeFileUnderConstruction节点，并把该节点加到目录树中（通过addNode，也是需要调用一系列方法），如果成功，就写操作日志（logOpenFile）。 unprotectedAddFile也用于在系统中创建一个目录或文件（非UnderConstruction），如果是文件，还会建立对应的block。FSDirectory中还有好几个unprotected*方法，它们不检查成员变量ready，不写日志，它们大量用于loadFSEdits中（这个时候ready当然是false，而且因为正在恢复日志，也不需要写日志）。 addToParent添加一个INode到目录树中，并返回它的上一级目录，它的实现和unprotectedAddFile是类似的。 persistBlocks比较有意思，用于往日志里记录某inode的block信息，其实并没有一个对应于persistBlocks的写日志方法，它用的是logOpenFile。这个大家可以去检查一下logOpenFile记录的信息。closeFile对应了logCloseFile。 addBlock和removeBlock对应，用于添加/删除数据块信息，同时它们还需要更新FSNamesystem.java中对应的信息。 unprotectedRenameTo和renameTo实现了UNIX的mv命令，主要的功能都在unprotectedRenameTo中完成，复杂的地方在于对各种各样情况的讨论。 setReplication和unprotectedSetReplication用于更新数据块的副本数，很简单的方法，注意，改变产生的对数据块的删除/复制是在FSNamesystem.java中实现。 setPermission，unprotectedSetPermission，setOwner和unprotectedSetOwner都是简单的方法。 Delete和unprotectedDelete又是一对方法，删除如果需要删除数据块，将通过FSNamesystem的removePathAndBlocks进行。 ……(后续的方法和前面介绍的，都比较类似，都是一些过程性的东西，就不再讨论了) Hadoop源代码分析（二四） 下面轮到FSNamesystem出场了。FSNamesystem.java一共有4573行，而整个namenode目录下所有的Java程序总共也只有16876行，把FSNamesystem搞定了，NameNode也就基本搞定。 FSNamesystem是NameNode实际记录信息的地方，保存在FSNamesystem中的数据有： l           文件名à数据块列表（存放在FSImage和日志中） l           合法的数据块列表（上面关系的逆关系） l           数据块àDataNode（只保存在内存中，根据DataNode发过来的信息动态建立） l           DataNode上保存的数据块（上面关系的逆关系） l           最近发送过心跳信息的DataNode（LRU） 我们先来分析FSNamesystem的成员变量。 private boolean isPermissionEnabled; 是否打开权限检查，可以通过配置项dfs.permissions来设置。 private UserGroupInformation fsOwner; 本地文件的用户文件属主和文件组，可以通过hadoop.job.ugi设置，如果没有设置，那么将使用启动HDFS的用户（通过whoami获得）和该用户所在的组（通过groups获得）作为值。 private String supergroup; 对应配置项dfs.permissions.supergroup，应用在defaultPermission中，是系统的超级组。 private PermissionStatus defaultPermission; 缺省权限，缺省用户为fsOwner，缺省用户组为supergroup，缺省权限为0777，可以通过dfs.upgrade.permission修改。 private long capacityTotal, capacityUsed, capacityRemaining; 系统总容量/已使用容量/剩余容量 private int totalLoad = 0; 系统总连接数，根据DataNode心跳信息跟新。 private long pendingReplicationBlocksCount, underReplicatedBlocksCount, scheduledReplicationBlocksCount; 分别是成员变量pendingReplications（正在复制的数据块），neededReplications（需要复制的数据块）的大小，scheduledReplicationBlocksCount是当前正在处理的复制工作数目。 public FSDirectory dir; 指向系统使用的FSDirectory对象。 BlocksMap blocksMap = new BlocksMap(); 保存数据块到INode和DataNode的映射关系 public CorruptReplicasMap corruptReplicas = new CorruptReplicasMap(); 保存损坏（如：校验没通过）的数据块到对应DataNode的关系，CorruptReplicasMap类图如下，类只有一个成员变量，保存Block到一个DatanodeDescriptor的集合的映射和这个映射上的一系列操作： Map datanodeMap = new TreeMap(); 保存了StorageID à DatanodeDescriptor的映射，用于保证DataNode使用的Storage的一致性。 private Map> recentInvalidateSets 保存了每个DataNode上无效但还存在的数据块（StorageID à ArrayList）。 Map> recentInvalidateSets 保存了每个DataNode上有效，但需要删除的数据块（StorageID à TreeSet），这种情况可能发生在一个DataNode故障后恢复后，上面的数据块在系统中副本数太多，需要删除一些数据块。 HttpServer infoServer; int infoPort; Date startTime; 用于内部信息传输的HTTP请求服务器（Servlet的容器）。现在有/fsck，/getimage，/listPaths/*，/data/*和/fileChecksum/*，我们后面还会继续讨论。 ArrayList heartbeats; 所有目前活着的DataNode，线程HeartbeatMonitor会定期检查。 private UnderReplicatedBlocks neededReplications 需要进行复制的数据块。UnderReplicatedBlocks的类图如下，它其实是一个数组，数组的下标是优先级（0的优先级最高，如果数据块只有一个副本，它的优先级是0），数组的内容是一个Block集合。UnderReplicatedBlocks提供一些方法，对Block进行增加，修改，查找和删除。 private PendingReplicationBlocks pendingReplications; 保存正在复制的数据块的相关信息。PendingReplicationBlocks的类图如下： 其中，pendingReplications保存了所有正在进行复制的数据块，使用Map是需要一些附加的信息PendingBlockInfo。这些信息包括时间戳，用于检测是否已经超时，和现在进行复制的数目numReplicasInProgress。timedOutItems是超时的复制项，超时的复制项在FSNamesystem的processPendingReplications方法中被删除，并从新复制。timerThread是用于检测复制超时的线程的句柄，对应的线程是PendingReplicationMonitor的一个实例，它的run方法每隔一段会检查是否有超时的复制项，如果有，将该数据块加到timedOutItems中。Timeout是run方法的检查间隔，defaultRecheckInterval是缺省值。PendingReplicationBlocks和PendingBlockInfo的方法都很简单。 public LeaseManager leaseManager = new LeaseManager(this); 租约管理器。 Hadoop源代码分析（二五） 继续对FSNamesystem进行分析。 Daemon hbthread = null; // HeartbeatMonitor thread public Daemon lmthread = null; // LeaseMonitor thread Daemon smmthread = null; // SafeModeMonitor thread public Daemon replthread = null; // Replication thread NameNode上的线程，分别对应DataNode心跳检查，租约检查，安全模式检查和数据块复制，我们会在后面介绍这些线程对应的功能。 volatile boolean fsRunning = true; long systemStart = 0; 系统运行标志和系统启动时间。 接下来是一堆系统的参数，比方说系统每个DataNode节点允许的最大数据块数，心跳检查间隔时间等… … // The maximum number of replicates we should allow for a single block private int maxReplication; // How many outgoing replication streams a given node should have at one time private int maxReplicationStreams; // MIN_REPLICATION is how many copies we need in place or else we disallow the write private int minReplication; // Default replication private int defaultReplication; // heartbeatRecheckInterval is how often namenode checks for expired datanodes private long heartbeatRecheckInterval; // heartbeatExpireInterval is how long namenode waits for datanode to report // heartbeat private long heartbeatExpireInterval; //replicationRecheckInterval is how often namenode checks for new replication work private long replicationRecheckInterval; //decommissionRecheckInterval is how often namenode checks if a node has finished decommission private long decommissionRecheckInterval; // default block size of a file private long defaultBlockSize = 0; private int replIndex = 0; 和neededReplications配合，记录下一个进行复制的数据块位置。 public static FSNamesystem fsNamesystemObject; 哈哈，不用介绍了，还是static的。 private String localMachine; private int port; 本机名字和RPC端口。 private SafeModeInfo safeMode; // safe mode information 记录安全模式的相关信息。 安全模式是这样一种状态，系统处于这个状态时，不接受任何对名字空间的修改，同时也不会对数据块进行复制或删除数据块。NameNode启动的时候会自动进入安全模式，同时也可以手工进入（不会自动离开）。系统启动以后，DataNode会报告目前它拥有的数据块的信息，当系统接收到的Block信息到达一定门槛，同时每个Block都有dfs.replication.min个副本后，系统等待一段时间后就离开安全模式。这个门槛定义的参数包括： l           dfs.safemode.threshold.pct：接受到的Block的比例，缺省为95%，就是说，必须DataNode报告的数据块数目占总数的95%，才到达门槛； l           dfs.replication.min：缺省为1，即每个副本都存在系统中； l           dfs.replication.min：等待时间，缺省为0，单位秒。 SafeModeInfo的类图如下： threshold，extension和safeReplication保存的是上面说的3个参数。Reached等于-1表明安全模式是关闭的，0表示安全模式打开但是系统还没达到threshold。blockTotal是计算threshold时的分母，blockSafe是分子，lastStatusReport用于控制写日志的间隔。 SafeModeInfo(Configuration conf)使用配置文件的参数，是NameNode正常启动时使用的构造函数，SafeModeInfo()中，this.threshold = 1.5f使得系统用于处于安全模式。 enter()使系统进入安全模式，leave()会使系统离开安全模式，canLeave()用于检查是否能离开安全模式而needEnter()，则判断是否应该进入安全模式。checkMode()检查系统状态，如果必要，则进入安全模式。其他的方法都比价简单，大多为对成员变量的访问。 讨论完类SafeModeInfo，我们来分析一下SafeModeMonitor，它用于定期检查系统是否能够离开安全模式（smmthread就是它的一个实例）。系统离开安全模式后，smmthread会被重新赋值为null。 · · 大小: 19.3 KB · · 大小: 98.7 KB Hadoop源代码分析（二六） (没想到需要分页啦) private Host2NodesMap host2DataNodeMap = new Host2NodesMap(); 保存了主机名（String）到DatanodeDescriptor数组的映射（Host2NodesMap唯一的成员变量为HashMap map，它的方法都是对这个map进行操作）。 NetworkTopology clusterMap = new NetworkTopology(); private DNSToSwitchMapping dnsToSwitchMapping; 定义了HDFS的网络拓扑，网络拓扑对应选择数据块副本的位置很重要。如在一个层次型的网络中，接到同一个交换机的两个节点间的网络速度，会比跨越多个交换机的两个节点间的速度快，但是，如果某交换机故障，那么它对接到它上面的两个节点会同时有影响，但跨越多个交换机的两个节点，这种影响会小得多。下面是NetworkTopology相关的类图： Hadoop实现了一个树状的拓扑结构抽象，其中，Node接口，定义了网络节点的一些方法，NodeBase是Node的一个实现，提供了叶子节点的一些方法（明显它没有子节点），而InnerNode则实现了树的内部节点，如果我们考虑一个网络部署的话，那么叶子节点是服务器，而InnerNode则是服务器所在的机架或交换机或路由器。Node提供了对网络位置信息（采用类似文件树的方式），节点名称和Node所在的树的深度的方法。NodeBase提供了一个简单的实现。InnerNode是NetworkTopology的内部类，对比NodeBase，它的clildren保存了所有的子节点，这样的话，就可以构造一个拓扑树。这棵树的叶子可能是服务器，也可能是机架，内部则是机架或者是路由器等设备，InnerNode提供了一系列的方法区分处理这些信息。 NetworkTopology的add方法和remove用于在拓扑结构中加入节点和删除节点，同时也给出一些get*方法，用于获取一些对象内部的信息，如getDistance，可以获取两个节点的距离，而isOnSameRack可以判断两个节点是否处于同一个机架。chooseRandom有两个实现，用于在一定范围内（另一个还有一个排除选项）随机选取一个节点。chooseRandom在选择数据块副本位置的时候调用。 DNSToSwitchMapping配合上面NetworkTopology，用于确定某一个节点的网络位置信息，它的唯一方法，可以通过一系列机器的名字找出它们对应的网络位置信息。目前有支持两种方法，一是通过命令行方式，将节点名作为输入，输出为网络位置信息（RawScriptBasedMapping执行命令CachedDNSToSwitchMapping缓存结果），还有一种就是利用配置参数hadoop.configured.node.mapping静态配置（StaticMapping）。 ReplicationTargetChooser replicator; 用于为数据块备份选择目标，例如，用户写文件时，需要选择一些DataNode，作为数据块的存放位置，这时候就利用它来选择目标地址。chooseTarget是ReplicationTargetChooser中最重要的方法，它通过内部的一个NetworkTopology对象，计算出一个DatanodeDescriptor数组，该数组就是选定的DataNode，同时，顺序就是最佳的数据流顺序（还记得我们讨论DataXceiver些数据的那个图吗？）。 private HostsFileReader hostsReader; 保存了系统中允许/不允许连接到NameNode的机器列表。 private Daemon dnthread = null; 线程句柄，该线程用于检测DataNode上的Decommission进程。例如，某节点被列入到不允许连接到NameNode的机器列表中（HostsFileReader），那么，该节点会进入Decommission状态，它上面的数据块会被复制到其它节点，复制结束后机器进入DatanodeInfo.AdminStates.DECOMMISSIONED，这台机器就可以从HDFS中撤掉。 private long maxFsObjects = 0; // maximum number of fs objects 系统能拥有的INode最大数（配置项dfs.max.objects，0为无限制）。 private final GenerationStamp generationStamp = new GenerationStamp(); 系统的时间戳生产器。 private int blockInvalidateLimit = FSConstants.BLOCK_INVALIDATE_CHUNK; 发送给DataNode删除数据块消息中，能包含的最大数据块数。比方说，如果某DataNode上有250个Block需要被删除，而这个参数是100，那么一共会有3条删除数据块消息消息，前面两条包含了100个数据块，最后一条是50个。 private long accessTimePrecision = 0; 用于控制文件的access时间的精度，也就是说，小于这个精度的两次对文件访问，后面的那次就不做记录了。 Hadoop源代码分析（二七） 我们接下来分析NameNode.java的成员变量，然后两个类综合起来，分析它提供的接口，并配合说明接口上请求对应的处理流程。 前面已经介绍过了，NameNode实现了接口ClientProtocol，DatanodeProtocol和NamenodeProtocol，分别提供给客户端/DataNode/从NameNode访问。由于NameNode的大部分功能在类FSNamesystem中实现，那么NameNode.java的成员变量就很少了。 public FSNamesystem namesystem; 指向FSNamesystem对象。 private Server server; NameNode的RPC服务器实例。 private Thread emptier; 处理回收站的线程句柄。 private int handlerCount = 2; 还记得我们分析RPC的服务器时提到的服务器请求处理线程（Server.Handle）吗？这个参数给出了server中服务器请求处理线程的数目，对应配置参数为dfs.namenode.handler.count。 private boolean supportAppends = true; 是否支持append操作，对应配置参数为dfs.support.append。 private InetSocketAddress nameNodeAddress = null; NameNode的地址，包括IP地址和监听端口。 下面我们来看NameNode的启动过程。main方法是系统的入口，它会调用createNameNode创建NameNode实例。createNameNode分析命令行参数，如果是FORMAT或FINALIZE，调用对应的方法后退出，如果是其他的参数，将创建NameNode对象。NameNode的构造函数会调用initialize，初始化NameNode的成员变量，包括创建RPC服务器，初始化FSNamesystem，启动RPC服务器和回收站线程。 FSNamesystem的构造函数会调用initialize方法，去初始化上面我们分析过的一堆成员变量。几个重要的步骤包括加载FSImage，设置系统为安全模式，启动各个工作线程和HTTP服务器。系统的一些参数是在setConfigurationParameters中初始化的，其中一些值的计算比较麻烦，而且也可能被其它部分的code引用的，就独立出来了，如getNamespaceDirs和getNamespaceEditsDirs。initialize对应的是close方法，很简单，主要是停止initialize中启动的线程。 对应于initialize方法，NameNode也提供了对应的stop方法，用于初始化时出错系统能正确地退出。 NameNode的format和finalize操作，都是先构造FSNamesystem，然后利用FSNamesystem的FSImage提供的对应方法完成的。我们在分析FSImage.java时，已经了解了这部分的功能。 Hadoop源代码分析（二八） 万事俱备，我们可以来分析NameNode上的流程啦。 首先我们来看NameNode上实现的ClientProtocol，客户端通过这个接口，可以对目录树进行操作，打开/关闭文件等。 getBlockLocations用于确定文件内容的位置，它的输入参数为：文件名，偏移量，长度，返回值是一个LocatedBlocks对象（如下图），它携带的信息很多，大部分字段我们以前都讨论过。 getBlockLocations直接调用NameSystem的同名方法。NameSystem中这样的方法首先会检查权限和对参数进行检查（如偏移量和长度要大于0），然后再调用实际的方法。找LocatedBlocks先找src对应的INode，然后通过INode的getBlocks方法，可以拿到该节点的Block列表，如果返回为空，表明该INode不是文件，返回null；如果Block列表长度为0，以空的Block数组构造返回的LocatedBlocks。 如果Block数组不为空，则通过请求的偏移量和长度，就可以把这个区间涉及的Block找出来，对于每一个block，执行： l           通过BlocksMap我们可以找到它存在于几个DataNode上（BlocksMap.numNodes方法）； l           计算包含该数据块但数据块是坏的DataNode的数目（通过NameSystem.countNodes方法，间接访问CorruptReplicasMap中的信息）； l           计算坏数据块的数目（CorruptReplicasMap.numCorruptReplicas方法，应该和上面的数相等）； l           通过上面的计算，我们得到现在还OK的数据块数目； l           从BlocksMap中找出所有OK的数据块对应的DatanodeDescriptor（DatanodeInfo的父类）； l           创建对应的LocatedBlock。 收集到每个数据块的LocatedBlock信息后，很自然就能构造LocatedBlocks对象。getBlockLocations其实只是一个读的方法，请求到了NameNode以后只需要查表就行了。 create方法，该方法用于在目录树上创建文件（创建目录使用mkdir），需要的参数比较多，包括文件名，权限，客户端名，是否覆盖已存在文件，副本数和块大小。NameNode的create调用NameSystem的startFile方法（startFile需要的参数clientMachine从线程局部变量获取）。 startFile方法先调用startFileInternal完成操作，然后调用logSync，等待日志写完后才返回。 startFileInternal不但服务于startFile，也被appendFile调用（通过参数append区分）。方法的最开始是一堆检查，包括：安全模式，文件名src是否正确，权限，租约，replication参数，overwrite参数（对append操作是判断src指向是否存在并且是文件）。租约检查很简单，如果通过FSDirectory.getFileINode(src)得到的文件是出于构造状态，表明有客户正在操作该文件，这时会抛出异常AlreadyBeingCreatedException。 如果对于创建操作，会通过FSDirectory的addFile往目录树上添加一个文件并在租约管理器里添加一条记录。 对于append操作，执行的是构造一个新的INodeFileUnderConstruction并替换原有的节点，然后在租约管理器里添加一条记录。 总的来说，最简单的create流程就是在目录树上创建一个INodeFileUnderConstruction对象并往租约管理器里添加一条记录。 我们顺便分析一下append吧，它的返回值是LocatedBlock，比起getBlockLocations，它只需要返回数组的一项。appendFile是NameSystem的实现方法，它首先调用上面讨论的startFileInternal方法（已经在租约管理器里添加了一条记录）然后写日志。然后寻找对应文件INodeFile中记录的最后一个block，并通过BlocksMap.getStoredBlock()方法得到BlockInfo，然后再从BlocksMap中获得所有的DatanodeDescriptor，就可以构造LocatedBlock了。需要注意的，如果该Block在需要被复制的集合（UnderReplicatedBlocks）中，移除它。 如果文件刚被创建或者是最后一个数据块已经写满，那么append会返回null，这是客户端需要使用addBlock，为文件添加数据块。 Hadoop源代码分析（二九） public boolean setReplication(String src, short replication ) throws IOException; setReplication，设置文件src的副本数为replication，返回值为boolean，在FSNameSystem中，调用方法setReplicationInternal，然后写日志。 setReplicationInternal上来自然是检查参数了，然后通过FSDirectory的setReplication，设置新的副本数，并获取老的副本数。根据新旧数，决定删除/复制数据块。 增加副本数通过调用updateNeededReplications，为了获取UnderReplicatedBlocks. update需要的参数，FSNameSystem提供了内部方法countNodes和getReplication，获得对应的数值（这两个函数都很简单）。 proccessOverReplicatedBlock用于减少副本数，它被多个方法调用： 主要参数有block，副本数，目标DataNode，源DataNode（用于删除）。proccessOverReplicatedBlock首先找出block所在的，处于非Decommission状态的DataNode的信息，然后调用chooseExcessReplicates。chooseExcessReplicates执行： l           按机架位置，对DatanodeDescriptor进行分组； l           将DataNode分为两个集合，分别是一个机架包含一个以上的数据块的和剩余的； l           选择可以删除的数据块（顺序是：源DataNode，同一个机架上的，剩余的），把它加到recentInvalidateSets中。 public void setPermission(String src, FsPermission permission ) throws IOException; setPermission，用于设置文件的访问权限。非常简单，首先检查是否有权限，然后调用FSDirectory.setPermission修改文件访问权限。 public void setOwner(String src, String username, String groupname ) throws IOException; public void setTimes(String src, long mtime, long atime) throws IOException; public void setQuota(String path, long namespaceQuota, long diskspaceQuota) throws IOException; setOwner，设置文件的文件主和文件组，setTimes，设置文件的访问时间和修改时间，setQuota,设置某路径的空间限额和空间额度，和setPermission类似，调用FSDirectory的对应方法，简单。 public boolean setSafeMode(FSConstants.SafeModeAction action) throws IOException; 前面我们已经介绍了NameNode的安全模式，客户端通过上面的方法，可以让NameNode进入（SAFEMODE_ENTER）/退出（SAFEMODE_LEAVE）安全模式或查询（SAFEMODE_GET）状态。FSNamesystem的setSafeMode处理这个命令，对于进入安全模式的请求，如果系统现在不处于安全模式，那么创建一个SafeModeInfo对象（创建的这个对象有别于启动时创建的那个SafeModeInfo，它不会自动退出，因为threshold=1.5f），这标志着系统进入安全模式。退出安全模式很简单，将safeMode赋空就可以啦。 public FileStatus[] getListing(String src) throws IOException; 分析完set*以后，我们来看get*。getListing对应于UNIX系统的ls命令，返回值是FileStatus数组，FileStatus的类图如下，它其实给出了文件的详细信息，如大小，文件主等等。其实，这些信息都存在INode*中，我们只需要把这些信息搬到FileStatus中就OK啦。FSNamesystem和FSDirectory中都有同名方法，真正干活的地方在FSDirectory中。getListing不需要写日志。 public long[] getStats() throws IOException; getStatus得到的是文件系统的信息，UNIX对应命令为du，它的实现更简单，所有的信息都存放在FSNamesystem对象里。 public DatanodeInfo[] getDatanodeReport(FSConstants.DatanodeReportType type) throws IOException; getDatanodeReport，获取当前DataNode的状态，可能的选项有DatanodeReportType.ALL, IVE和DEAD。FSNamesystem的同名方法调用getDatanodeListForReport，通过HostsFileReader读取对应信息。 public long getPreferredBlockSize(String filename) throws IOException; getPreferredBlockSize，返回INodeFile.preferredBlockSize，数据块大小。 public FileStatus getFileInfo(String src) throws IOException; 和getListing类似，不再分析。 public ContentSummary getContentSummary(String path) throws IOException; 得到文件树的一些信息，如下图： public void metaSave(String filename) throws IOException; 这个也很简单，它把系统的metadata输出/添加到指定文件上（NameNode所在的文件系统）。 Hadoop源代码分析（三零） 软柿子都捏完了，我们开始啃硬骨头。前面已经分析过getBlockLocations，create，append，setReplication，setPermission和setOwner，接下来我们继续回来讨论和文件内容相关的操作。 public void abandonBlock(Block b, String src, String holder ) throws IOException; abandonBlock用于放弃一个数据块。普通的文件系统中并没有“放弃”操作，HDFS出现放弃数据块的原因，如下图所示。当客户端通过其他操作（如下面要介绍的addBlock方法）获取LocatedBlock后，可以打开到一个block的输出流，由于从DataNode出错到NameNode发现这个信息，需要有一段时间（NameNode长时间收到DataNode心跳），打开输出流可能出错，这时客户端可以向NameNode请求放弃这个数据块。 abandonBlock的处理不是很复杂，首先检查租约（调用checkLease方法。block对应的文件存在，文件处于构造状态，租约拥有者匹配），如果通过检查，调用FSDirectory的removeBlock，从INodeFileUnderConstruction/BlocksMap/CorruptReplicasMap中删除block，然后通过logOpenFile()记录变化（logOpenFile真是万能啊）。 public LocatedBlock addBlock(String src, String clientName) throws IOException; 写HDFS的文件时，如果数据块被写满，客户端可以通过addBlock创建新的数据块。具体的创建工作由FSNamesystem的getAdditionalBlock方法完成，当然上来就是一通检查（是否安全模式，命名/存储空间限额，租约，数据块副本数，保证DataNode已经上报数据块状态），然后通过ReplicationTargetChooser，选择复制的目标（如果目标数不够副本数，又是一个异常），然后，就可以分配数据块了。allocateBlock创建一个新的Block对象，然后调用addBlock，检查参数后把数据块加到BlocksMap对象和对应的INodeFile对象中。allocateBlock返回后，getAdditionalBlock还会继续更新一些需要记录的信息，最后返回一个新构造的LocatedBlock。 public boolean complete(String src, String clientName) throws IOException; 当客户端完成对数据块的写操作后，调用complete完成写操作。方法complete如果返回是false，那么，客户端需要继续调用complete方法。 FSNamesystem的同名方法调用completeFileInternal，它会： l           检查环境； l           获取src对应的INode； l           如果INode存在，并且处于构造状态，获取数据块； l           如果获取数据块返回空，返回结果CompleteFileStatus.OPERATION_FAILED，FSNamesystem的complete会抛异常返回； l           如果上报文件完成的DataNode数不够系统最小的副本数，返回STILL_WAITING； l           调用finalizeINodeFileUnderConstruction； l           返回成功COMPLETE_SUCCESS 其中，对finalizeINodeFileUnderConstruction的处理包括： l           释放租约； l           将对应的INodeFileUnderConstruction对象转换为INodeFile对象，并在FSDirectory进行替换； l           调用FSDirectory.closeFile关闭文件，其中会写日志logCloseFile(path, file)。 l           检查副本数，如果副本数小于INodeFile中的目标数，那么添加数据块复制任务。 我们可以看到，complete一个文件还是比较复杂的，需要释放很多的资源。 public void reportBadBlocks(LocatedBlock[] blocks) throws IOException; 调用reportBadBlocks的地方比较多，客户端可能调用，DataNode上也可能调用。 由于上报的是个数组，reportBadBlocks会循环处理，调用FSNamesystem的markBlockAsCorrupt方法。markBlockAsCorrupt方法需要两个参数，blk（数据块）和dn（所在的DataNode信息）。如果系统目前副本数大于要求，那么直接调用invalidateBlock方法。 方法invalidateBlock很简单，在检查完系统环境以后，先调用addToInvalidates方法往FSNamesystem.recentInvalidateSets添加一项，然后调用removeStoredBlock方法。 removeStoredBlock被多个方法调用，它会执行： l           从BlocksMap中删除记录removeNode(block, node)； l           如果目前系统中还有其他副本，调用decrementSafeBlockCount（可能的调整安全模式参数）和updateNeededReplications（跟新可能存在的block复制信息，例如，现在系统中需要复制1个数据块，那么更新后，需要复制2个数据块）； l           如果目前系统中有多余数据块等待删除（在excessReplicateMap中），那么移除对应记录； l           删除在CorruptReplicasMap中的记录（可能有）。 removeStoredBlock其实也是涉及了多处表操作，包括BlocksMap，excessReplicateMap和CorruptReplicasMap。 我们回到markBlockAsCorrupt，如果系统目前副本数小于要求，那么很显然，我们需要对数据块进行复制。首先将现在的数据块加入到CorruptReplicasMap中，然后调用updateNeededReplications，跟新复制信息。 markBlockAsCorrupt这个流程太复杂了，我们还是画个图吧： · · 大小: 96.2 KB · · 大小: 57.9 KB · · 大小: 34.9 KB · · 大小: 31.8 KB Hadoop源代码分析（三一） 下面是和目录树相关的方法。 public boolean rename(String src, String dst) throws IOException; 更改文件名。调用FSNamesystem的renameTo，干活的是renameToInternal，最终调用FSDirectory的renameTo方法，如果成功，更新租约的文件名，如下： changeLease(src, dst, dinfo); public boolean delete(String src) throws IOException; public boolean delete(String src, boolean recursive) throws IOException; 第一个已经废弃不用，使用第二个方法。 最终使用deleteInternal，该方法调用FSDirectory.delete()。 public boolean mkdirs(String src, FsPermission masked) throws IOException; 在做完一系列检查以后，调用FSDirectory.mkdirs()。 public FileStatus[] getListing(String src) throws IOException; 前面我们已经讨论了。 下面是其它和系统维护管理的方法。 public void renewLease(String clientName) throws IOException; 就是调用了一下leaseManager.renewLease(holder)，没有其他的事情需要做，简单。 public void refreshNodes() throws IOException; 还记得我们前面分析过NameNode上有个DataNode在线列表和DataNode离线列表吗，这个命令可以让NameNode从新读这两个文件。当然，根据前后DataNode的状态，一共有4种情况，其中有3种需要修改。 对于从工作状态变为离线的，需要将上面的DataNode复制到其他的DataNode，需要调用updateNeededReplications方法（前面我们已经讨论过这个方法了）。 对于从离线变为工作的DataNode，只需要改变一下状态。 public void finalizeUpgrade() throws IOException; finalize一个升级，确认客户端有超级用户权限以后，调用FSImage.finalizeUpgrade()。 public void fsync(String src, String client) throws IOException; 将文件信息持久化。在检查租约信息后，调用FSDirectory的persistBlocks，将文件的原信息通过logOpenFile(path, file)写日志。 Hadoop源代码分析（三二） 搞定ClientProtocol，接下来是DatanodeProtocol部分。接口如下： public DatanodeRegistration register(DatanodeRegistration nodeReg ) throws IOException 用于DataNode向NameNode登记。输入和输出参数都是DatanodeRegistration，类图如下： 前面讨论DataNode的时候，我们已经讲过了DataNode的注册过程，我们来看NameNode的过程。下面是主要步骤： l           检查该DataNode是否能接入到NameNode； l           准备应答，更新请求的DatanodeID； l           从datanodeMap（保存了StorageID à DatanodeDescriptor的映射，用于保证DataNode使用的Storage的一致性）得到对应的DatanodeDescriptor，为nodeS； l           从Host2NodesMap（主机名到DatanodeDescriptor数组的映射）中获取DatanodeDescriptor，为nodeN； l           如果nodeN!=null同时nodeS!=nodeN（后面的条件表明表明DataNode上使用的Storage发生变化），那么我们需要先在系统中删除nodeN（removeDatanode，下面再讨论），并在Host2NodesMap中删除nodeN； l           如果nodeS存在，表明前面已经注册过，则： 1. 更新网络拓扑（保存在NetworkTopology），首先在NetworkTopology中删除nodeS，然后跟新nodeS的相关信息，调用resolveNetworkLocation，获得nodeS的位置，并从新加到NetworkTopology里； 2. 更新心跳信息（register也是心跳）； l           如果nodeS不存在，表明这是一个新注册的DataNode，执行 1. 如果注册信息的storageID为空，表明这是一个全新的DataNode，分配storageID； 2. 创建DatanodeDescriptor，调用resolveNetworkLocation，获得位置信息； 3. 调用unprotectedAddDatanode（后面分析）添加节点； 4. 添加节点到NetworkTopology中； 5. 添加到心跳数组中。 上面的过程，我们遗留了两个方法没分析，removeDatanode的流程如下： l           更新系统的状态，包括capacityTotal，capacityUsed，capacityRemaining和totalLoad； l           从心跳数组中删除节点，并标记节点isAlive属性为false； l           从BlocksMap中删除这个节点上的所有block，用了（三零）分析到的removeStoredBlock方法； l           调用unprotectedAddDatanode； l           从NetworkTopology中删除节点信息。 unprotectedAddDatanode很简单，它只是更新了Host2NodesMap的信息。 Hadoop源代码分析（三三） 下面来看一个大家伙： public DatanodeCommand sendHeartbeat(DatanodeRegistration nodeReg, long capacity, long dfsUsed, long remaining, int xmitsInProgress, int xceiverCount) throws IOException DataNode发送到NameNode的心跳信息。细心的人会发现，请求的内容还是DatanodeRegistration，应答换成DatanodeCommand了。DatanodeCommand类图如下： 前面介绍DataNode时，已经分析过了DatanodeCommand支持的命令： DNA_TRANSFER：拷贝数据块到其他DataNode DNA_INVALIDATE：删除数据块 DNA_SHUTDOWN：关闭DataNode DNA_REGISTER：DataNode重新注册 DNA_FINALIZE：提交升级 DNA_RECOVERBLOCK：恢复数据块 有了上面这些基础，我们来看FSNamesystem.handleHeartbeat的处理过程： l           调用getDatanode方法找对应的DatanodeDescriptor，保存于变量nodeinfo（可能为null）中，如果现有NameNode上记录的StorageID和请求的不一样，返回DatanodeCommand.REGISTER，让DataNode从新注册。 l           如果发现当前节点需要关闭（已经isDecommissioned），抛异常DisallowedDatanodeException。 l           nodeinfo是空或者现在状态不是活的，返回DatanodeCommand.REGISTER，让DataNode从新注册。 l           更新系统的状态，包括capacityTotal，capacityUsed，capacityRemaining和totalLoad； l           接下来按顺序看有没有可能的恢复数据块/拷贝数据块到其他DataNode/删除数据块/升级命令（不讨论）。一次返回只能有一条命令，按上面优先顺序。 下面分析应答的命令是如何构造的。 首先是DNA_RECOVERBLOCK（恢复数据块），那是个非常长的流程，同时需要回去讨论DataNode上的一些功能，我们在后面介绍它。 对于DNA_TRANSFER（拷贝数据块到其他DataNode），从DatanodeDescriptor.replicateBlocks中取出尽可能多的项目，放到BlockCommand中。在DataNode中，命令由transferBlocks执行，前面我们已经分析过啦。 删除数据块DNA_INVALIDATE也很简单，从DatanodeDescriptor.invalidateBlocks中获取尽可能多的项目，放到BlockCommand中，DataNode中的动作，我们也分析过。 我们来讨论DNA_RECOVERBLOCK（恢复数据块），在讨论DataNode的过程中，我们没有讲这个命令是用来干什么的，还有它在DataNode上的处理流程，是好好分析分析这个流程的时候了。DNA_RECOVERBLOCK命令通过DatanodeDescriptor.getLeaseRecoveryCommand获取，获取过程很简单，将DatanodeDescriptor对象中队列recoverBlocks的所有内容取出，放入BlockCommand的Block中，设置BlockCommand为DNA_RECOVERBLOCK，就OK了。 关键是，这个队列里的信息是用来干什么的。我们先来看那些操作会向这个队列加东西，调用关系图如下： 租约有两个超时时间，一个被称为软超时（1分钟），另一个是硬超时（1小时）。如果租约软超时，那么就会触发internalReleaseLease方法，如下： void internalReleaseLease(Lease lease, String src) throws IOException 该方法执行： l           检查src对应的INodeFile，如果不存在，不处于构造状态，返回； l           文件处于构造状态，而文件目标DataNode为空，而且没有数据块，则finalize该文件（该过程在completeFileInternal中已经讨论过，租约在过程中被释放），并返回； l           文件处于构造状态，而文件目标DataNode为空，数据块非空，则将最后一个数据块存放的DataNode目标取出（在BlocksMap中），然后设置为文件现在的目标DataNode； l           调用INodeFileUnderConstruction.assignPrimaryDatanode，该过程会挑选一个目前还活着的DataNode，作为租约的主节点，并把加到该DataNode的recoverBlocks队列中； l           更新租约。 上面分析了租约软超时的情况下NameNode发生租约恢复的过程。DataNode上收到这个命令后，将会启动一个新的线程，该线程为每个Block调用recoverBlock方法：recoverBlock(blocks[i], false, targets[i], true)。 private LocatedBlock recoverBlock(Block block, boolean keepLength, DatanodeID[] datanodeids, boolean closeFile) throws IOException 它的流程并不复杂，但是分支很多，如下图（蓝线是上面输入，没有异常走的流程）： 首先是判断进来的Block是否在ongoingRecovery中，如果存在，返回，不存在，加到ongoingRecovery中。 接下来是个循环（框内部分是循环体，奇怪，没找到表示循环的符号），对每一个DataNode，获取Block的BlockMetaDataInfo（下面还会分析），这需要调用到DataNode间通信的接口上的方法getBlockMetaDataInfo。然后分情况看要不要把信息保存下来（图中间的几个判断），其中包括要进行同步的节点。 根据参数，更新数据块信息，然后调用syncBlock并返回syncBlock生产的LocatedBlock。 上面的这一圈，对于我们这个输入常数来说，就是把Block的长度，更新成为拥有最新时间戳的最小长度值，并得到要更新的节点列表，然后调用syncBlock更新各节点。 getBlockMetaDataInfo用于获取Block的BlockMetaDataInfo，包括Block的generationStamp，最后校验时间，同时它还会检查数据块文件的元信息，如果出错，会抛出异常。 syncBlock定义如下： private LocatedBlock syncBlock(Block block, List syncList, boolean closeFile) 它的流程是： l           如果syncList为空，通过commitBlockSynchronization向NameNode提交这次恢复； l           syncList不为空，那么先NameNode申请一个新的Stamp，并根据上面得到的长度，构造一个新的数据块信息newblock； l           对于没一个syncList中的DataNode，调用它们上面的updateBlock，更新信息；更新信息如果返回OK，记录下来； l           如果更新了信息的DataNode不为空，调用commitBlockSynchronization提交这次恢复；并生成LocatedBlock； l           如果更新的DataNode为空，抛异常。 通过syncBlock，所有需要恢复的DataNode上的Block信息都被更新。 DataNode上的updateBlock方法我们前面已经介绍了，就不再分析。 下面我们来看NameNode的commitBlockSynchronization方法，它在上面的过程中用于提交数据块恢复： public void commitBlockSynchronization(Block block, long newgenerationstamp, long newlength, boolean closeFile, boolean deleteblock, DatanodeID[] newtargets ) 参数分别是block，数据块；newgenerationstamp，新的时间戳；newlength，新长度；closeFile，是否关闭文件，deleteblock，是否删除文件；newtargets，新的目标列表。 上面的两次调用，输入参数分别是： commitBlockSynchronization(block, 0, 0, closeFile, true, DatanodeID.EMPTY_ARRAY); commitBlockSynchronization(block, newblock.getGenerationStamp(), newblock.getNumBytes(), closeFile, false, nlist); 处理流程是： l           参数检查； l           获取对应的文件，记为pendingFile； l           BlocksMap中删除老的信息； l           如果deleteblock为true，从pendingFile删除Block记录； l           否则，更新Block的信息； l           如果不关闭文件，那么写日志保存更新，返回； l           关闭文件的话，调用finalizeINodeFileUnderConstruction。 这块比较复杂，不仅涉及了NameNode和DataNode间的通信，而且还存在对于DataNode和DataNode间的通信（DataNode间的通信就只支持这两个方法，如下图）。后面介绍DFSClient的时候，我们还会再回来分析它的功能，以获取全面的理解。 Hadoop源代码分析（三四） 继续对NameNode实现的接口做分析。 public DatanodeCommand blockReport(DatanodeRegistration nodeReg, long[] blocks) throws IOException DataNode向NameNode报告它拥有的所有数据块，其中，参数blocks包含了数组化以后数据块的信息。FSNamesystem.processReport处理这个请求。一番检查以后，调用DatanodeDescriptor的reportDiff，将上报的数据块分成三组，分别是： l           删除：其它情况； l           加入：BlocksMap中有数据块，但目前的DatanodeDescriptor上没有对应信息； l           使无效：BlocksMap中没有找到数据块。 对于删除的数据块，调用removeStoredBlock，这个方法我们前面已经分析过啦。 对应需要加入的数据块，调用addStoredBlock方法，处理流程如下： l           从BlocksMap获取现在的信息，记为storedBlock；如果为空，返回； l           记录block和DatanodeDescriptor的关系； l           新旧数据块记录不是同一个（我们这个流程是肯定不是啦）： 1. 如果现有数据块长度为0，更新为上报的block的值； 2. 如果现有数据块长度比新上报的长，invalidateBlock（前面分析过，很简单的一个方法）当前数据块； 3. 如果现有数据块长度比新上报的小，那么会删除所有老的数据块（还是通过invalidateBlock），并更新BlocksMap中数据块的大小信息； 4. 跟新可用存储空间等信息； l           根据情况确定数据块需要复制的数目和目前副本数； l           如果文件处于构建状态或系统现在是安全模式，返回； l           处理当前副本数和文件的目标副本数不一致的情况； l           如果当前副本数大于系统设定门限，开始删除标记为无效的数据块。 还是给个流程图吧： 对于标记为使无效的数据块，调用addToInvalidates方法，很简单的方法，直接加到FSNamesystem的成员变量recentInvalidateSets中。 public void blockReceived(DatanodeRegistration registration, Block blocks[], String[] delHints) DataNode可以通过blockReceived，向NameNode报告它最近接受到的数据块，同时给出如果数据块副本数太多时，可以删除数据块的节点（参数delHints）。在DataNode中，这个信息是通过方法notifyNamenodeReceivedBlock，记录到对应的列表中。 NameNode上的处理不算复杂，对输入参数进行检查以后，调用上面分析的addStoredBlock方法。然后在PendingReplicationBlocks对象中删除相应的block。 public void errorReport(DatanodeRegistration registration, int errorCode, String msg) 向NameNode报告DataNode上的一个错误，如果错误是硬盘错，会删除该DataNode，其它情况只是简单地记录收到一条出错信息。 public NamespaceInfo versionRequest() throws IOException; 从NameNode上获取NamespaceInfo，该信息用于构造DataNode上的DataStorage。 UpgradeCommand processUpgradeCommand(UpgradeCommand comm) throws IOException; 我们不讨论。 public void reportBadBlocks(LocatedBlock[] blocks) throws IOException 报告错误的数据块。NameNode会循环调用FSNamesystem的markBlockAsCorrupt方法。处理流程不是很复杂，找对应的INodeFile，如果副本数够，那么调用invalidateBlock，使该DataNode上的Block无效；如果副本数不够，加Block到CorruptReplicasMap中，然后准备对好数据块进行复制。 目前为止，我们已经完成了NameNode上的ClientProtocol和DatanodeProtocol的分析了，NamenodeProtocol我们在理解从NameNode的时候，才会进行分析。 Hadoop源代码分析（三五） 除了对外提供的接口，NameNode上还有一系列的线程，不断检查系统的状态，下面是这些线程的功能分析。 在NameNode中，定义了如下线程： Daemon hbthread = null; // HeartbeatMonitor thread public Daemon lmthread = null; // LeaseMonitor thread Daemon smmthread = null; // SafeModeMonitor thread public Daemon replthread = null; // Replication thread private Daemon dnthread = null; PendingReplicationBlocks中也有一个线程： Daemon timerThread = null; NameNode内嵌的HTTP服务器中自然也有线程，这块我们就不分析啦。 HttpServer infoServer; 心跳线程用于对DataNode的心态进行检查，以间隔heartbeatRecheckInterval运行heartbeatCheck方法。如果在一定时间内没收到DataNode的心跳信息，我们就认为该节点已经死掉，调用removeDatanode（前面分析过）将DataNode标记为无效。 租约lmthread用于检查租约的硬超时，如果租约硬超时，调用前面分析过的internalReleaseLease，释放租约。 smmthread运行的SafeModeMonitor我们前面已经分析过了。 replthread运行ReplicationMonitor，这个线程会定期调用computeDatanodeWork和processPendingReplications。 computeDatanodeWork会执行computeDatanodeWork或computeInvalidateWork。computeDatanodeWork从neededReplications中扫描，取出需要复制的项，然后： l 检查文件不存在或者处于构造状态；如果是，从队列中删除复制项，退出对复制项的处理（接着处理下一个）； l 得到当前数据块副本数并选择复制的源DataNode，如果空，退出对复制项的处理； l 再次检查副本数（很可能有DataNode从故障中恢复），如果发现不需要复制，从队列中删除复制项，退出对复制项的处理； l 选择复制的目标，如果目标空，退出对复制项的处理； l 将复制的信息（数据块和目标DataNode）加入到源目标DataNode中；在目标DataNode中记录复制请求； l 从队列中将复制项移动到pendingReplications。 可见，这个方法执行后，复制项从neededReplications挪到pendingReplications中。DataNode在某次心跳的应答中，可以拿到相应的信息，执行复制操作。 computeInvalidateWork当然是用于删除无效的数据块。它的主要工作在invalidateWorkForOneNode中完成。和上面computeDatanodeWork类似，不过它的处理更简单，将recentInvalidateSets的数据通过DatanodeDescriptor.addBlocksToBeInvalidated挪到DataNode中。 dnthread执行的是DecommissionedMonitor，它的run方法周期调用decommissionedDatanodeCheck，再到checkDecommissionStateInternal，定期将完成Decommission任务的DataNode状态从DECOMMISSION_INPROGRESS改为DECOMMISSIONED。 PendingReplicationMonitor中的线程用于对处在等待复制状态的数据块进行检查。如果发现长时间该数据块没被复制，那么会将它挪到timedOutItems中。请参考PendingReplicationBlocks的讨论。 infoServer的相关线程我们就不分析了，它们都用于处理HTTP请求。 上面已经总结了NameNode上的一些为特殊任务启动的线程，除了这些线程，NameNode上还运行着RPC服务器的相关线程，具体可以看前面章节。 在我们开始分析Secondary NameNode前，我们给出了以NameNode上一些状态转移图，大家可以通过这个图，更好理解NameNode。 NameNode： DataNode： 文件： Block，比较复杂： 上面的图不是很严格，只是用于帮助大家理解NameNode对Block复杂的处理过程。 稍微说明一下，“Block in inited DataNode”表明这个数据块在一个刚初始化的DataNode上。“Block in INodeFile”是数据块属于某个文件，“Block in INodeFileUnderConstruction” 表明这数据块属于一个正在构建的文件，当然，处于这个状态的Block可能因为租约恢复而转移到“Block in Recover”。右上方描述了需要复制的数据块的状态，UnderReplicatedBlocks和PendingReplicationBlocks的区别在于Block是否被插入到某一个DatanodeDescriptor中。Corrupt和Invalidate的就好理解啦。 Hadoop源代码分析（三六） 转战进入Secondary NameNode，前面的分析我们有事也把它称为从NameNode，从NameNode在HDFS里是个小配角。 跟Secondary NameNode有关系的类不是很多，如下图： 首先要讨论的是NameNode和Secondary NameNode间的通信。NameNode上实现了接口NamenodeProtocol（如下图），就是用于NameNode和Secondary NameNode间的命令通信。 NameNode和Secondary NameNode间数据的通信，使用的是HTTP协议，HTTP的容器用的是jetty，TransferFsImage是文件传输的辅助类。 GetImageServlet的doGet方法目前支持取FSImage(getimage)，取日志(getedit)和存FSImage(putimage)。例如： http://localhost:50070/getimage?getimage 可以获取FSImage。 http://localhost:50070/getimage?getedit 可以获取日志文件。 保存FSImage需要更多的参数，它的流程很好玩，Secondary NameNode发送一个HTTP请求到NameNode，启动NameNode上一个HTTP客户端到Secondary NameNode上去下载FSImage，下载需要的一些信息，都放在从NameNode的HTTP请求中。 我们先来考察Secondary NameNode持久化保存的信息： [hadoop@localhost namesecondary]$ ls –R .: current image in_use.lock previous.checkpoint ./current: edits fsimage fstime VERSION ./image: fsimage ./previous.checkpoint: edits fsimage fstime VERSION in_use.lock的用法和前面NameNode，DataNode的是一样的。对比NameNode保存的信息，我们可以发现Secondary NameNode上保存多了一个previous.checkpoint。CheckpointStorage就是应用于Secondary NameNode的存储类，它继承自FSImage，只添加了很少的方法。 previous.checkpoint目录保存了上一个checkpoint的信息（current里的永远是最新的），临时目录用于创建新checkpoint，成功后，老的checkpoint保存在previous.checkpoint目录中。状态图如下（基类FSImage用的是黑色）： 至于上面目录下文件的内容，和FSImage是一样的。 CheckpointStorage除了上面图中的startCheckpoint和endCheckpoint方法（上图给出了正常流程），还有： void recoverCreate(Collection dataDirs, Collection editsDirs) throws IOException 和FSImage.coverTransitionRead类似，用于分析现有目录，创建目录（如果不存在）并从可能的错误中恢复。 private void doMerge(CheckpointSignature sig) throws IOException doMerge被类SecondaryNameNode的同名方法调用，我们后面再分析。 Hadoop源代码分析（三七） Secondary NameNode的成员变量很少，主要的有： private CheckpointStorage checkpointImage; Secondary NameNode使用的Storage private NamenodeProtocol namenode; 和NameNode通信的接口 private HttpServer infoServer; 传输文件用的HTTP服务器 main方法是Secondary NameNode的入口，它最终启动线程，执行SecondaryNameNode的run。启动前的对SecondaryNameNode的构造过程也很简单，主要是创建和NameNode通信的接口和启动HTTP服务器。 SecondaryNameNode的run方法每隔一段时间执行doCheckpoint()，从NameNode的主要工作都在这一个方法里。这个方法，总的来说，会从NameNode上取下FSImage和日志，然后再本地合并，再上传回NameNode。这个过程结束后，从NameNode上保持了NameNode上持久化信息的一个备份，同时，NameNode上已经完成合并到FSImage的日志可以抛弃，一箭双雕。 具体的的流程是： 1：调用startCheckpoint，为接下来的工作准备空间。startCheckpoint会在内部做一系列的检查，然后调用CheckpointStorage的startCheckpoint方法，创建目录。 2：调用namenode的rollEditLog方法，开始一次新的检查点过程。调用会返回一个CheckpointSignature（检查点签名），在上传合并完的FSImage时，会使用这个签名。 Namenode的rollEditLog方法最终调用的是FSImage的同名方法，前面提到过这个方法，作用是关闭往edits上写的日志，打开日志到edits.new。明显，在Secondary NameNode下载fsimage和日志的时候，对命名空间的修改，将保持在edits.new的日志中。 注意，如果FSImage这个时候的状态（看下面的状态机，前面出现过一次）不是出于CheckpointStates.ROLLED_EDITS，将抛异常结束这个过程。 3：通过downloadCheckpointFiles下载fsimage和日志，并设置本地检查点状态为CheckpointStates.UPLOAD_DONE。 4：合并日志的内容到fsimage中。过程很简单，CheckpointStorage利用继承自FSImage的loadFSImage加载fsimage，loadFSEdits应用日志，然后通过saveFSImage保存。很明显，现在保存在硬盘上的fsimage是合并日志的内容以后的文件。 5：使用putFSImage上传合并日志后的fsimage（让NameNode通过HTTP到从NameNode取文件）。这个过程中，NameNode会： 调用NameNode的FSImage.validateCheckpointUpload，检查现在的状态； 利用HTTP，从Secondary NameNode获取新的fsimage； 更新结束后设置新状态。 6：调用NameNode的rollFsImage，最终调用FSImage的rollFsImage方法，前面我们已经分析过了。 7：调用本地endCheckpoint方法，结束一次doCheckpoint流程。 其实前面在分析FSImage的时候，我们在不了解Secondary NameNode的情况下，分析了很多和Checkpoint相关的方法，现在我们终于可以有一个比较统一的了解了，下面给出NameNode和Secondary NameNode的存储系统在这个流程中的状态转移图，方便大家理解。 图中右侧的状态转移图： 文件系统上的目录的变化（三六中出现）： Hadoop源代码分析（三八） 我们可以开始从系统的外部来了解HDFS了，DFSClient提供了连接到HDFS系统并执行文件操作的基本功能。DFSClient也是个大家伙，我们先分析它的一些内部类。我们先看LeaseChecker。租约是客户端对文件写操作时需要获取的一个凭证，前面分析NameNode时，已经了解了租约，INodeFileUnderConstruction的关系，INodeFileUnderConstruction只有在文件写的时候存在。客户端的租约管理很简单，包括了增加的put和删除的remove方法，run方法会定期执行，并通过ClientProtocl的renewLease，自动延长租约。 接下来我们来分析内部为文件读引入的类。 InputStream是系统的虚类，提供了3个read方法，一个skip（跳过数据）方法，一个available方法（目前流中可读的字节数），一个close方法和几个在输入流中做标记的方法（mark：标记，reset：回到标记点和markSupported：能力查询）。 FSInputStream也是一个虚类，它将接口Seekable和PositionedReadable混插到类中。Seekable提供了可以在流中定位的能力（seek，getPos和seekToNewSource），而PositionedReadable提高了从某个位置开始读的方法（一个read方法和两个readFully方法）。 FSInputChecker在FSInputStream的基础上，加入了HDFS中需要的校验功能。校验在readChecksumChunk中实现，并在内部的read1方法中调用。所有的read调用，最终都是使用read1读数据并做校验。如果校验出错，抛出异常ChecksumException。 有了支持校验功能的输入流，就可以开始构建基于Block的输入流了。我们先回顾前面提到的读数据块的请求协议： 然后我们来分析一下创建BlockReader需要的参数，newBlockReader最复杂的请求如下： public static BlockReader newBlockReader( Socket sock, String file, long blockId, long genStamp, long startOffset, long len, int bufferSize, boolean verifyChecksum, String clientName) throws IOException 其中，sock为到DataNode的socket连接，file是文件名（只是用于日志输出），其它的参数含义都很清楚，和协议基本是一一对应的。该方法会和DataNode进行对话，发送上面的读数据块的请求，处理应答并构造BlockReader对象（BlockReader的构造函数基本上只有赋值操作）。 BlockReader的readChunk用于处理DataNode送过来的数据，格式前面我们已经讨论过了，如下图。 读数据用的read，会调用父类FSInputChecker的read，最后调用readChunk，如下： read如果发现读到正确的校验码，则用过checksumOk方法，向DataNode发送成功应达。 BlockReader的主要流程就介绍完了，接下来分析DFSInputStream，它封装了DFSClient读文件内容的功能。在它的内部，不但要处理和NameNode的通信，同时通过BlockReader，处理和DataNode的交互。 DFSInputStream记录Block的成员变量是： private LocatedBlocks locatedBlocks = null; 它不但保持了文件对应的Block序列，还保持了管理Block的DataNode的信息，是DFSInputStream中最重要的成员变量。DFSInputStream的构造函数，通过类内部的openInfo方法，获取这个变量的值。openInfo间接调用了NameNode的getBlockLocations，获取LocatedBlocks。 DFSInputStream中处理数据块位置的还有下面一些函数： synchronized List getAllBlocks() throws IOException private LocatedBlock getBlockAt(long offset) throws IOException private synchronized List getBlockRange(long offset, long length) private synchronized DatanodeInfo blockSeekTo(long target) throws IOException 它们的功能都很清楚，需要注意的是他们处理过程中可能会调用再次调用NameNode的getBlockLocations，使得流程比较复杂。blockSeekTo还会创建对应的BlockReader对象，它被几个重要的方法调用（如下图）。在打开到DataNode之前，blockSeekTo会调用chooseDataNode，选择一个现在活着的DataNode。 通过上面的分析，我们已经知道了在什么时候会连接NameNode，什么时候会打开到DataNode的连接。下面我们来看读数据。read方法定义如下： public int read(long position, byte[] buffer, int offset, int length) 该方法会从流的position位置开始，读取最多length个byte到buffer中offset开始的空间中。参数检测完以后，通过getBlockRange获取要读取的数据块对应的block范围，然后，利用fetchBlockByteRange方法，读取需要的数据。 fetchBlockByteRange从某一个数据块中读取一段数据，定义如下： private void fetchBlockByteRange(LocatedBlock block, long start, long end, byte[] buf, int offset) 由于读取的内容都在一个数据块内部，这个方法会创建BlockReader，然后利用BlockReader的readAll方法，读取数据。读的过程中如果发生校验错，那么，还会通过reportBadBlocks，向NameNode报告校验错。 另一个读方法是： public synchronized int read(byte buf[], int off, int len) throws IOException 它在流的当前位置（可以通过seek方法调整）读取数据。首先它会判断当前流的位置，如果已经越过了对象现在的blockReader能读取的范围（当上次read读到数据块的尾部时，会发生这中情况），那么通过blockSeekTo打开到下一个数据块的blockReader。然后，read在当前的这个数据块中通过readBuffer读数据。主要，这个read方法只在一块数据块中读取数据，就是说，如果还有空间可以存放数据但已经到了数据块的尾部，它不会打开到下一个数据块的BlockReader继续读，而是返回，返回值包含了以读取数据的长度。 DFSDataInputStream是一个Wrapper(DFSInputStream)，我们就不讨论了。 Hadoop源代码分析（三九） 接下来当然是分析输出流了。 处于继承体系的最上方是OutputStream，它实现了Closeable（方法close）和Flushable（方法flush）接口，提供了3个不同形式的write方法，这些方法的含义都很清楚。接下来的是FSOutputSummer，它引入了HDFS写数据时需要的计算校验和的功能。FSOutputSummer的write方法会调用write1，write1中计算校验和并将用户输入的数据拷贝到对象的缓冲区中，缓冲区满了以后会调用flushBuffer，flushBuffer最终调用还是虚方法的writeChunk，这个时候，缓冲区对应的校验和缓冲区对的内容都已经准备好了。通过这个类，HDFS可以把一个流转换成为DataNode数据接口上的包格式（前面我们讨论过这个包的格式，如下）。 DFSOutputStream继承自FSOutputSummer，是一个非常复杂的类，它包含了几个内部类。我们先分析Packet，其实它对应了上面的数据包，有了上面的图，这个类就很好理解了，它的成员变量和上面数据块包含的信息基本一一对应。构造函数需要的参数有pktSize，包的大小，chunksPerPkt，chunk的数目（chunk是一个校验单元）和该包在Block中的偏移量offsetInBlock。writeData和writeChecksum用于往缓冲区里写数据/校验和。getBuffer用户获得整个包，包括包头和数据。 DataStreamer和ResponseProcessor用于写包/读应答，和我们前面讨论DataNode的Pipe写时类似，客户端写数据也需要两个线程，下图扩展了我们在讨论DataNode处理写时的示意图，包含了客户端： DataStreamer启动后进入一个循环，在没有错误和关闭标记为false的情况下，该循环首先调用processDatanodeError，处理可能的IO错误，这个过程比较复杂，我们在后面再讨论。 接着DataStreamer会在dataQueue（数据队列）上等待，直到有数据出现在队列上。DataStreamer获取一个数据包，然后判断到DataNode的连接是否是打开的，如果不是，通过DFSOutputStream.nextBlockOutputStream打开到DataNode的连接，并启动ResponseProcessor线程。 DataNode的连接准备好以后，DataStreamer获取数据包缓冲区，然后将数据包从dataQueue队列挪到ackQueue队列，最后通过blockStream，写数据。如果数据包是最后一个，那么，DataStreamer将会写一个长度域为0的包，指示DataNode数据传输结束。 DataStreamer的循环在最后一个数据包写出去以后，会等待直到ackQueue队列为空（表明所有的应答已经被接收），然后做清理动作（包括关闭socket连接，ResponseProcessor线程等），退出线程。 ResponseProcessor相对来说比较简单，就是等待来自DataNode的应答。如果是成功的应答，则删除在ackQueue的包，如果有错误，那么，记录出错的DataNode，并设置标志位。 Hadoop源代码分析（四零） 有了上面的基础，我们可以来解剖DFSOutputStream了。先看构造函数： private DFSOutputStream(String src, long blockSize, Progressable progress, int bytesPerChecksum) throws IOException DFSOutputStream(String src, FsPermission masked, boolean overwrite, short replication, long blockSize, Progressable progress, int buffersize, int bytesPerChecksum) throws IOException DFSOutputStream(String src, int buffersize, Progressable progress, LocatedBlock lastBlock, FileStatus stat, int bytesPerChecksum) throws IOException { 这些构造函数的参数主要有：文件名src；进度回调函数progress（预留接口，目前未使用）；数据块大小blockSize；Block副本数replication；每个校验chunk的大小bytesPerChecksum；文件权限masked；是否覆盖原文件标记overwrite；文件状态信息stat；文件的最后一个Block信息lastBlock；buffersize（？未见引用）。 后面两个构造函数会调用第一个构造函数，这个函数会调用父类的构造函数，并设置对象的src，blockSize，progress和checksum属性。 第二个构造函数会调用namenode.create方法，在文件空间中建立文件，并启动DataStreamer，它被DFSClient的create方法调用。第三个构造函数被DFSClient的append方法调用，显然，这种情况比价复杂，文件拥有一些数据块，添加数据往往添加在最后的数据块上。同时，append方法调用时，Client已经知道了最后一个Block的信息和文件的一些信息，如FileStatus中包含的Block大小，文件权限位等等。结合这些信息，构造函数需要计算并设置一些对象成员变量的值，并试图从可能的错误中恢复（调用processDatanodeError），最后启动DataStreamer。 我们先看正常流程，前面已经分析过，通过FSOutputSummer，HDFS客户端能将流转换成package，这个包是通过writeChunk，发送出去的，下面是它们的调用关系。 在检查完一系列的状态以后，writeChunk先等待，直到dataQueue中未发送的包小于门限值。如果现在没有可用的Packet对象，则创建一个Packet对象，往Packet中写数据，包括校验值和数据。如果数据包被写满，那么，将它放入发送队列dataQueue中。writeChunk的过程比较简单，这里的写入，也只是把数据写到本地队列，等待DataStreamer发送，没有实际写到DataNode上。 createBlockOutputStream用于建立到第一个DataNode的连接，它的声明如下： private boolean createBlockOutputStream(DatanodeInfo[] nodes, String client, boolean recoveryFlag) nodes是所有接收数据的DataNode列表，client就是客户端名称，recoveryFlag指示是否是为错误恢复建立的连接。createBlockOutputStream很简单，打开到第一个DataNode的连接，然后发送下面格式的数据包，并等待来自DataNode的Ack。如果出错，记录出错的DataNode在nodes中的位置，设置errorIndex并返回false。 当recoveryFlag指示为真时，意味着这次写是一次恢复操作，对于DataNode来说，这意味着为写准备的临时文件（在tmp目录中）可能已经存在，需要进行一些特殊处理，具体请看FSDataset的实现。 当Client写数据需要一个新的Block的时候，可以调用nextBlockOutputStream方法。 private DatanodeInfo[] nextBlockOutputStream(String client) throws IOException 这个方法的实现很简单，首先调用locateFollowingBlock（包含了重试和出错处理），通过namenode.addBlock获取一个新的数据块，返回的是DatanodeInfo列表，有了这个列表，就可以建立写数据的pipe了。下一个大动作就是调用上面的createBlockOutputStream，建立到DataNode的连接了。 有了上面的准备，我们来分析processDatanodeError，它的主要流程是： l 参数检查； l 关闭可能还打开着的blockStream和blockReplyStream； l 将未收到应答的数据块（在ackQueue中）挪到dataQueue中； l 循环执行： 1. 计算目前还活着的DataNode列表； 2. 选择一个主DataNode，通过DataNode RPC的recoverBlock方法启动它上面的恢复过程； 3. 处理可能的出错； 4. 处理恢复后Block可能的变化（如Stamp变化）； 5. 调用createBlockOutputStream到DataNode的连接。 l 启动ResponseProcessor。 这个过程涉及了DataNode上的recoverBlock方法和createBlockOutputStream中可能的Block恢复，是一个相当耗资源的方法，当系统出错的概率比较小，而且数据块上能恢复的数据很多（平均32M），还是值得这样做的。 写的流程就分析到着，接下来我们来看流的关闭，这个过程也涉及了一系列的方法，它们的调用关系如下： flushInternal会一直等待到发送队列（包括可能的currentPacket）和应答队列都为空，这意味着数据都被DataNode顺利接收。 sync作用和UNIX的sync类似，将写入数据持久化。它首先调用父类的flushBuffer方法，将可能还没拷贝到DFSOutputStream的数据拷贝回来，然后调用flushInternal，等待所有的数据都写完。然后调用namenode.fsync，持久化命名空间上的数据。 closeInternal比较复杂一点，它首先调用父类的flushBuffer方法，将可能还没拷贝到DFSOutputStream的数据拷贝回来，然后调用flushInternal，等待所有的数据都写完。接着结束两个工作线程，关闭socket，最后调用amenode.complete，通知NameNode结束一次写操作。close方法先调用closeInternal，然后再本地的leasechecker中移除对应的信息。 Hadoop源代码分析（四一） 前面分析的DFSClient内部类，占据了这个类的实现部分的2/3，我们来看剩下部分。 DFSClient的成员变量不多，而且大部分是系统的缺省配置参数，其中比较重要的是到NameNode的RPC客户端： public final ClientProtocol namenode; final private ClientProtocol rpcNamenode; 它们的差别是namenode在rpcNamenode的基础上，增加了失败重试功能。DFSClient中提供可各种构造它们的static函数，createClientDatanodeProtocolProxy用于生成到DataNode的RPC客户端。 DFSClient的构造函数也比价简单，就是初始化成员变量，close用于关闭DFSClient。 下面的功能，DFSClient只是简单地调用NameNode的对应方法（加一些简单的检查），就不罗嗦了： setReplication/rename/delete/exists（通过getFileInfo的返回值是否为空判断）/listPaths/getFileInfo/setPermission/setOwner/getDiskStatus/totalRawCapacity/totalRawUsed/datanodeReport/setSafeMode/refreshNodes/metaSave/finalizeUpgrade/mkdirs/getContentSummary/setQuota/setTimes DFSClient提供了各种create方法，它们最后都是构造一个OutputStream，并将文件名和生成的OutputStream加到leasechecker，完成创建动作。 append操作是通过namenode.append，获取最后的Block信息，然后构造一个OutputStream，并将文件名和生成的OutputStream加到leasechecker，完成append动作。 getFileChecksum用于获取文件的校验信息，它在得到数据块的位置信息后利用DataNode提供的OP_BLOCK_CHECKSUM操作，获取需要的数据，并综合起来。过程简单，方法主要是在处理OP_BLOCK_CHECKSUM需要交换的数据包。 DFSClient内部还有一些其它的辅助方法，都比较简单，就不再分析了。 Hadoop源代码分析（MapReduce概论） 大家都熟悉文件系统，在对HDFS进行分析前，我们并没有花很多的时间去介绍HDFS的背景，毕竟大家对文件系统的还是有一定的理解的，而且也有很好的文档。在分析Hadoop的MapReduce部分前，我们还是先了解系统是如何工作的，然后再进入我们的分析部分。下面的图来自http://horicky.blogspot.com/2008/11/hadoop-mapreduce-implementation.html，是我看到的讲MapReduce最好的图。 以Hadoop带的wordcount为例子（下面是启动行）： hadoop jar hadoop-0.19.0-examples.jar wordcount /usr/input /usr/output 用户提交一个任务以后，该任务由JobTracker协调，先执行Map阶段（图中M1，M2和M3），然后执行Reduce阶段（图中R1和R2）。Map阶段和Reduce阶段动作都受TaskTracker监控，并运行在独立于TaskTracker的Java虚拟机中。 我们的输入和输出都是HDFS上的目录（如上图所示）。输入由InputFormat接口描述，它的实现如ASCII文件，JDBC数据库等，分别处理对于的数据源，并提供了数据的一些特征。通过InputFormat实现，可以获取InputSplit接口的实现，这个实现用于对数据进行划分（图中的splite1到splite5，就是划分以后的结果），同时从InputFormat也可以获取RecordReader接口的实现，并从输入中生成对。有了，就可以开始做map操作了。 map操作通过context.collect（最终通过OutputCollector. collect）将结果写到context中。当Mapper的输出被收集后，它们会被Partitioner类以指定的方式区分地写出到输出文件里。我们可以为Mapper提供Combiner，在Mapper输出它的时，键值对不会被马上写到输出里，他们会被收集在list里（一个key值一个list），当写入一定数量的键值对时，这部分缓冲会被Combiner中进行合并，然后再输出到Partitioner中（图中M1的黄颜色部分对应着Combiner和Partitioner）。 Map的动作做完以后，进入Reduce阶段。这个阶段分3个步骤：混洗（Shuffle），排序（sort）和reduce。 混洗阶段，Hadoop的MapReduce框架会根据Map结果中的key，将相关的结果传输到某一个Reducer上（多个Mapper产生的同一个key的中间结果分布在不同的机器上，这一步结束后，他们传输都到了处理这个key的Reducer的机器上）。这个步骤中的文件传输使用了HTTP协议。 排序和混洗是一块进行的，这个阶段将来自不同Mapper具有相同key值的对合并到一起。 Reduce阶段，上面通过Shuffle和sort后得到的会送到Reducer. reduce方法中处理，输出的结果通过OutputFormat，输出到DFS中。 Hadoop源代码分析（MapTask） 接下来我们来分析Task的两个子类，MapTask和ReduceTask。MapTask的相关类图如下： MapTask其实不是很复杂，复杂的是支持MapTask工作的一些辅助类。MapTask的成员变量少，只有split和splitClass。我们知道，Map的输入是split，是原始数据的一个切分，这个切分由org.apache.hadoop.mapred.InputSplit的子类具体描述（前面我们是通过org.apache.hadoop.mapreduce.InputSplit介绍了InputSplit，它们对外的API是一样的）。splitClass是InputSplit子类的类名，通过它，我们可以利用Java的反射机制，创建出InputSplit子类。而split是一个BytesWritable，它是InputSplit子类串行化以后的结果，再通过InputSplit子类的readFields方法，我们可以回复出对应的InputSplit对象。 MapTask最重要的方法是run。run方法相当简单，配置完系统的TaskReporter后，就根据情况执行runJobCleanupTask，runJobSetupTask，runTaskCleanupTask或执行Mapper。由于MapReduce现在有两套API，MapTask需要支持这两套API，使得MapTask执行Mapper分为runNewMapper和runOldMapper，run*Mapper后，MapTask会调用父类的done方法。 接下来我们来分析runOldMapper，最开始部分是构造Mapper处理的InputSplit，更新Task的配置，然后就开始创建Mapper的RecordReader，rawIn是原始输入，然后分正常（使用TrackedRecordReader，后面讨论）和跳过部分记录（使用SkippingRecordReader，后面讨论）两种情况，构造对应的真正输入in。 跳过部分记录是Map的一种出错恢复策略，我们知道，MapReduce处理的数据集合非常大，而有些任务对一部分出错的数据不进行处理，对结果的影响很小（如大数据集合的一些统计量），那么，一小部分的数据出错导致已处理的大量结果无效，是得不偿失的，跳过这部分记录，成了Mapper的一种选择。 Mapper的输出，是通过MapOutputCollector进行的，也分两种情况，如果没有Reducer，那么，用DirectMapOutputCollector（后面讨论），否则，用MapOutputBuffer（后面讨论）。 构造完Mapper的输入输出，通过构造配置文件中配置的MapRunnable，就可以执行Mapper了。目前系统有两个MapRunnable：MapRunner和MultithreadedMapRunner，如下图。 原有API在这块的处理上和新API有很大的不一样。接口MapRunnable是原有API中Mapper的执行器，run方法就是用于执行用户的Mapper。MapRunner是单线程执行器，相当简单，首先，当MapTask调用： MapRunnable runner = ReflectionUtils.newInstance(job.getMapRunnerClass(), job); MapRunner的configure会在newInstance的最后被调用，configure执行的过程中，对应的Mapper会通过反射机制构造出来。 MapRunner的run方法，会先创建对应的key，value对象，然后，对InputSplit的每一对，调用Mapper的map方法，循环结束后，Mapper对应的清理方法会被调用。我们需要注意，key，value对象在run方法中是被重复使用的，就是说，每次传入Mapper的map方法的key，value都是同一个对象，只不过是里面的内容变了，对象并没有变。如果你需要保留key，value的内容，需要实现clone机制，克隆出对象的一个新备份。 相对于新API的多线程执行器，老API的MultithreadedMapRunner就比较复杂了，总体来说，就是通过阻塞队列配合Java的多线程执行器，将分发到多个线程中去处理。需要注意的是，在这个过程中，这些线程共享一个Mapper实例，如果Mapper有共享的资源，需要有一定的保护机制。 runNewMapper用于执行新版本的Mapper，比runOldMapper稍微复杂，我们就不再讨论了。 Hadoop源代码分析（MapTask辅助类 I） MapTask的辅助类主要针对Mapper的输入和输出。首先我们来看MapTask中用的的Mapper输入，在类图中，这部分位于右上角。 MapTask.TrackedRecordReader是一个Wrapper，在原有输入RecordReader的基础上，添加了收集上报统计数据的功能。 MapTask.SkippingRecordReader也是一个Wrapper，它在MapTask.TrackedRecordReader的基础上，添加了忽略部分输入的功能。在分析MapTask.SkippingRecordReader之前，我们先看一下类SortedRanges和它相关的类。 类SortedRanges.Ranges表示了一个范围，以开始位置和范围长度（这样的话就可以表示长度为0的范围）来表示一个范围，并提供了一系列的范围操作方法。注意，方法getEndIndex得到的右端点并不包含在范围内（应理解为开区间）。SortedRanges包含了一系列不重叠的范围，为了保证包含的范围不重叠，在add方法和remove方法上需要做一些处理，保证不重叠的约束。SkipRangeIterator是访问SortedRanges包含的Ranges的迭代器。 MapTask.SkippingRecordReader的实现很简单，因为要忽略的输入都保持在SortedRanges.Ranges，只需要在next方法中，判断目前范围时候落在SortedRanges.Ranges中，如果是，忽略，并将忽略的记录写文件（可配置） NewTrackingRecordReader和NewOutputCollector被新API使用，我们不分析。 MapTask的输出辅助类都继承自MapOutputCollector，它只是在OutputCollector的基础上添加了close和flush方法。 DirectMapOutputCollector用在Reducer的数目为0，就是不需要Reduce阶段的时候。它是直接通过 out = job.getOutputFormat().getRecordWriter(fs, job, finalName, reporter); 得到对应的RecordWriter，collect直接到RecordWriter上。 如果Mapper后续有reduce任务，系统会使用MapOutputBuffer做为输出，这是个比较复杂的类，有1k行左右的代码。 我们知道，Mapper是通过OutputCollector将Map的结果输出，输出的量很大，Hadoop的机制是通过一个circle buffer 收集Mapper的输出, 到了io.sort.mb * percent量的时候，就spill到disk，如下图。图中出现了两个数组和一个缓冲区，kvindices保持了记录所属的（Reduce）分区，key在缓冲区开始的位置和value在缓冲区开始的位置，通过kvindices，我们可以在缓冲区中找到对应的记录。kvoffets用于在缓冲区满的时候对kvindices的partition进行排序，排完序的结果将输出到输出到本地磁盘上，其中索引（kvindices）保持在spill{spill号}.out.index中，数据保存在spill{spill号}.out中。 当Mapper任务结束后，有可能会出现多个spill文件，这些文件会做一个归并排序，形成Mapper的一个输出（spill.out和spill.out.index），如下图： 这个输出是按partition排序的，这样的话，Mapper的输出被分段，Reducer要获取的就是spill.out中的一段。（注意，内存和硬盘上的索引结构不一样） （感谢彭帅的Hadoop Map Stage流程分析 http://www.cnblogs.com/OnlyXP/archive/2009/05/25/1488811.html） Hadoop源代码分析（MapTask辅助类，II） 有了上面Mapper输出的内存存储结构和硬盘存储结构讨论，我们来仔细分析MapOutputBuffer的流程。 首先是成员变量。最先初始化的是作业配置job和统计功能reporter。通过配置，MapOutputBuffer可以获取本地文件系统（localFs和rfs），Reducer的数目和Partitioner。 SpillRecord是文件spill.out{spill号}.index在内存中的对应抽象（内存数据和文件数据就差最后的校验和），该文件保持了一系列的IndexRecord，如下图： IndexRecord有3个字段，分别是startOffset：记录偏移量，rawLength：初始长度，partLength：实际长度（可能有压缩）。SpillRecord保持了一系列的IndexRecord，并提供方法用于添加记录（没有删除记录的操作，因为不需要），获取记录，写文件，读文件（通过构造函数）。 接下来是一些和输出缓存区kvbuffer，缓存区记录索引kvindices和缓存区记录索引排序工作数组kvoffsets相关的处理，下面的图有助于说明这段代码。 这部分依赖于3个配置参数，io.sort.spill.percent是kvbuffer，kvindices和kvoffsets的总大小（以M为单位，缺省是100，就是100M，这一部分是MapOutputBuffer中占用存储最多的）。io.sort.record.percent是kvindices和kvoffsets占用的空间比例（缺省是0.05）。前面的分析我们已经知道kvindices和kvoffsets，如果记录数是N的话，它占用的空间是4N*4bytes，根据这个关系和io.sort.record.percent的值，我们可以计算出kvindices和kvoffsets最多能有多少个记录，并分配相应的空间。参数io.sort.spill.percent指示当输出缓冲区或kvindices和kvoffsets记录数量到达对应的占用率的时候，会启动spill，将内存缓冲区的记录存放到硬盘上，softBufferLimit和softRecordLimit为对应的字节数。 值对输出到缓冲区是通过Serializer串行化的，这部分的初始化跟在上面输出缓存后面。接下来是一些计数器和可能的数据压缩处理器的初始化，可能的Combiner和combiner工作的一些配置。 最后是启动spillThread，该Thread会检查内存中的输出缓存区，在满足一定条件的时候将缓冲区中的内容spill到硬盘上。这是一个标准的生产者-消费者模型，MapTask的collect方法是生产者，spillThread是消费者，它们之间同步是通过spillLock（ReentrantLock）和spillLock上的两个条件变量（spillDone和spillReady）完成的。 先看生产者，MapOutputBuffer.collect的主要流程是： l 报告进度和参数检测（符合Mapper的输出约定）； l spillLock.lock()，进入临界区； l 如果达到spill条件，设置变量并通过spillReady.signal()，通知spillThread；并等待spill结束（通过spillDone.await()等待）； l spillLock.unlock()； l 输出key，value并更新kvindices和kvoffsets（注意，方法collect是synchronized，key和value各自输出，它们也会占用连续的输出缓冲区）； kvstart，kvend和kvindex三个变量在判断是否需要spill和spill是否结束的过程中很重要，kvstart是有效记录开始的下标，kvindex是下一个可做记录的位置，kvend的作用比较特殊，它在一般情况下kvstart==kvend，但开始spill的时候它会被赋值为kvindex的值，spill结束时，它的值会被赋给kvstart，这时候kvstart==kvend。这就是说，如果kvstart不等于kvend，系统正在spill，否则，kvstart==kvend，系统处于普通工作状态。其实在代码中，我们可以看到很多kvstart==kvend的判断。 下面我们分情况，讨论kvstart，kvend和kvindex的配合。初始化的时候，它们都被赋值0。 下图给出了一个没有spill的记录添加过程： 注意kvindex和kvnext的关系，取模实现了循环缓冲区 如果在添加记录的过程中，出现spill（多种条件），那么，主要的过程如下： 首先还是计算kvnext，主要，这个时候kvend==kvstart（图中没有画出来）。如果spill条件满足，那么，kvindex的值会赋给kvend（这是kvend不等于kvstart），从kvstart和kvend的大小关系，我们可以知道记录位于数组的那一部分（左边是kvstart对输出使用的缓冲区，也有类似的过程。 Collect在处理输出时，会处理一个MapBufferTooSmallException，这是value的串行化结果太大，不能一次放入缓冲区的指示，这种情况下我们需要调用spillSingleRecord，特殊处理。 Hadoop源代码分析（MapTask辅助类，III） 接下来讨论的是key，value的输出，这部分比较复杂，不过有了前面kvstart，kvend和kvindex配合的分析，有利于我们理解这部分的代码。 输出缓冲区中，和kvstart，kvend和kvindex对应的是bufstart，bufend和bufmark。这部分还涉及到变量bufvoid，用于表明实际使用的缓冲区结尾（见后面BlockingBuffer.reset分析），和变量bufmark，用于标记记录的结尾。这部分代码需要bufmark，是因为key或value的输出是变长的，（前面元信息记录大小是常量，就不需要这样的变量）。 最好的情况是缓冲区没有翻转和value串行化结果很小，如下图： 先对key串行化，然后对value做串行化，临时变量keystart，valstart和valend分别记录了key结果的开始位置，value结果的开始位置和value结果的结束位置。 串行化过程中，往缓冲区写是最终调用了Buffer.write方法，我们后面再分析。 如果key串行化后出现bufindex < keystart，那么会调用BlockingBuffer的reset方法。原因是在spill的过程中需要对排序，这种情况下，传递给RawComparator的必须是连续的二进制缓冲区，通过BlockingBuffer.reset方法，解决这个问题。下图解释了如何解决这个问题： 当发现key的串行化结果出现不连续的情况时，我们会把bufvoid设置为bufmark，见缓冲区开始部分往后挪，然后将原来位于bufmark到bufvoid出的结果，拷到缓冲区开始处，这样的话，key串行化的结果就连续存放在缓冲区的最开始处。 上面的调整有一个条件，就是bufstart前面的缓冲区能够放下整个key串行化的结果，如果不能，处理的方式是将bufindex置0，然后调用BlockingBuffer内部的out的write方法直接输出，这实际调用了Buffer.write方法，会启动spill过程，最终我们会成功写入key串行化的结果。 下面我们看write方法。key，value串行化过程中，往缓冲区写数据是最终调用了Buffer.write方法，又是一个复杂的方法。 l do-while循环，直到我们有足够的空间可以写数据（包括缓冲区和kvindices和kvoffsets） u 首先我们计算缓冲区连续写是否写满标志buffull和缓冲区非连续情况下有足够写空间标志wrap（这个实在拗口），见下面的讨论；条件（buffull && !wrap）用于判断目前有没有足够的写空间； u 在spill没启动的情况下（kvstart == kvend），分两种情况，如果数组中有记录(kvend != kvindex)，那么，根据需要（目前输出空间不足或记录数达到spill条件）启动spill过程；否则，如果空间还是不够（buffull && !wrap），表明这个记录非常大，以至于我们的内存缓冲区不能容下这么大的数据量，抛MapBufferTooSmallException异常； u 如果空间不足同时spill在运行，等待spillDone； l 写数据，注意，如果buffull，则写数据会不连续，则写满剩余缓冲区，然后设置bufindex=0，并从bufindex处接着写。否则，就是从bufindex处开始写。 下图给出了缓冲区连续写是否写满标志buffull和缓冲区非连续情况下有足够写空间标志wrap计算的几种可能: 情况1和情况2中，buffull判断为从bufindex到bufvoid是否有足够的空间容纳写的内容，wrap是图中白颜色部分的空间是否比输入大，如果是，wrap为true；情况3和情况4中，buffull判断bufindex到bufstart的空间是否满足条件，而wrap肯定是false。明显，条件（buffull && !wrap）满足时，目前的空间不够一次写。 接下来我们来看spillSingleRecord，只是用于写放不进内存缓冲区的对。过程很流水，首先是创建SpillRecord记录，输出文件和IndexRecord记录，然后循环，构造SpillRecord并在恰当的时候输出记录（如下图），最后输出spill{n}.index文件。 前面我们提过spillThread，在这个系统中它是消费者，这个消费者相当简单，需要spill时调用函数sortAndSpill，进行spill。sortAndSpill和spillSingleRecord类似，函数的开始也是创建SpillRecord记录，输出文件和IndexRecord记录，然后，需要在kvoffsets上做排序，排完序后顺序访问kvoffsets，也就是按partition顺序访问记录。 按partition循环处理排完序的数组，如果没有combiner，则直接输出记录，否则，调用combineAndSpill，先做combin然后输出。循环的最后记录IndexRecord到SpillRecord。 sortAndSpill最后是输出spill{n}.index文件。 combineAndSpill比价简单，我们就不分析了。 BlockingBuffer中最后要分析的方法是flush方法。调用flush方法，意味着Mapper的结果都已经collect了，需要对缓冲区做一些最后的清理，并合并spill{n}文件产生最后的输出。 缓冲区处理部分很简单，先等待可能的spill过程完成，然后判断缓冲区是否为空，如果不是，则调用sortAndSpill，做最后的spill，然后结束spill线程。 flush合并spill{n}文件是通过mergeParts方法。如果Mapper最后只有一个spill{n}文件，简单修改该文件的文件名就可以。如果Mapper没有任何输出，那么我们需要创建哑输出（dummy files）。如果spill{n}文件多于1个，那么按partition循环处理所有文件，将处于处理partition的记录输出。处理partition的过程中可能还会再次调用combineAndSpill，最记录再做一次combination，其中还涉及到工具类Merger，我们就不再深入研究了。 Hadoop源代码分析（Task的内部类和辅助类） 从前面的图中，我们可以发现Task有很多内部类，并拥有大量类成员变量，这些类配合Task完成相关的工作，如下图。 MapOutputFile管理着Mapper的输出文件，它提供了一系列get方法，用于获取Mapper需要的各种文件，这些文件都存放在一个目录下面。 我们假设传入MapOutputFile的JobID为job_200707121733_0003，TaskID为task_200707121733_0003_m_000005。MapOutputFile的根为 {mapred.local.dir}/taskTracker/jobcache/{jobid}/{taskid}/output 在下面的讨论中，我们把上面的路径记为{MapOutputFileRoot} 以上面JogID和TaskID为例，我们有： {mapred.local.dir}/taskTracker/jobcache/job_200707121733_0003/task_200707121733_0003_m_000005/output 需要注意的是，{mapred.local.dir}可以包含一系列的路径，那么，Hadoop会在这些根路径下找一个满足要求的目录，建立所需的文件。MapOutputFile的方法有两种，结尾带ForWrite和不带ForWrite，带ForWrite用于创建文件，它需要一个文件大小作为参数，用于检查磁盘空间。不带ForWrite用于获取以建立的文件。 getOutputFile：文件名为{MapOutputFileRoot}/file.out； getOutputIndexFile：文件名为{MapOutputFileRoot}/file.out.index getSpillFile：文件名为{MapOutputFileRoot}/spill{spillNumber}.out getSpillIndexFile：文件名为{MapOutputFileRoot}/spill{spillNumber}.out.index 以上四个方法用于Task子类MapTask中； getInputFile：文件名为{MapOutputFileRoot}/map_{mapId}.out 用于ReduceTask中。我们到使用到他们的地方再介绍相应的应用场景。 介绍完临时文件管理以后，我们来看Task.CombineOutputCollector，它继承自org.apache.hadoop.mapred.OutputCollector，很简单，只是一个OutputCollector到IFile.Writer的Adapter，活都让IFile.Writer干了。 ValuesIterator用于从RawKeyValueIterator（Key，Value都是DataInputBuffer，ValuesIterator要求该输入已经排序）中获取符合RawComparator comparator的值的迭代器。它在Task中有一个简单子类，CombineValuesIterator。 Task.TaskReporter用于向JobTracker提交计数器报告和状态报告，它实现了计数器报告Reporter和状态报告StatusReporter。为了不影响主线程的工作，TaskReporter有一个独立的线程，该线程通过TaskUmbilicalProtocol接口，利用Hadoop的RPC机制，向JobTracker报告Task执行情况。 FileSystemStatisticUpdater用于记录对文件系统的对/写操作字节数，是个简单的工具类。 Hadoop源代码分析（mapreduce.lib.partition/reduce/output） Map的结果，会通过partition分发到Reducer上，Reducer做完Reduce操作后，通过OutputFormat，进行输出，下面我们就来分析参与这个过程的类。 Mapper的结果，可能送到可能的Combiner做合并，Combiner在系统中并没有自己的基类，而是用Reducer作为Combiner的基类，他们对外的功能是一样的，只是使用的位置和使用时的上下文不太一样而已。 Mapper最终处理的结果对，是需要送到Reducer去合并的，合并的时候，有相同key的键/值对会送到同一个Reducer那，哪个key到哪个Reducer的分配过程，是由Partitioner规定的，它只有一个方法，输入是Map的结果对和Reducer的数目，输出则是分配的Reducer（整数编号）。系统缺省的Partitioner是HashPartitioner，它以key的Hash值对Reducer的数目取模，得到对应的Reducer。 Reducer是所有用户定制Reducer类的基类，和Mapper类似，它也有setup，reduce，cleanup和run方法，其中setup和cleanup含义和Mapper相同，reduce是真正合并Mapper结果的地方，它的输入是key和这个key对应的所有value的一个迭代器，同时还包括Reducer的上下文。系统中定义了两个非常简单的Reducer，IntSumReducer和LongSumReducer，分别用于对整形/长整型的value求和。 Reduce的结果，通过Reducer.Context的方法collect输出到文件中，和输入类似，Hadoop引入了OutputFormat。OutputFormat依赖两个辅助接口：RecordWriter和OutputCommitter，来处理输出。RecordWriter提供了write方法，用于输出和close方法，用于关闭对应的输出。OutputCommitter提供了一系列方法，用户通过实现这些方法，可以定制OutputFormat生存期某些阶段需要的特殊操作。我们在TaskInputOutputContext中讨论过这些方法（明显，TaskInputOutputContext是OutputFormat和Reducer间的桥梁）。 OutputFormat和RecordWriter分别对应着InputFormat和RecordReader，系统提供了空输出NullOutputFormat（什么结果都不输出，NullOutputFormat.RecordWriter只是示例，系统中没有定义），LazyOutputFormat（没在类图中出现，不分析），FilterOutputFormat（不分析）和基于文件FileOutputFormat的SequenceFileOutputFormat和TextOutputFormat输出。 基于文件的输出FileOutputFormat利用了一些配置项配合工作，包括mapred.output.compress：是否压缩；mapred.output.compression.codec：压缩方法；mapred.output.dir：输出路径；mapred.work.output.dir：输出工作路径。FileOutputFormat还依赖于FileOutputCommitter，通过FileOutputCommitter提供一些和Job，Task相关的临时文件管理功能。如FileOutputCommitter的setupJob，会在输出路径下创建一个名为_temporary的临时目录，cleanupJob则会删除这个目录。 SequenceFileOutputFormat输出和TextOutputFormat输出分别对应输入的SequenceFileInputFormat和TextInputFormat，我们就不再详细分析啦。 Hadoop源代码分析（IFile） Mapper的输出，在发送到Reducer前是存放在本地文件系统的，IFile提供了对Mapper输出的管理。我们已经知道，Mapper的输出是对，IFile以记录的形式存放了这些数据。为了保存键值对的边界，很自然IFile需要保存key-len和value-len。 和IFile相关的类图如下： 其中，文件流形式的输入和输出是由IFIleInputStream和IFIleOutputStream抽象。以记录形式的读/写操作由IFile.Reader/IFile.Writer提供，IFile.InMemoryReader用于读取存在于内存中的IFile文件格式数据。 我们以输出为例，来分析这部分的实现。首先是下图的和序列化反序列化相关的Serialization/Deserializer，这部分的code是在包org.apache.hadoop.io.serializer。序列化由Serializer抽象，通过Serializer的实现，用户可以利用serialize方法把对象序列化到通过open方法打开的输出流里。Deserializer提供的是相反的过程，对应的方法是deserialize。hadoop.io.serializer中还实现了配合工作的Serialization和对应的工厂SerializationFactory。两个具体的实现是WritableSerialization和JavaSerialization，分别对应了Writeble的序列化反序列化和Java本身带的序列化反序列化。 有了Serializer/Deserializer，我们来分析IFile.Writer。Writer的构造函数是： public Writer(Configuration conf, FSDataOutputStream out, Class keyClass, Class valueClass, CompressionCodec codec, Counters.Counter writesCounter) conf，配置参数，out是Writer的输出，keyClass 和valueClass 是输出的Kay，Value的class属性，codec是对输出进行压缩的方法，参数writesCounter用于对输出字节数进行统计的Counters.Counter。通过这些参数，我们可以构造我们使用的支持压缩功能的输出流（类成员out，类成员rawOut保存了构造函数传入的out），相关的计数器，还有就是Kay，Value的Serializer方法。 Writer最主要的方法是append方法（居然不是write方法，呵呵），有两种形式： public void append(K key, V value) throws IOException { public void append(DataInputBuffer key, DataInputBuffer value) append(K key, V value)的主要过程是检查参数，然后将key和value序列化到DataOutputBuffer中，并获取序列化后的长度，最后把长度（2个）和DataOutputBuffer中的结果写到输出，并复位DataOutputBuffer和计数。append(DataInputBuffer key, DataInputBuffer value)处理过程也比较类似，就不再分析了。 close方法中需要注意的是，我们需要标记文件尾，或者是流结束。目前是通过写2个值为EOF_MARKER的长度来做标记。 IFileOutputStream是用于配合Writer的输出流，它会在IFiles的最后添加校验数据。当Writer调用IFileOutputStream的write操作时，IFileOutputStream计算并保持校验和，流被close的时候，校验结果会写到对应文件的文件尾。实际上存放在磁盘上的文件是一系列的记录和校验结果。 Reader的相关过程，我们就不再分析了。 Hadoop源代码分析（*IDs类和*Context类） 我们开始来分析Hadoop MapReduce的内部的运行机制。用户向Hadoop提交Job（作业），作业在JobTracker对象的控制下执行。Job被分解成为Task（任务），分发到集群中，在TaskTracker的控制下运行。Task包括MapTask和ReduceTask，是MapReduce的Map操作和Reduce操作执行的地方。这中任务分布的方法比较类似于HDFS中NameNode和DataNode的分工，NameNode对应的是JobTracker，DataNode对应的是TaskTracker。JobTracker，TaskTracker和MapReduce的客户端通过RPC通信，具体可以参考HDFS部分的分析。 我们先来分析一些辅助类，首先是和ID有关的类，ID的继承树如下： 这张图可以看出现在Hadoop的org.apache.hadoop.mapred向org.apache.hadoop.mapreduce迁移带来的一些问题，其中灰色是标注为@Deprecated的。ID携带一个整型，实现了WritableComparable接口，这表明它可以比较，而且可以被Hadoop的io机制串行化/解串行化（必须实现compareTo/readFields/write方法）。JobID是系统分配给作业的唯一标识符，它的toString结果是job__。例子：job_200707121733_0003表明这是jobtracker 200707121733（利用jobtracker的开始时间作为ID）的第3号作业。 作业分成任务执行，任务号TaskID包含了它所属的作业ID，同时也有任务ID，同时还保持了这是否是一个Map任务（成员变量isMap）。任务号的字符串表示为task___[m|r]_，如task_200707121733_0003_m_000005表示作业200707121733_0003的000005号任务，改任务是一个Map任务。 一个任务有可能有多个执行（错误恢复/消除Stragglers等），所以必须区分任务的多个执行，这是通过类TaskAttemptID来完成，它在任务号的基础上添加了尝试号。一个任务尝试号的例子是attempt_200707121733_0003_m_000005_0，它是任务task_200707121733_0003_m_000005的第0号尝试。 JVMId用于管理任务执行过程中的Java虚拟机，我们后面再讨论。 为了使Job和Task工作，Hadoop提供了一系列的上下文，这些上下文保存了Job和Task工作的信息。 处于继承树的最上方是org.apache.hadoop.mapreduce.JobContext，前面我们已经介绍过了，它提供了Job的一些只读属性，两个成员变量，一个保存了JobID，另一个类型为JobConf，JobContext中除了JobID外，其它的信息都保持在JobConf中。它定义了如下配置项： l mapreduce.inputformat.class：InputFormat的实现 l mapreduce.map.class：Mapper的实现 l mapreduce.combine.class: Reducer的实现 l mapreduce.reduce.class：Reducer的实现 l mapreduce.outputformat.class: OutputFormat的实现 l mapreduce.partitioner.class: Partitioner的实现 同时，它提供方法，使得通过类名，利用Java反射提供的Class.forName方法，获得类对应的Class。org.apache.hadoop.mapred的JobContext对象比org.apache.hadoop.mapreduce.JobContext多了成员变量progress，用于获取进度信息，它类型为JobConf成员job指向mapreduce.JobContext对应的成员，没有添加任何新功能。 JobConf继承自Configuration，保持了MapReduce执行需要的一些配置信息，它管理着46个配置参数，包括上面mapreduce配置项对应的老版本形式，如mapreduce.map.class 对应mapred.mapper.class。这些配置项我们在使用到它们的时候再介绍。 org.apache.hadoop.mapreduce.JobContext的子类Job前面也已经介绍了，后面在讨论系统的动态行为时，再回来看它。 TaskAttemptContext用于任务的执行，它引入了标识任务执行的TaskAttemptID和任务状态status，并提供新的访问接口。org.apache.hadoop.mapred的TaskAttemptContext继承自mapreduce的对应版本，只是增加了记录进度的progress。 TaskInputOutputContext和它的子类都在包org.apache.hadoop.mapreduce中，前面已经分析过了，我们就不再罗嗦。 Hadoop源代码分析（包hadoop.mapred中的MapReduce接口） 前面已经完成了对org.apache.hadoop.mapreduce的分析，这个包提供了Hadoop MapReduce部分的应用API，用于用户实现自己的MapReduce应用。但这些接口是给未来的MapReduce应用的，目前MapReduce框架还是使用老系统（参考补丁HADOOP-1230）。下面我们来分析org.apache.hadoop.mapred，首先还是从mapred的MapReduce框架开始分析，下面的类图（灰色部分为标记为@Deprecated的类/接口）： 我们把包mapreduce的类图附在下面，对比一下，我们就会发现，org.apache.hadoop.mapred中的MapReduce API相对来说很简单，主要是少了和Context相关的类，那么，好多在mapreduce中通过context来完成的工作，就需要通过参数来传递，如Map中的输出，老版本是： output.collect(key, result); // output’s type is: OutputCollector 新版本是： context.write(key, result); // output’s type is: Context 它们分别使用OutputCollector和Mapper.Context来输出map的结果，显然，原有OutputCollector的新API中就不再需要。总体来说，老版本的API比较简单，MapReduce过程中关键的对象都有，但可扩展性不是很强。同时，老版中提供的辅助类也很多，我们前面分析的FileOutputFormat，也有对应的实现，我们就不再讨论了。 · · 大小: 141 KB Hadoop源代码分析（包mapreduce.lib.input） 接下来我们按照MapReduce过程中数据流动的顺序，来分解org.apache.hadoop.mapreduce.lib.*的相关内容，并介绍对应的基类的功能。首先是input部分，它实现了MapReduce的数据输入部分。类图如下： 类图的右上角是InputFormat，它描述了一个MapReduce Job的输入，通过InputFormat，Hadoop可以： l 检查MapReduce输入数据的正确性； l 将输入数据切分为逻辑块InputSplit，这些块会分配给Mapper； l 提供一个RecordReader实现，Mapper用该实现从InputSplit中读取输入的对。 在org.apache.hadoop.mapreduce.lib.input中，Hadoop为所有基于文件的InputFormat提供了一个虚基类FileInputFormat。下面几个参数可以用于配置FileInputFormat： l mapred.input.pathFilter.class：输入文件过滤器，通过过滤器的文件才会加入InputFormat； l mapred.min.split.size：最小的划分大小； l mapred.max.split.size：最大的划分大小； l mapred.input.dir：输入路径，用逗号做分割。 类中比较重要的方法有： protected List listStatus(Configuration job) 递归获取输入数据目录中的所有文件（包括文件信息），输入的job是系统运行的配置Configuration，包含了上面我们提到的参数。 public List getSplits(JobContext context) 将输入划分为InputSplit，包含两个循环，第一个循环处理所有的文件，对于每一个文件，根据输入的划分最大/最小值，循环得到文件上的划分。注意，划分不会跨越文件。 FileInputFormat没有实现InputFormat的createRecordReader方法。 FileInputFormat有两个子类，SequenceFileInputFormat是Hadoop定义的一种二进制形式存放的键/值文件（参考http://hadoop.apache.org/core/docs/current/api/org/apache/hadoop/io/SequenceFile.html），它有自己定义的文件布局。由于它有特殊的扩展名，所以SequenceFileInputFormat重载了listStatus，同时，它实现了createRecordReader，返回一个SequenceFileRecordReader对象。TextInputFormat处理的是文本文件，createRecordReader返回的是LineRecordReader的实例。这两个类都没有重载FileInputFormat的getSplits方法，那么，在他们对于的RecordReader中，必须考虑FileInputFormat对输入的划分方式。 FileInputFormat的getSplits，返回的是FileSplit。这是一个很简单的类，包含的属性（文件名，起始偏移量，划分的长度和可能的目标机器）已经足以说明这个类的功能。 RecordReader用于在划分中读取对。RecordReader有五个虚方法，分别是： l initialize：初始化，输入参数包括该Reader工作的数据划分InputSplit和Job的上下文context； l nextKey：得到输入的下一个Key，如果数据划分已经没有新的记录，返回空； l nextValue：得到Key对应的Value，必须在调用nextKey后调用； l getProgress：得到现在的进度； l close，来自java.io的Closeable接口，用于清理RecordReader。 我们以LineRecordReader为例，来分析RecordReader的构成。前面我们已经分析过FileInputFormat对文件的划分了，划分完的Split包括了文件名，起始偏移量，划分的长度。由于文件是文本文件，LineRecordReader的初始化方法initialize会创建一个基于行的读取对象LineReader（定义在org.apache.hadoop.util中，我们就不分析啦），然后跳过输入的最开始的部分（只在Split的起始偏移量不为0的情况下进行，这时最开始的部分可能是上一个Split的最后一行的一部分）。nextKey的处理很简单，它使用当前的偏移量作为Key，nextValue当然就是偏移量开始的那一行了（如果行很长，可能出现截断）。进度getProgress和close都很简单。 Hadoop源代码分析（包mapreduce.lib.map） Hadoop的MapReduce框架中，Map动作通过Mapper类来抽象。一般来说，我们会实现自己特殊的Mapper，并注册到系统中，执行时，我们的Mapper会被MapReduce框架调用。Mapper类很简单，包括一个内部类和四个方法，静态结构图如下： 内部类Context继承自MapContext，并没有引入任何新的方法。 Mapper的四个方法是setup，map，cleanup和run。其中，setup和cleanup用于管理Mapper生命周期中的资源，setup在完成Mapper构造，即将开始执行map动作前调用，cleanup则在所有的map动作完成后被调用。方法map用于对一次输入的key/value对进行map动作。run方法执行了上面描述的过程，它调用setup，让后迭代所有的key/value对，进行map，最后调用cleanup。 org.apache.hadoop.mapreduce.lib.map中实现了Mapper的三个子类，分别是InverseMapper（将输入 map为输出），MultithreadedMapper（多线程执行map方法）和TokenCounterMapper（对输入的value分解为token并计数）。其中最复杂的是MultithreadedMapper，我们就以它为例，来分析Mapper的实现。 MultithreadedMapper会启动多个线程执行另一个Mapper的map方法，它会启动mapred.map.multithreadedrunner.threads（配置项）个线程执行Mapper：mapred.map.multithreadedrunner.class（配置项）。MultithreadedMapper重写了基类Mapper的run方法，启动N个线程（对应的类为MapRunner）执行mapred.map.multithreadedrunner.class（我们称为目标Mapper）的run方法（就是说，目标Mapper的setup和cleanup会被执行多次）。目标Mapper共享同一份InputSplit，这就意味着，对InputSplit的数据读必须线程安全。为此，MultithreadedMapper引入了内部类SubMapRecordReader，SubMapRecordWriter，SubMapStatusReporter，分别继承自RecordReader，RecordWriter和StatusReporter，它们通过互斥访问MultithreadedMapper的Mapper.Context，实现了对同一份InputSplit的线程安全访问，为Mapper提供所需的Context。这些类的实现方法都很简单。 Hadoop源代码分析（包org.apache.hadoop.mapreduce） 有了前一节的分析，我们来看一下具体的接口，它们都处于包org.apache.hadoop.mapreduce中。 上面的图中，类可以分为4种。右上角的是从Writeable继承的，和Counter（还有CounterGroup和Counters，也在这个包中，并没有出现在上面的图里）和ID相关的类，它们保持MapReduce过程中需要的一些计数器和标识；中间大部分是和Context相关的*Context类，它为Mapper和Reducer提供了相关的上下文；关于Map和Reduce，对应的类是Mapper，Reducer和描述他们的Job（在Hadoop 中一次计算任务称之为一个job，下面的分析中，中文为“作业”，相应的task我们称为“任务”）；图中其他类是配合Mapper和Reduce工作的一些辅助类。 如果你熟悉HTTPServlet， 那就能很轻松地理解Hadoop采用的结构，把整个Hadoop看作是容器，那么Mapper和Reduce就是容器里的组件，*Context保存了组件的一些配置信息，同时也是和容器通信的机制。 和ID相关的类我们就不再讨论了。我们先看JobContext，它位于*Context继承树的最上方，为Job提供一些只读的信息，如Job的ID，名称等。下面的信息是MapReduce过程中一些较关键的定制信息： （来自http://www.ibm.com/developerworks/cn/opensource/os-cn-hadoop2/index.html）： 参数 作用 缺省值 其它实现 InputFormat 将输入的数据集切割成小数据集 InputSplits, 每一个 InputSplit 将由一个 Mapper 负责处理。此外 InputFormat 中还提供一个 RecordReader 的实现, 将一个 InputSplit 解析成 对提供给 map 函数。 TextInputFormat (针对文本文件，按行将文本文件切割成 InputSplits, 并用 LineRecordReader 将 InputSplit 解析成 对，key 是行在文件中的位置，value 是文件中的一行) SequenceFileInputFormat OutputFormat 提供一个 RecordWriter 的实现，负责输出最终结果 TextOutputFormat (用 LineRecordWriter 将最终结果写成纯文件文件,每个 对一行，key 和 value 之间用 tab 分隔) SequenceFileOutputFormat OutputKeyClass 输出的最终结果中 key 的类型 LongWritable OutputValueClass 输出的最终结果中 value 的类型 Text MapperClass Mapper 类，实现 map 函数，完成输入的 到中间结果的映射 IdentityMapper (将输入的 原封不动的输出为中间结果) LongSumReducer, LogRegexMapper, InverseMapper CombinerClass 实现 combine 函数，将中间结果中的重复 key 做合并 null (不对中间结果中的重复 key 做合并) ReducerClass Reducer 类，实现 reduce 函数，对中间结果做合并，形成最终结果 IdentityReducer (将中间结果直接输出为最终结果) AccumulatingReducer, LongSumReducer InputPath 设定 job 的输入目录, job 运行时会处理输入目录下的所有文件 null OutputPath 设定 job 的输出目录，job 的最终结果会写入输出目录下 null MapOutputKeyClass 设定 map 函数输出的中间结果中 key 的类型 如果用户没有设定的话，使用 OutputKeyClass MapOutputValueClass 设定 map 函数输出的中间结果中 value 的类型 如果用户没有设定的话，使用 OutputValuesClass OutputKeyComparator 对结果中的 key 进行排序时的使用的比较器 WritableComparable PartitionerClass 对中间结果的 key 排序后，用此 Partition 函数将其划分为R份,每份由一个 Reducer 负责处理。 HashPartitioner (使用 Hash 函数做 partition) KeyFieldBasedPartitioner PipesPartitioner Job继承自JobContext，提供了一系列的set方法，用于设置Job的一些属性（Job更新属性，JobContext读属性），同时，Job还提供了一些对Job进行控制的方法，如下： l mapProgress：map的进度（0—1.0）； l reduceProgress：reduce的进度（0—1.0）； l isComplete：作业是否已经完成； l isSuccessful：作业是否成功； l killJob：结束一个在运行中的作业； l getTaskCompletionEvents：得到任务完成的应答（成功/失败）； l killTask：结束某一个任务；