Hadoop源代码深度解析：分布式云计算核心技术剖析

4星 · 超过85%的资源需积分: 10 90 浏览量更新于2024-07-23 收藏 6.06MB PDF 举报

Hadoop源代码分析完整版深入探讨了Apache Hadoop这一分布式云计算领域的核心技术，特别关注的是其核心组件Hadoop Distributed File System (HDFS) 和 MapReduce 的工作原理与实现。Hadoop最初源于Google的五个关键项目——Chubby、GFS、BigTable和MapReduce，Apache社区将其开源并发展为Hadoop生态系统，其中Chubby对应ZooKeeper，GFS演变成HDFS，BigTable演化为HBase，MapReduce则是Hadoop的主要计算模型。 HDFS作为分布式文件系统，是Hadoop架构的基础，它解决了大数据存储和访问的问题。HDFS的设计目标是高容错性、高吞吐量以及数据的备份和恢复能力。HDFS通过将大文件分割成多个小块，并在多台机器上进行分布式存储，实现了数据的冗余和负载均衡。这种设计使得Hadoop能够处理PB级别的数据，非常适合大规模数据处理任务。 Hadoop的包结构复杂，反映出其模块化的开发思路。例如，conf包负责读取系统配置，依赖于fs包以访问文件系统，这表明了不同模块间的紧密协作。而fs包又封装了文件系统功能的抽象，使得代码能更好地隔离底层细节。蓝色部分代表Hadoop的核心组件，如HDFS的FsShell、Hadoop Distributed FileSystem (HDFS) Client、Hadoop Common等，这些都是后续分析的重点。 MapReduce是Hadoop的另一个核心组件，它是一种编程模型，允许开发者编写处理大规模数据集的应用程序。MapReduce将复杂的计算任务分解为一系列简单的Map和Reduce操作，分别执行在不同的节点上，从而实现并行化处理。这个模型简化了程序员编写分布式应用程序的复杂性，使其易于理解和维护。 Hadoop的源代码分析深入到了具体的包依赖和接口设计，这对于理解Hadoop的工作流程，优化性能，以及进行扩展和定制具有重要意义。通过学习和分析Hadoop源代码，开发者可以更好地掌握分布式计算的基本原理，提高在实际项目中的应用能力。同时，随着大数据和云计算的发展，掌握Hadoop技术对于数据科学家和工程师来说是至关重要的技能。

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 上。

Ha doop

源代码分析（一五）

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

首先，

proxyReply

和

不一样，这是因为发起请求的节点和提供数据的节点并不是同一个。写数据块发起请求方也提供数据，替换数据块请求方

不提供数据，而是提供了一个数据源（

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，因为我们知道 BlockSender 中，发送数据的同时，会对数据进行校验。verifyBlock 只需要读一个

Block 到一个空输出设备（NullOutputStream），如果有异常，那么校验失败，如果正常，校验成功。

剩余81页未读，继续阅读

hugejiletuhugejiltu

粉丝: 5
资源: 111

Hadoop源代码深度解析：分布式云计算核心技术剖析

搭建Hadoop运行环境

hadoop@ubuntu:/usr/local/hadoop$ ./bin/hadoop version ERROR: Cannot execute /usr/local/hadoop-3.1.3/libexec/hadoop-config.sh.

Apache Hadoop下载

hadoop伪分布式安装启动hdfs

Hadoop如何找回VERSION文件

hadoop3.1.3下载

Hadoop Submarine

cdh6.3.2源码下载

怎么创建运行上述程序的Hadoop Job程序，具体完整的且能够跑通的代码是什么？

Linux进hive

最新资源