深入解析Hadoop源码：分布式存储与MapReduce框架

需积分: 10 170 浏览量更新于2024-07-21 1 收藏 6.2MB PDF 举报

Hadoop代码分析深入探讨了Google的分布式云计算核心技术在Apache Hadoop框架中的应用和实现。Hadoop项目起源于对Google Cluster、Chubby、GFS、BigTable和MapReduce这五项关键技术的模仿和开源。其中： 1. Chubby被Apache的ZooKeeper所继承，提供了分布式环境下的统一命名服务，确保数据的一致性。 2. Google File System (GFS)的理念被转化为Hadoop Distributed File System (HDFS)，它是一个高可用、高扩展性的分布式文件系统，为大数据处理提供存储支持。 3. BigTable的概念演化成了HBase，一个分布式列式数据库，适合海量数据的查询和管理。 4. MapReduce的实现则是Hadoop的核心组件，负责分布式任务的调度和执行，通过其编程模型简化了大规模数据处理的复杂性。 Hadoop包间的依赖关系复杂，因为HDFS作为一个基础模块，提供了API来隐藏底层实现细节，使得不同层次的功能可以无缝集成。例如，conf包依赖于fs包，因为系统配置的读取涉及到了文件系统操作，而fs包进一步抽象了文件系统功能，形成了一种网状的依赖结构。 Hadoop的关键部分主要集中在图中的蓝色区域，即HDFS和MapReduce。对于HDFS的分析，有助于理解其他基于Hadoop的系统，因为它构成了整个框架的基石。而MapReduce则包含了编程接口和分布式计算引擎，是Hadoop的灵魂。后续的Hadoop源代码分析将分别对各个包的功能进行深入剖析，包括提供实用工具的tool包，以及与数据处理、文件操作等紧密相关的其他组件。通过对Hadoop源代码的逐个模块研究，开发者可以更好地掌握Hadoop的工作原理，优化性能，以及如何扩展和定制这一强大的大数据处理框架。

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 对象时的差别：

剩余82页未读，继续阅读

青山酒旗

粉丝: 3
资源: 12

深入解析Hadoop源码：分布式存储与MapReduce框架

Hadoop源码分析（完整版）

Hadoop源码分析视频下载

Hadoop源代码分析(完整版).pdf

hadoop源码分析

Hadoop源码分析

hadoop 源码解析_yarn源码解析

hadoop-2.7.1：Hadoop源码分析

hadoop 源码分析 文档

hadoop 源码分析全

YOLO算法-城市电杆数据集-496张图像带标签-电杆.zip

最新资源

hadoop 源码分析文档