Hadoop 入门笔记十六 : HDFS核心源码解析 hadoop

一. HDFS客户端核心代码 1. Configuration
Configuration提供对配置参数的访问，通常称之为配置文件类。主要用于加载或者设定程序运行时相关的参数属性。
1. Configuration加载默认配置首先加载了静态方法和静态代码块，其中在静态代码块中显示默认加载了两个配置文件：
core-default.xml以及core-site.xml

文章图片

2. Configuration 加载用户设置通过conf.set设置的属性也会被加载。

文章图片

文章图片

1. FileSystem
FileSystem类是一个通用的文件系统的抽象基类。具体来说它可以实现为一个分布式的文件系统，也可以实现为一个本地文件系统。所有的可能会使用到HDFS的用户代码在进行编写时都应该使用FileSystem对象。
代表本地文件系统的实现是LocalFileSystem，代表分布式文件系统的实现是DistributedFileSystem。当然针对其他hadoop支持的文件系统也有不同的具体实现。
因此HDFS客户端在进行读写操作之前，需要创建FileSystem对象的实例。
1. 获取FileSystem 实例
文章图片

FileSystem对象是通过调用getInternal方法得到的。

文章图片

首先在getInternal方法中调用了createFileSystem方法，进去该方法：
FileSystem实例是通过反射的方式获得的，具体实现是通过调用反射工具类ReflectionUtils的newInstance方法并将class对象以及Configuration对象作为参数传入最终得到了FileSystem实例。

文章图片

二. HDFS通信协议 1. 简介
HDFS作为一个分布式文件系统，它的某些流程是非常复杂的（例如读、写文件等典型流程），常常涉及数据节点、名字节点和客户端三者之间的配合、相互调用才能实现。为了降低节点间代码的耦合性，提高单个节点代码的内聚性， HDFS将这些节点间的调用抽象成不同的接口。
HDFS节点间的接口主要有两种类型：
Hadoop RPC接口：基于Hadoop RPC框架实现的接口；
流式接口：基于TCP或者HTTP实现的接口；
2. Hadoop RPC 接口
1. RPC 介绍 RPC 全称 Remote Procedure Call——远程过程调用。就是为了解决远程调用服务的一种技术，使得调用者像调用本地服务一样方便透明。

文章图片

通信模块: 传输RPC请求和响应的网络通信模块，可以基于TCP协议，也可以基于UDP协议，可以是同步，也可以是异步的。
客户端Stub程序：服务器和客户端都包括Stub程序。在客户端，Stub程序表现的就像本地程序一样，但底层却会将调用请求和参数序列化并通过通信模块发送给服务器。之后Stub程序等待服务器的响应信息，将响应信息反序列化并返回给请求程序。
服务器端Stub程序：在服务器端，Stub程序会将远程客户端发送的调用请求和参数反序列化，根据调用信息触发对应的服务程序，然后将服务程序返回的响应信息序列化并发回客户端。
请求程序：请求程序会像调用本地方法一样调用客户端Stub程序，然后接收Stub程序返回的响应信息。
服务程序：服务器会接收来自Stub程序的调用请求，执行对应的逻辑并返回执行结果。
Hadoop RPC调用使得HDFS进程能够像本地调用一样调用另一个进程中的方法，并且可以传递Java基本类型或者自定义类作为参数，同时接收返回值。如果远程进程在调用过程中出现异常，本地进程也会收到对应的异常。目前Hadoop RPC调用是基于Protobuf实现的。
Hadoop RPC接口主要定义在org.apache.hadoop.hdfs.protocol包和org.apache.hadoop.hdfs.server.protocol包中，核心的接口有：
ClientProtocol、ClientDatanodeProtocol、DatanodeProtocol。
2. ClientProtocol ClientProtocol定义了客户端与名字节点间的接口，这个接口定义的方法非常多，客户端对文件系统的所有操作都需要通过这个接口，同时客户端读、写文件等操作也需要先通过这个接口与Namenode协商之后，再进行数据块的读出和写入操作。
ClientProtocol定义了所有由客户端发起的、由Namenode响应的操作。这个接口非常大，有80多个方法，核心的是：HDFS文件读相关的操作、HDFS文件写以及追加写的相关操作。

读数据相关的方法
ClientProtocol中与客户端读取文件相关的方法主要有两个： getBlockLocations（）和reportBadBlocks()
客户端会调用ClientProtocol.getBlockLocations）方法获取HDFS文件指定范围内所有数据块的位置信息。这个方法的参数是HDFS文件的文件名以及读取范围，返回值是文件指定范围内所有数据块的文件名以及它们的位置信息，使用LocatedBlocks对象封装。每个数据块的位置信息指的是存储这个数据块副本的所有Datanode的信息，这些Datanode会以与当前客户端的距离远近排序。客户端读取数据时，会首先调用getBlockLocations（）方法获取HDFS文件的所有数据块的位置信息，然后客户端会根据这些位置信息从数据节点读取数据块。

文章图片

客户端会调用ClientProtocol.reportBadBlocks（）方法向Namenode汇报错误的数据块。当客户端从数据节点读取数据块且发现数据块的校验和并不正确时，就会调用这个方法向Namenode汇报这个错误的数据块信息。

文章图片
写、追加数据相关方法
在HDFS客户端操作中最重要的一部分就是写入一个新的HDFS文件，或者打开一个已有的HDFS文件并执行追加写操作。ClientProtocol中定义了8个方法支持HDFS文件的写操作： create（）、 append（）、 addBlock（）、 complete（）， abandonBlockO），getAddtionnalDatanodes（）、updateBlockForPipeline()和updatePipeline()。

create（）方法用于在HDFS的文件系统目录树中创建一个新的空文件，创建的路径由src参数指定。这个空文件创建后对于其他的客户端是“可读”的，但是这些客户端不能删除、重命名或者移动这个文件，直到这个文件被关闭或者租约过期。客户端写一个新的文件时，会首先调用create方法在文件系统目录树中创建一个空文件，然后调用addBlock方法获取存储文件数据的数据块的位置信息，最后客户端就可以根据位置信息建立数据流管道，向数据节点写入数据了。

文章图片

当客户端完成了整个文件的写入操作后，会调用complete（）方法通知Namenode。这个操作会提交新写入HDFS文件的所有数据块，当这些数据块的副本数量满足系统配置的最小副本系数（默认值为1），也就是该文件的所有数据块至少有一个有效副本时， complete（）方法会返回true，这时Namenode中文件的状态也会从构建中状态转换为正常状态；否则， complete会返回false，客户端就需要重复调用complete操作，直至该方法返回true

ClientDatanodeProtocol
客户端与数据节点间的接口。ClientDatanodeProtocol中定义的方法主要是用于客户端获取数据节点信息时调用，而真正的数据读写交互则是通过流式接口进行的。
ClientDatanodeProtocol中定义的接口可以分为两部分：一部分是支持HDFS文件读取操作的，例如getReplicaVisibleLength（）以及getBlockLocalPathInfo）；另一部分是支持DFSAdmin中与数据节点管理相关的命令。我们重点关注第一部分。

getReplicaVisibleLength
客户端会调用getReplicaVisibleLength（）方法从数据节点获取某个数据块副本真实的数据长度。当客户端读取一个HDFS文件时，需要获取这个文件对应的所有数据块的长度，用于建立数据块的输入流，然后读取数据。但是Namenode元数据中文件的最后一个数据块长度与Datanode实际存储的可能不一致，所以客户端在创建输入流时就需要调用getReplicaVisibleLength（）方法从Datanode获取这个数据块的真实长度。
getBlockLocalPathInfo
HDFS对于本地读取，也就是Client和保存该数据块的Datanode在同一台物理机器上时，是有很多优化的。Client会调用ClientProtocol.getBlockLocalPathInfo）方法获取指定数据块文件以及数据块校验文件在当前节点上的本地路径，然后利用这个本地路径执行本地读取操作，而不是通过流式接口执行远程读取，这样也就大大优化了读取的性能。
DatanodeProtocol
数据节点通过这个接口与名字节点通信，同时名字节点会通过这个接口中方法的返回值向数据节点下发指令。注意，这是名字节点与数据节点通信的唯一方式。这个接口非常重要，数据节点会通过这个接口向名字节点注册、汇报数据块的全量以及增量的存储情况。同时，名字节点也会通过这个接口中方法的返回值，将名字节点指令带回该数据块，根据这些指令，数据节点会执行数据块的复制、删除以及恢复操作。
可以将DatanodeProtocol定义的方法分为三种类型： Datanode启动相关、心跳相关以及数据块读写相关。
2. 基于TCP/HTTP流式接口
HDFS除了定义RPC调用接口外，还定义了流式接口，流式接口是HDFS中基于TCP或者HTTP实现的接口。在HDFS中，流式接口包括了基于TCP的DataTransferProtocol接口，以及HA架构中Active Namenode和Standby Namenode之间的HTTP接口。
1. DataTransferProtocolDataTransferProtocol是用来描述写入或者读出Datanode上数据的基于TCP的流式接口，HDFS客户端与数据节点以及数据节点与数据节点之间的数据块传输就是基于DataTransferProtocol接口实现的。HDFS没有采用Hadoop RPC来实现HDFS文件的读写功能，是因为Hadoop RPC框架的效率目前还不足以支撑超大文件的读写，而使用基于TCP的流式接口有利于批量处理数据，同时提高了数据的吞吐量。
DataTransferProtocol中最重要的方法就是readBlock（）和writeBlock（）。
readBlock：从当前Datanode读取指定的数据块
writeBlock：将指定数据块写入数据流管道（pipeLine）中。
DataTransferProtocol接口调用并没有使用Hadoop RPC框架提供的功能，而是定义了用于发送DataTransferProtocol请求的Sender类，以及用于响应DataTransferProtocol请求的Receiver类。
Sender类和Receiver类都实现了DataTransferProtocol接口。我们假设DFSClient发起了一个DataTransferProtocol.readBlock（）操作，那么DFSClient会调用Sender将这个请求序列化，并传输给远端的Receiver。远端的Receiver接收到这个请求后,会反序列化请求,然后调用代码执行读取操作。

【Hadoop 入门笔记十六 : HDFS核心源码解析】

文章图片

3. 数据写入流程分析 1. 写入流程图

文章图片

2. 写入数据代码

package cn.itcast.hdfs.write; import org.apache.commons.io.IOUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.io.FileInputStream; public class HDFSWriteDemo { public static void main(String[] args) throws Exception{ // 设置客户端用户身份：root 具备在hdfs读写权限 System.setProperty("HADOOP_USER_NAME","root"); // 创建Conf对象 Configuration conf = new Configuration(); // 设置操作的文件系统是HDFS 默认是file:/// conf.set("fs.defaultFS","hdfs://node1:8020"); // 创建FileSystem对象其是一个通用的文件系统的抽象基类 FileSystem fs = FileSystem.get(conf); // 设置文件输出的路径 Path path = new Path("/helloworld.txt"); // 调用create方法创建文件 FSDataOutputStream out = fs.create(path); // 创建本地文件输入流 FileInputStream in = new FileInputStream("D:\\datasets\\hdfs\\helloworld.txt"); // IO工具类实现流对拷贝 IOUtils.copy(in,out); // 关闭连接 fs.close(); } }

3. 写入数据流程梳理
1. 客户端请求NameNode创建 HDFS客户端通过对DistributedFileSystem对象调用create()请求创建文件。DistributedFileSystem为客户端返回FSDataOutputStream输出流对象。通过源码注释可以发现FSDataOutputStream是一个包装类，所包装的是DFSOutputStream。
可以通过create方法调用不断跟下去，可以发现最终的调用也验证了上述结论，返回的是DFSOutputStream 。

文章图片

点击进入代码DFSOutputStream dfsos = dfs.create可以发现，DFSOutputStream这个类是从DFSClient类的create方法中返回过来的

文章图片

点击进入代码DFSOutputStream dfsos = dfs.create可以发现，DFSOutputStream这个类是从DFSClient类的create方法中返回过来的。

文章图片

DFSClient类中的DFSOutputStream实例对象是通过调用DFSOutputStream类的newStreamForCreate方法产生的。

文章图片

2. Namenode 执行请求检查 DistributedFileSystem对namenode进行RPC调用，请求上传文件。namenode执行各种检查判断：目标文件是否存在、父目录是否存在、客户端是否具有创建该文件的权限。检查通过，namenode就会为创建新文件记录一条记录。否则，文件创建失败并向客户端抛出一个IOException。
3. DataStreamer类在之前的newStreamForCreate方法中，我们发现了最终返回的是out对象，并且在返回之前，调用了out对象的start方法。

文章图片

DataStreamer类是DFSOutputSteam的一个内部类，在这个类中，有一个方法叫做run方法，数据写入的关键代码就在这个run方法中实现。
4. DataStreamer写数据在客户端写入数据时，DFSOutputStream将它分成一个个数据包（packet 默认64kb）,并写入一个称之为数据队列（data queue）的内部队列。DataStreamer请求NameNode挑选出适合存储数据副本的一组DataNode。这一组DataNode采用pipeline机制做数据的发送。默认是3副本存储。
DataStreamer将数据包流式传输到pipeline的第一个datanode,该DataNode存储数据包并将它发送到pipeline的第二个DataNode。同样，第二个DataNode存储数据包并且发送给第三个（也是最后一个）DataNode。
DFSOutputStream也维护着一个内部数据包队列来等待DataNode的收到确认回执，称之为确认队列（ack queue）,收到pipeline中所有DataNode确认信息后，该数据包才会从确认队列删除。
客户端完成数据写入后，将在流上调用close()方法关闭。该操作将剩余的所有数据包写入DataNode pipeline，并在联系到NameNode告知其文件写入完成之前，等待确认。
因为namenode已经知道文件由哪些块组成（DataStream请求分配数据块），因此它仅需等待最小复制块即可成功返回。数据块最小复制是由参数dfs.namenode.replication.min指定，默认是1.
4. 数据读取流程分析 1. 读取流程图

文章图片

2. 读取数据代码

package cn.itcast.hdfs.read; import org.apache.commons.io.IOUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.io.FileInputStream; import java.io.FileOutputStream; public class HDFSReadDemo { public static void main(String[] args) throws Exception{ //设置客户端用户身份：root 具备在hdfs读写权限 System.setProperty("HADOOP_USER_NAME","root"); //创建Conf对象 Configuration conf = new Configuration(); //设置操作的文件系统是HDFS 默认是file:/// conf.set("fs.defaultFS","hdfs://node1:8020"); //创建FileSystem对象其是一个通用的文件系统的抽象基类 FileSystem fs = FileSystem.get(conf); //调用open方法读取文件 FSDataInputStream in = fs.open(new Path("/helloworld.txt")); //创建本地文件输出流 FileOutputStream out = new FileOutputStream("D:\\helloworld.txt"); //IO工具类实现流对拷贝 IOUtils.copy(in,out); //关闭连接 fs.close(); } }