HDFS--Hadoop分布式文件系统

HDFS是什么

HDFS设计特性和概念

HDFS，全称是Hadoop Distributed Filesystem，是一个分布式的文件系统，以流式数据访问模式来存储超大文件（一次写入、多次读取）。

HDFS具有如下设计特性：

（1）处理超大文件，指的是GB、TB、PB级别的文件。百度、淘宝都有PB级别的HDFS，百度应该有国内最大规模的HDFS，几十PB。

（2）流式数据访问，一次写入，多次读取，所处理的场景中，读取整个数据的延迟比读取第一条记录的时间延迟重要。

（3）运行在普通商用PC即可，比如3万级别的普通PC服务器（16-32G ECC内存，8-16核CPU）。

（4）是为高数据吞吐量优化的，以高时间延迟为代价。

（5）推荐处理大量小文件，由于namenode将文件系统的元数据存储在内存中，故文件总数受制于namenode节点内存。根据经验，一个文件/目录/block大约占用150自己，所以亿级别文件还可以，10亿级别内存就不够了。

（6）对于写入，只能有一个写入操作，也只能把内容添加在文件的末尾。

概念：

（1）数据块（block），默认64M，一般用128M，相对于文件系统块（几K字节大小）、磁盘块（一般512毕节），HDFS的块设计明显大的多，这是为了最小为寻址开销（寻址占传输的百分比，比如：寻址10S，传输100MB/S，则寻址时间仅占传输时间的1%）。

（2）名称节点（namenode），是管理者，维护整个HDFS的文件系统树及树内所有的文件和目录。

（3）数据节点（datanode），是文件系统工作节点，根据namenode调度，存储并检索数据块，定期向namenode发送它所存储的块列表。

namenode单点风险的2种解决办法：

（1）备份那些组成文件系统元数据持久状态的文件，比如，持久状态写入本地磁盘的同时，写入一个远程的文件系统。

（2）运行一个辅助namecode，由于辅助namecode的滞后性，所以namecode损坏时，难免会丢失部分数据。

Hadoop抽象文件系统

Hadoop有一个抽象文件系统，由org.apache.hadoop.fs.FileSystem定义，HDSF只是其中的一个实现。

Hadoop所实现的文件系统列表，大致如下图所示：



Hadoop对文件系统提供了许多接口，它一般使用URI方案来选取合适的文件系统实例进行交互。，比如如下代码：

String uri = "hdfs:///test/input/t/temperature.txt";

Configuration conf = new Configuration();

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

即时根据hdsfs://来判断，使用hdfs.DistributedFileSystem；如果改成file://，则使用fs.LocalFileSystem。

HDFS命令行接口

查看所有命令：

hadoop fs -help

hadoop fsck -help

也可以通过Web界面浏览文件系统：http://192.168.1.10:50070/

HDFS JAVA API

从Hadoop URL中读取数据

[java]
view plaincopyprint?





import java.io.InputStream;
import java.net.URL;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
import org.apache.zookeeper.common.IOUtils;

public class URLCat {
static {
URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
}

public static void main(String[] args) throws Exception {
InputStream in = null;
try {
in = new URL("hdfs:///test/input/t/temperature.txt").openStream();
IOUtils.copyBytes(in, System.out, 4096, false);
} finally {
IOUtils.closeStream(in);
}
}
}

把hdfs换成file，则可以直接处理当前本地文件系统。
以上的方法是很简单的，利用的java.net.URL对象打开数据流，从中读取数据。但是这个方法有个限制，Java虚拟机只能调用这个set方法一次，这个限制意味着如果有其他不受控制的第三方组件（已经声明了URLStreamHandlerFactory实例），则我们无法再使用这种方法读取数据，因而，不推荐使用。

通过FileSystem API读取数据

[java]
view plaincopyprint?





import java.io.InputStream;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.zookeeper.common.IOUtils;

public class FileSystemCat {
public static void main(String[] args) throws Exception {
String uri = "hdfs:///test/input/t/temperature.txt";
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
InputStream in = null;
try {
in = fs.open(new Path(uri));
IOUtils.copyBytes(in, System.out, 4096, false);
} finally {
IOUtils.closeStream(in);
}
}
}

实际上，open方法返回的是FSDataInputStream对象，是继承java.io.DataInputStream的一个特殊类，支持随机访问，由此可以从流的任意位置读取数据，比如，我们把try段的代码变成：

[java]
view plaincopyprint?





in = fs.open(new Path(uri));
IOUtils.copyBytes(in, System.out, 4096, false);
((FSDataInputStream) in).seek(0); //go back to the start of the file
IOUtils.copyBytes(in, System.out, 4096, false)

则，会显示两遍文件temperature.txt文件的内容。

将本地文件复制到Hadoop文件系统

[java]
view plaincopyprint?





import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

public class FileCopyWithProgress {

public static void main(String[] args) throws Exception
{
String localSrc = "/home/hadoop/temperature.txt";
String dst = "hdfs:///test/input/t/temperature2.txt";
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 void progress(){
System.out.println(".");
}
});
IOUtils.copyBytes(in,out,4096,true);
}
}

每次Hadoop调用progress()方法时，也就是每次讲64KB数据包写入datanode后。

列出文件

[java]
view plaincopyprint?





import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;

public class ListStatus {

public static void main(String[] args) throws Exception {
String uri = "hdfs:///test/input/t/";
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
// 显示一组路径的目录列表的并集
/*
* Path[] paths = new Path[] { new Path("hdfs:///test/input/t/"), new
* Path("hdfs:///test/input/wc") }; FileStatus[] status =
* fs.listStatus(paths);
*/
// 通配方式
FileStatus[] status = fs.globStatus(new Path(
"hdfs:///test/input/wc/*02.txt"));

Path[] listedPaths = FileUtil.stat2Paths(status);
for (Path p : listedPaths) {
System.out.println(p);
}
}
}

文件元数据FileStatus，封装了文件系统中文件和目录的元数据，包括文件长度、块大小、备份、修改时间、所有者以及权限信息。

删除数据

使用FileSystem的delete()方法可以永久性删除文件或目录。

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

如果f是一个文件或空目录，那么recursive的值就会被忽略。如果f是一个非空目录，则只有recursive为true才能删除，否则会抛出IOException。

Hadoop数据流

文件读取剖析

文件写入剖析

复本的布局策略（以3个为例）：1、运行客户端的节点，2、离架节点，3、2所在机架的随机节点。

一致模型

文件系统的一致模型，描述了对文件读/写的数据可见性。HDFS为性能牺牲了一些POSIX要求，因此一些操作与你期望的不同。

[java]
view plaincopyprint?





Path p = new Path("p");
Fs.create(p);

[java]
view plaincopyprint?





Path p = new Path("p");
OutputStream out = fs.create(p);
out.write("content".getBytes("UTF-8"));
out.flush();

以上两段代码都不能这个文件在文件系统立即可见，除非在out.flush();后面增加一行out.sync()，目的是强制所有的缓存与数据节点同步，另外，在HDFS中关闭文件out.close(0其实隐含执行了sync()方法。
这个一致模型对应用设计的重要性：

如果不调用sync()，可能因客户端故障而丢失数据，而经常调用sync()也会有额外性能开销，所以需要在数据健壮性和吞吐量直接有所取舍，这与具体的应用有关，通过设置不同的调用sync()的频率来衡量应用的性能，最终找到一个合适的频率。

通过distcp进行并行复制

前面介绍的都是单线程的HDFS访问模型，distcp是一个分布式的复制程序。典型应用是在两个HDFS直接传输 数据，如果两个集群运行相同版本的Hadoop，则可以如下：

hadoop distcp hdfs://集群1的某节点/foo hdfs://集群2的某节点/foo

可以通过-overrite，指定覆盖现有的文件；通过-update指定仅更新修改过的文件。

如果两个集群版本不一样，可以如下这样：

hadoop distcp hftp://集群1的某节点:50070/foo hdfs://集群2的某节点/foo

默认情况下，每个集群节点（tasktracker），最多分配20个map任务，如果复制1000G数据到100个节点的集群，一共会有2000个map任务，每个map任务平均分配512M数据；可以指定-m参数，减少map任务数，比如-m 1000，将分配1000个map任务，平均每个复制1GB数据；但是一般不推荐这么做，可能导致集群不平衡。

Hadoop存档

Hadoop存档文件（HAR文件），是一个高效的文件存档工具，它将文件存入HDFS块，减少namenode内存使用的同时，依然允许对文件进行透明的访问（即Hadoop文档可以作为MapReduce的输入）。

命令演示如下：

hadoop fs -ls -R /test/input

hadoop archive -archiveName files.har -p /test input /test/file

hadoop fs -ls /test/file

hadoop fs -ls /test/file/files.har

hadoop fs -ls -R har:///test/file/files.har

hadoop fs -rm -r /test/file

har文件的不足：

（1）创建一个存档文件会创建原始文件的一个复本，因此需要额外的和原始文件一样大小的磁盘空间。当然，创建了存档文件，可以删除原始文件。har是不压缩的，非常类似于tar文件

（2）一旦创建，存档文件不能修改。事实上，一般不会修改存档文件，因为它们是定期成批存档的，比如每日或每周。

（3）Har文件作为mapreduce输入时，InputFormat类并不知道文件已经存档，尽管该类可以将多个文件打包成一个MapReduce分片，所以即使在har文件中处理许多小文件，依然和原来一样低效。