Hadoop 分布式文件系统

作者: 姜和然 | 来源:发表于2019-08-14 14:59 被阅读0次

HDFS的设计目标

Hadoop分布式文件系统(HDFS)被设计成适合运行在通用硬件(commodity hardware)上的分布式文件系统。它和现有的分布式文件系统有很多共同点。但同时，它和其他的分布式文件系统的区别也是很明显的。HDFS是一个高度容错性的系统，适合部署在廉价的机器上。HDFS能提供高吞吐量的数据访问，非常适合大规模数据集上的应用。HDFS放宽了一部分POSIX约束，来实现流式读取文件系统数据的目的。HDFS在最开始是作为Apache Nutch搜索引擎项目的基础架构而开发的。

超大文件

这里非常大指的是几百M、G、或者TB级别。实际应用中已有很多集群存储的数据达到PB级别。

流式数据访问

HDFS构建思路是这样的：最有效的数据处理模式是一次写入、多次读取。数据集经常从数据源生成或者从数据源复制而来，接着长时间在此数据集上进行各种分析。然后每次分析工作经常读取其中的大部分数据甚至是全部。因此读取整个数据集所需时间比读取第一条记录的延时更重要。

商用硬件

hadoop设计运行在商用硬件的集群上的。因此至少对于庞大的集群来说节点故障的几率还是非常高的。

有些场景不适合采用hdfs存储数据

低时间延迟的数据访问

hdfs不适合要求低时间延迟的数据访问应用，如几十毫秒。记住，HDFS是未该数据吞吐量应用优化的，这可能会以提高时间延迟为代价。

大量的小文件

文件的元数据（如目录结构，文件block的节点列表，block-node mapping）保存在NameNode的内存中，整个文件系统的文件数量会受限于NameNode的内存大小。经验而言，一个文件/目录/文件块一般占有150字节的元数据内存空间。如果有100万个文件，每个文件占用1个文件块，则需要大约300M的内存。因此十亿级别的文件数量在现有商用机器上难以支持。

多用户写入，任意修改文件

HDFS中的文件写入只支持单个写入，并且写操作也总是以追加的方式。

HDFS的概念

数据块

HDFS的数据块大小为128Mb。HDFS上的文件被划分为块大小的多个分块，作为独立的存储单元。比Block小的文件不会占用整个Block，只会占据实际大小。

HDFS的Block为什么这么大？ 是为了最小化查找（seek）时间，控制定位文件与传输文件所用的时间比例。

Block抽象的好处

1、block的拆分使得单个文件大小可以大于整个磁盘的容量，构成文件的Block可以分布在整个集群，理论上，单个文件可以占据集群中所有机器的磁盘。

2、使用抽象块而非整个文件作为存储单元，简化了存储系统的设计。对于Block，无需关注其权限，所有者等内容（这些内容都在文件级别上进行控制）。

3、Block非常适合用于数据备份进而提供数据容错能力和挺高可用性。

Namenode & Datanode

HDFS具有主/从架构。HDFS集群由单个NameNode，一个管理文件系统命名空间的主服务器和管理客户端对文件的访问组成。此外，还有许多DataNode，通常是群集中每个节点一个，用于管理连接到它们运行的节点的存储。HDFS公开文件系统命名空间，并允许用户数据存储在文件中。在内部，文件被分成一个或多个块，这些块存储在一组DataNode中。NameNode执行文件系统命名空间操作，如打开，关闭和重命名文件和目录。它还确定了块到DataNode的映射。DataNode负责提供来自文件系统客户端的读写请求。DataNode还根据NameNode的指令执行块创建，删除和复制。

块缓存

DataNode通常直接从磁盘读取数据，但是对于访问频繁的文件，其对应的块可能被显式地缓存在datanode的内存中，以堆外块缓存的形式存在。默认情况下，一个Block只有一个数据节点会缓存。但是可以针对每个文件可以个性化配置。作业调度器可以利用缓存提升性能，例如MapReduce可以把任务运行在有Block缓存的节点上。用户或者应用可以向NameNode发送缓存指令（缓存哪个文件，缓存多久），缓存池的概念用于管理一组缓存的权限和资源。

联邦HDFS

我们知道NameNode的内存会制约文件数量，HDFS Federation提供了一种横向扩展NameNode的方式。在Federation模式中，每个NameNode管理命名空间的一部分，例如一个NameNode管理/user目录下的文件，另一个NameNode管理/share目录下的文件。

每个NameNode管理一个namespace volumn，所有volumn构成文件系统的元数据。每个NameNode同时维护一个Block Pool，保存Block的节点映射等信息。各NameNode之间是独立的，一个节点的失败不会导致其他节点管理的文件不可用。

客户端使用mount table将文件路径映射到NameNode。mount table是在Namenode群组之上封装了一层，这一层也是一个Hadoop文件系统的实现，通过viewfs:协议访问。

HDFS的高可用性

在HDFS集群中，NameNode依然是单点故障（SPOF）。元数据同时写到多个文件系统以及Second NameNode定期checkpoint有利于保护数据丢失，但是并不能提高可用性。

HDFS的解决方式是采用HA的HDFS集群配置两个NameNode，分别处于Active和Standby状态。当Active NameNode故障之后，Standby接过责任继续提供服务，用户没有明显的中断感觉。一般耗时在几十秒到数分钟。

HA涉及到的主要实现逻辑有

1）主备需共享edit log存储。

主NameNode和待命的NameNode共享一份edit log，当主备切换时，Standby通过回放edit log同步数据。

共享存储通常有2种选择：

NFS：传统的网络文件系统

QJM：quorum journal manager

QJM是专门为HDFS的HA实现而设计的，用来提供高可用的edit log。QJM运行一组journal node，edit log必须写到大部分的journal nodes。通常使用3个节点，因此允许一个节点失败，类似ZooKeeper。注意QJM没有使用ZK，虽然HDFS HA的确使用了ZK来选举主Namenode。一般推荐使用QJM。

2）DataNode需要同时往主备发送Block Report

因为Block映射数据存储在内存中（不是在磁盘上），为了在Active NameNode挂掉之后，新的NameNode能够快速启动，不需要等待来自Datanode的Block Report，DataNode需要同时向主备两个NameNode发送Block Report。

3）客户端需要配置failover模式（对用户透明）

Namenode的切换对客户端来说是无感知的，通过客户端库来实现。客户端在配置文件中使用的HDFS URI是逻辑路径，映射到一对Namenode地址。客户端会不断尝试每一个Namenode地址直到成功。

4）Standby替代Secondary NameNode

如果没有启用HA，HDFS独立运行一个守护进程作为Secondary Namenode。定期checkpoint，合并镜像文件和edit日志。

命令行接口

HDFS提供了各种交互方式，例如通过Java API、HTTP、shell命令行的。命令行的交互主要通过hadoop fs来操作。例如：

1、hadoop fs -copyFromLocal // 从本地复制文件到HDFS

2、hadoop fs mkdir // 创建目录

3、hadoop fs -ls // 列出文件列表

Hadoop中，文件和目录的权限类似于POSIX模型，包括读、写、执行3种权限：

读权限（r）：用于读取文件或者列出目录中的内容

写权限（w）：对于文件，就是文件的写权限。目录的写权限指在该目录下创建或者删除文件（目录）的权限。

执行权限（x）：文件没有所谓的执行权限，被忽略。对于目录，执行权限用于访问器目录下的内容。

每个文件或目录都有owner，group，mode三个属性，owner指文件的所有者，group为权限组。mode

由所有者权限、文件所属的组中组员的权限、非所有者非组员的权限组成。下图表示其所有者root拥有读写权限，supergroup组的组员有读权限，其他人有读权限。

文件权限是否开启通过dfs.permissions.enabled属性来控制，这个属性默认为false，没有打开安全限制，因此不会对客户端做授权校验，如果开启安全限制，会对操作文件的用户做权限校验。特殊用户superuser是Namenode进程的标识，不会针对该用户做权限校验。

Hadoop文件系统

Hadoop有一个抽象的文件系统概念，HDFS只是其中的一种实现。Hadoop提供的实现如下图：

hadoop文件系统

接口

Hadoop对于文件系统提供了许多的接口，主要为HTTP、C语言、NFS 和 java。

HTTP

WebHDFS和SWebHDFS协议将文件系统暴露HTTP操作，这种交互方式比原生的Jav客户端慢，不适合操作大文件。通过HTTP，有2种访问方式，直接访问和通过代理访问。

直接访问

直接访问的示意图如下：

Namenode和Datanode默认打开了嵌入式web server，即dfs.webhdfs.enabled默认为true。webhdfs通过这些服务器来交互。元数据的操作通过namenode完成，文件的读写首先发到namenode，然后重定向到datanode读取（写入）实际的数据流。

通过HDFS代理

采用代理的示意图如上所示。使用代理的好处是可以通过代理实现负载均衡或者对带宽进行限制，或者防火墙设置。代理通过HTTP或者HTTPS暴露为WebHDFS，对应为webhdfs和swebhdfs URL Schema。

代理作为独立的守护进程，独立于namenode和datanode，使用httpfs.sh脚本，默认运行在14000端口

C语言

Hadoop提供了一个名为libhdfs的C语言库，改语言库时java FileSystem接口类的一个镜像。它使用java的原生接口调用java文件系统客户端。它与java的api非常相似，但是滞后于java API。

NFS

使用Hadoop的NFSv3网关将HDFS挂载为本地客户端的文件系统是可行的。然后你可以使用Unix实用程序如（ls 和 cat）与改文件系统交互，上传文件，通过任意一种编程语言调用POSIX库来访问文件系统。

java 接口

实际的应用中，对HDFS的大多数操作还是通过FileSystem来操作，这部分重点介绍一下相关的接口，主要关注HDFS的实现类DistributedFileSystem及相关类。

读操作

可以使用URL来读取数据，或者而直接使用FileSystem操作。

从Hadoop URL读取数据

java.net.URL类提供了资源定位的统一抽象，任何人都可以自己定义一种URL Schema，并提供相应的处理类来进行实际的操作。hdfs schema便是这样的一种实现。

InputStream in = null;

try {

in = new URL("hdfs://master/user/hadoop").openStream();

}finally{

IOUtils.closeStream(in);

}

为了使用自定义的Schema，需要设置URLStreamHandlerFactory，这个操作一个JVM只能进行一次，多次操作会导致不可用，通常在静态块中完成。下面的截图是一个使用示例：

运行结果

使用FileSystem API读取数据

1）首先获取FileSystem实例，一般使用静态get工厂方法

public static FileSystem get(Configuration conf) throws IOException

public static FileSystem get(URI uri , Configuration conf) throws IOException

public static FileSystem get(URI uri , Configuration conf，String user) throws IOException

如果是本地文件，通过getLocal获取本地文件系统对象：

public static LocalFileSystem getLocal(COnfiguration conf) thrown IOException

2）调用FileSystem的open方法获取一个输入流:

public FSDataInputStream open(Path f) throws IOException

public abstarct FSDataInputStream open(Path f , int bufferSize) throws IOException

默认情况下，open使用4KB的Buffer，可以根据需要自行设置。

3）使用FSDataInputStream进行数据操作

FSDataInputStream是java.io.DataInputStream的特殊实现，在其基础上增加了随机读取、部分读取的能力

public class FSDataInputStream extends DataInputStream

implements Seekable, PositionedReadable,

ByteBufferReadable, HasFileDescriptor, CanSetDropBehind, CanSetReadahead,

HasEnhancedByteBufferAccess

随机读取操作通过Seekable接口定义：

public interface Seekable {

void seek(long pos) throws IOException;

long getPos() throws IOException;

}

seek操作开销昂贵，慎用。

部分读取通过PositionedReadable接口定义：

public interface PositionedReadable{

public int read(long pistion ,byte[] buffer,int offser , int length) throws IOException;

public int readFully(long pistion ,byte[] buffer,int offser , int length) throws IOException;

public int readFully(long pistion ,byte[] buffer) throws IOException;

}

写数据

在HDFS中，文件使用FileSystem类的create方法及其重载形式来创建，create方法返回一个输出流FSDataOutputStream，可以调用返回输出流的getPos方法查看当前文件的位移，但是不能进行seek操作，HDFS仅支持追加操作。

创建时，可以传递一个回调接口Peofressable，获取进度信息

append(Path f)方法用于追加内容到已有文件，但是并不是所有的实现都提供该方法，例如Amazon的文件实现就没有提供追加功能。

下面是一个例子：

String localSrc = args[0];

String dst = args[1];

InputStream in = new BufferedInputStream(new FileInputStream(localSrc));

COnfiguration conf = new Configuration();

FileSystem fs = FileSystem.get(URI.create(dst),conf);

OutputStream out = fs.create(new Path(dst), new Progressable(){

public vid progress(){

System.out.print(.);

}

});

IOUtils.copyBytes(in , out, 4096,true);

目录操作

使用mkdirs（）方法,会自动创建没有的上级目录

HDFS中元数据封装在FileStatus类中，包括长度、block size，replicaions，修改时间、所有者、权限等信息。使用FileSystem提供的getFileStatus方法获取FileStatus。exists()方法判断文件或者目录是否存在；

列出文件（list），则使用listStatus方法，可以查看文件或者目录的信息

public abstract FileStatus[] listStatus(Path f) throws FileNotFoundException,

IOException;

Path是个文件的时候，返回长度为1的数组。FileUtil提供的stat2Paths方法用于将FileStatus转化为Path对象。

globStatus则使用通配符对文件路径进行匹配：

public FileStatus[] globStatus(Path pathPattern) throws IOException

PathFilter用于自定义文件名过滤，不能根据文件属性进行过滤，类似于java.io.FileFilter。例如下面这个例子排除到给定正则表达式的文件：

public interfacePathFilter{

boolean accept(Path path);

}

删除数据

使用FileSystem的delete()方法

public boolean delete(Path f , boolean recursive) throws IOException;

recursive参数在f是个文件的时候被忽略。如果f是文件并且recursice为true，则删除整个目录，否则抛出异常。

数据流

接下来详细介绍HDFS读写数据的流程，以及一致性模型相关的一些概念。

读文件

大致读文件的流程如下：

客户端读取HDFS中的数据

1）客户端传递一个文件Path给FileSystem的open方法

2）DFS采用RPC远程获取文件最开始的几个block的datanode地址。Namenode会根据网络拓扑结构决定返回哪些节点（前提是节点有block副本），如果客户端本身是Datanode并且节点上刚好有block副本，直接从本地读取。

3）客户端使用open方法返回的FSDataInputStream对象读取数据（调用read方法）

4）DFSInputStream（FSDataInputStream实现了改类）连接持有第一个block的、最近的节点，反复调用read方法读取数据

5）第一个block读取完毕之后，寻找下一个block的最佳datanode，读取数据。如果有必要，DFSInputStream会联系Namenode获取下一批Block 的节点信息(存放于内存，不持久化），这些寻址过程对客户端都是不可见的。

6）数据读取完毕，客户端调用close方法关闭流对象

在读数据过程中，如果与Datanode的通信发生错误，DFSInputStream对象会尝试从下一个最佳节点读取数据，并且记住该失败节点，后续Block的读取不会再连接该节点

读取一个Block之后，DFSInputStram会进行检验和验证，如果Block损坏，尝试从其他节点读取数据，并且将损坏的block汇报给Namenode。

客户端连接哪个datanode获取数据，是由namenode来指导的，这样可以支持大量并发的客户端请求，namenode尽可能将流量均匀分布到整个集群。

Block的位置信息是存储在namenode的内存中，因此相应位置请求非常高效，不会成为瓶颈。

写文件

客户端将数据写入HDFS

步骤分解

1）客户端调用DistributedFileSystem的create方法

2）DistributedFileSystem远程RPC调用Namenode在文件系统的命名空间中创建一个新文件，此时该文件没有关联到任何block。这个过程中，Namenode会做很多校验工作，例如是否已经存在同名文件，是否有权限，如果验证通过，返回一个FSDataOutputStream对象。如果验证不通过，抛出异常到客户端。

3）客户端写入数据的时候，DFSOutputStream分解为packets，并写入到一个数据队列中，该队列由DataStreamer消费。

4）DateStreamer负责请求Namenode分配新的block存放的数据节点。这些节点存放同一个Block的副本，构成一个管道。 DataStreamer将packer写入到管道的第一个节点，第一个节点存放好packer之后，转发给下一个节点，下一个节点存放之后继续往下传递。

5）DFSOutputStream同时维护一个ack queue队列，等待来自datanode确认消息。当管道上的所有datanode都确认之后，packer从ack队列中移除。

6）数据写入完毕，客户端close输出流。将所有的packet刷新到管道中，然后安心等待来自datanode的确认消息。全部得到确认之后告知Namenode文件是完整的。 Namenode此时已经知道文件的所有Block信息（因为DataStreamer是请求Namenode分配block的），只需等待达到最小副本数要求，然后返回成功信息给客户端。

Namenode如何决定副本存在哪个Datanode？

HDFS的副本的存放策略是可靠性、写带宽、读带宽之间的权衡。默认策略如下：

第一个副本放在客户端相同的机器上，如果机器在集群之外，随机选择一个（但是会尽可能选择容量不是太慢或者当前操作太繁忙的）

第二个副本随机放在不同于第一个副本的机架上。

第三个副本放在跟第二个副本同一机架上，但是不同的节点上，满足条件的节点中随机选择。

更多的副本在整个集群上随机选择，虽然会尽量便面太多副本在同一机架上。

副本的位置确定之后，在建立写入管道的时候，会考虑网络拓扑结构。下面是可能的一个存放策略：

一个典型的复本管线

这样选择很好滴平衡了可靠性、读写性能

可靠性：Block分布在两个机架上

写带宽：写入管道的过程只需要跨越一个交换机

读带宽：可以从两个机架中任选一个读取

一致性模型

一致性模型描述文件系统中读写操纵的可见性。HDFS中，文件一旦创建之后，在文件系统的命名空间中可见：

Path p = new Path("p");

fs.create(p);

assertTaht(fs.exists(p),is(true));

但是任何被写入到文件的内容不保证可见，即使对象流已经被刷新。

Path p = new Path(“p”);

OutputStream out = fs.create(p);

out.write(“content”.getBytes(“UTF-8”));

out.flush();

assertTaht(fs.getFileStatus(p).getLen,0L); // 为0，即使调用了flush

如果需要强制刷新数据到Datanode，使用FSDataOutputStream的hflush方法强制将缓冲刷到datanode

hflush之后，HDFS保证到这个时间点为止写入到文件的数据都到达所有的数据节点。

Path p = new Path("p");

OutputStream out = fs.create(p);

out.write("content".getBytes("UTF-8"));

out.flush();

assertTaht(fs.getFileStatus(p).getLen,is(((long,"content".length())));

关闭对象流时，内部会调用hflush方法,但是hflush不保证datanode数据已经写入到磁盘，只是保证写入到datanode的内存，因此在机器断电的时候可能导致数据丢失，如果要保证写入磁盘，使用hsync方法，hsync类型与fsync（）的系统调用，fsync提交某个文件句柄的缓冲数据。

FileOutputStreamout = new FileOutPutStream(localFile);

out.write("content".getBytes("UTF-8"));

out.flush();

out.getFD().sync();

assertTaht(localFile.getLen,is(((long,"content".length())));

使用hflush或hsync会导致吞吐量下降，因此设计应用时，需要在吞吐量以及数据的健壮性之间做权衡。

另外，文件写入过程中，当前正在写入的Block对其他Reader不可见。

Hadoop节点距离

在读取和写入的过程中，namenode在分配Datanode的时候，会考虑节点之间的距离。HDFS中，距离没有采用带宽来衡量，因为实际中很难准确度量两台机器之间的带宽。

Hadoop把机器之间的拓扑结构组织成树结构，并且用到达公共父节点所需跳转数之和作为距离。事实上这是一个距离矩阵的例子。下面的例子简明地说明了距离的计算：

网格距离

Hadoop集群的拓扑结构需要手动配置，如果没配置，Hadoop默认所有节点位于同一个数据中心的同一机架上。

通过distcp并行复制

前面的关注点都在于单线程的访问，如果需要并行处理文件，需要自己编写应用。Hadoop提供的distcp工具用于并行导入数据到Hadoop或者从Hadoop导出。一些例子：

hadoop distcp file1 file2 //可以作为fs -cp命令的高效替代

hadoop distcp dir1 dir2

hadoop distcp -update dir1 dir2 #update参数表示只同步被更新的文件，其他保持不变

distcp是底层使用MapReduce实现，只有map实现，没有reduce。在map中并行复制文件。 distcp尽可能在map之间平均分配文件。map的数量可以通过-m参数指定:

hadoop distcp -update -delete -p hdfs://master1:9000/foo hdfs://master2/foo

这样的操作常用于在两个集群之间复制数据，update参数表示只同步被更新过的数据，delete会删除目标目录中存在，但是源目录不存在的文件。p参数表示保留文件的全校、block大小、副本数量等属性。

如果两个集群的Hadoop版本不兼容，可以使用webhdfs协议：

hadoop distcp webhdfs: //namenode1: 50070/foo webhdfs: //namenode2: 50070/foo

原文链接：https://blog.csdn.net/bingduanlbd/article/details/51914550