3.The Hadoop Distributed File System

3. The Hadoop Distributed File System
3.1. The Design of HDFS
HDFS设计的针对对象：适合流式访问的超大文件、在使用便宜的硬件搭建的集群上运行。
HDFS不足：
低延迟数据访问（Hbase是个好选择）、小文件多的时候出现问题（HDFS将文件Meta信息存储在内存中，内存限制了可以控制的文件数量）、对文件的多个wirter进行写入或者任意位置的修改。
3.2. HDFS Concept
3.2.1. Blocks
HDFS中Block的大小默认是64M，小于块大小的的文件并不占据整个块的全部空间（一个块可能存有多个文件）。
使用Blocks的好处：
1） 可以存储大文件，一个文件的大小可以大于任何一个单块硬盘的容量
2） 把存储单元抽象成块而不是文件，简化了存储子系统：简化了数据管理、取消元数据关注
3） 能很好适应数据复制，数据复制保证系统的容错和可用性。
3.2.2. Namenodes and Datanodes
Namenode：master
Datanode：worker
Namenode管理文件系统名字空间（filesystem namespace），它维持了一个filesystem tree,所有文件的metadata和目录都在里面。信息被以两种文件的形式持久化在硬盘上，namespace image,edit log.
Hdfs提供了两种namenode的容错机制：
1) 备份存储持久化状态的文件系统元数据的文件
2) 提供secondary namenode。Secondary的主要角色是合并namespace image和edit log，防止edit log过大。但是secondary namenode的数据较master namenode的数据有所延迟，所有数据恢复以后肯定会有数据丢失
3.3. The Command-line Interface
以伪分布式为例
基本的文件系统操作：
1） 将本地数据拷贝到hdfs上
% hadoop fs -copyFromLocal input/docs/quangle.txt hdfs://localhost/user/tom/quangle.txt
hdfs://可省去，这样变成
% hadoop fs -copyFromLocal input/docs/quangle.txt /user/tom/quangle.txt
也可以使用相对路径：
% hadoop fs -copyFromLocal input/docs/quangle.txt quangle.txt
2） 将数据从hdfs上拷贝到本地硬盘并检查文件时候一致
% hadoop fs -copyToLocal quangle.txt quangle.copy.txt
% md5 input/docs/quangle.txt quangle.copy.txt
MD5 (input/docs/quangle.txt) = a16f231da6b05e2ba7a339320e7dacd9
MD5 (quangle.copy.txt) = a16f231da6b05e2ba7a339320e7dacd9

3） Hdfs文件列表
% hadoop fs -mkdir books
% hadoop fs -ls .
Found 2 items
drwxr-xr-x  -   tom  supergroup  0      2009-04-02   22:41   /user/tom/books
-rw-r--r--   1   tom  supergroup  118    2009-04-02   22:29   /user/tom/quangle.txt
第一列：文件模式（类似posix）
第二列：文件被复制的份数
第三列：文件拥有者
第四列：文件拥有者的group
第五列：文件大小，目录显示为0
第六列：文件最后修改日期
第七列：文件最后修改时间
第八列：文件的绝对路径
3.4. Hadoop Filesystems
Hadoop有一个对文件系统的抽象，HDFS只是其中的一个实现。Java的抽象类org.apache.hadoop.fs.FileSystem代表了Hadoop中的文件系统，还有其他的几种实现（48页）：

3.4.1. Interfaces
Hadoop用Java写成，所有Hadoop文件的交互都通过Java api来完成。
还有另外的与Hadoop文件系统交互的库：Thrift、C、FUSE、WebDAV等
3.5. The Java Interface
3.5.1. Reading Data from a Hadoop URL
最简单的方式是用java.net.URL对象打开一个流来读取。如下：
InputStream in = null;
try {
in = new URL("hdfs://host/path").openStream();
// process in
} finally {
IOUtils.closeStream(in);
}
这里需要进行一点额外的工作才能使得URL识别hdfs的uri。我们要使用java.net.URL的setURLStreamHandlerFactory()方法设置URLStreamHandlerFactory，这里需要传递一个FsUrlStreamHandlerFactory。这个操作对一个jvm只能使用一次，我们可以在静态块中调用。
public class URLCat {
static {
URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
}

public static void main(String[] args) throws Exception {
InputStream in = null;
try {
in = new URL(args[0]).openStream();
IOUtils.copyBytes(in, System.out, 4096, false);
} finally {
IOUtils.closeStream(in);
}
}
}
IOUtils是一个工具类，用来在finally从句中关闭流，也可以用来拷贝数据到输出流中。copyBytes方法的四个参数代表的含义分别是：拷贝的来源，去处，拷贝的字节数已经在拷贝完成之后是否关闭流。本例会有如下结果呈现：
% hadoop URLCat hdfs://localhost/user/tom/quangle.txt
On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.
3.5.2. Reading Data Using the FileSystem API
在某些情况下设置URLStreamHandlerFactory的方式并不一定回生效。在这种情况下，需要用FileSystem API来打开一个文件的输入流。
文件的位置是使用Hadoop Path呈现在Hadoop中的，与java.io中的不一样。
有两种方式获取FileSystem的实例：
public static FileSystem get(Configuration conf) throws IOException
public static FileSystem get(URI uri, Configuration conf) throws IOException
Configuration封装了client或者server的配置，这些配置从classpath中读取，比如被classpath指向的conf/core-site.xml文件.
第一个方法从默认位置（conf/core-site.xml）读取配置，第二个方法根据传入的uri查找适合的配置文件，若找不到则返回使用第一个方法，即从默认位置读取。

在获得FileSystem实例之后，我们可以调用open()方法来打开输入流：
public FSDataInputStream open(Path f) throws IOException
public abstract FSDataInputStream open(Path f, int bufferSize) throws IOException
第一个方法的参数f是文件位置，第二个方法的bufferSize就是输入流的缓冲大小。
下面的代码是使用FileSystem打开输入流的示例：
public class FileSystemCat {
public static void main(String[] args) throws Exception {
String uri = args[0];
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);
}
}
}
输出结果如下：
% hadoop FileSystemCat hdfs://localhost/user/tom/quangle.txt
On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.
3.5.2.1. FSDataInputStream
FileSystem的open方法返回了FSDataInputStream对象，而不是标准的java.io。
package org.apache.hadoop.fs;
public class FSDataInputStream extends DataInputStream
implements Seekable, PositionedReadable {
// implementation elided
}
FSDataInputStream实现了Seekable接口，这样使其具有了随机访问的能力。
下面是Seekable接口的定义。
public interface Seekable {
void seek(long pos) throws IOException;
long getPos() throws IOException;
boolean seekToNewSource(long targetPos) throws IOException;
}
seek()方法提供了从文件开始查找某一位置的能力。getPos()方法则返回当前相对于文件起始位置的偏移量。seekToNewSource方法不常用。
下面的程序将hdfs中的文件显示了两次：
public class FileSystemDoubleCat {
public static void main(String[] args) throws Exception {
String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
FSDataInputStream in = null;
try {
in = fs.open(new Path(uri));
IOUtils.copyBytes(in, System.out, 4096, false);
in.seek(0); // go back to the start of the file
IOUtils.copyBytes(in, System.out, 4096, false);
} finally {
IOUtils.closeStream(in);
}
}
}
运行结果如下：
% hadoop FileSystemDoubleCat hdfs://localhost/user/tom/quangle.txt
On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.
On the top of the Crumpetty Tree
The Quangle Wangle sat,
But his face you could not see,
On account of his Beaver Hat.
FSDataInputStream也实现了PositionedReadable接口，从而能读取指定offset开始的数据。
public interface PositionedReadable {
public int read(long position, byte[] buffer, int offset, int length) throws IOException;
public void readFully(long position, byte[] buffer, int offset, int length) throws IOException;
public void readFully(long position, byte[] buffer) throws IOException;
}
read()方法最多读取length bytes。Position是相对offset的偏移，buffer存放读取的数据。
readFully()方法读取length bytes的数据到buffer中，第二个readFully则是读取buffer.length bytes的数据到buffer中。以上的方法均不会改变offset的值。
最后，seek()是一个开销比较大的操作，注意节省使用。
3.5.3. Writing Data
创建文件的最简单方法：
public FSDataOutputStream create(Path f) throws IOException
这个方法有几个重载，实现了一些例如：文件复制多少份，buffer的大小等。
此方法将创建文件的任何不存在的上级目录。例如，要创建文件/usr/a/text.txt，但是/usr/a/这个目录不存在，这时候本方法将创建/usr/a/这个目录。
还有的重载函数传递了progressable接口，这使得我们可以获得写入过程的进度信息，progressable接口定义如下：
package org.apache.hadoop.util;
public interface Progressable {
public void progress();
}
下面的方法用于追加数据：
public FSDataOutputStream append(Path f) throws IOException

下面程序演示了如何拷贝一个本地文件到Hadoop filesystem中：
public class FileCopyWithProgress {
public static void main(String[] args) throws Exception {
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 void progress() {
System.out.print(".");
}
});
IOUtils.copyBytes(in, out, 4096, true);
}
}
该程序每当写入64k数据之后就调用一次progress()方法。
结果如下：
% hadoop FileCopyWithProgress input/docs/1400-8.txt hdfs://localhost/user/tom/1400-8.txt
...............

3.5.3.1. FSDataOutPutStream
FileSystem的create()方法返回了一个FSDataOutPutStream。提供了获取当前位置的方法，但是没有提供seek方法。
package org.apache.hadoop.fs;
public class FSDataOutputStream extends DataOutputStream implements Syncable {
public long getPos() throws IOException {
// implementation elided
}
// implementation elided
}

3.5.4. Directories
下面的方法创建目录，可以创建不存在的上级目录。因为在FileSystem的creat()方法中也能创建不存在的上级目录，所以一般不用这个方法。
public boolean mkdirs(Path f) throws IOException
3.5.5. Querying the FileSystem
3.5.5.1. File metadata：FileStatus
FileStatus封装了文件和目录的信息。包括他们的长度，块大小，复制的份数，修改时间，ownership，权限等信息。
FileSystem的getFileStatus()提供了获取某一文件或者目录的FileStatus的方法。
下面是一个例子：
public class ShowFileStatusTest {
private MiniDFSCluster cluster; // use an in-process HDFS cluster for
// testing
private FileSystem fs;

@Before
public void setUp() throws IOException {
Configuration conf = new Configuration();
if (System.getProperty("test.build.data") == null) {
System.setProperty("test.build.data", "/tmp");
}
cluster = new MiniDFSCluster(conf, 1, true, null);
fs = cluster.getFileSystem();
OutputStream out = fs.create(new Path("/dir/file"));
out.write("content".getBytes("UTF-8"));
out.close();
}

@After
public void tearDown() throws IOException {
if (fs != null) {
fs.close();
}
if (cluster != null) {
cluster.shutdown();
}
}

@Test(expected = FileNotFoundException.class)
public void throwsFileNotFoundForNonExistentFile() throws IOException {
fs.getFileStatus(new Path("no-such-file"));
}

@Test
public void fileStatusForFile() throws IOException {
Path file = new Path("/dir/file");
FileStatus stat = fs.getFileStatus(file);
assertThat(stat.getPath().toUri().getPath(), is("/dir/file"));
assertThat(stat.isDir(), is(false));
assertThat(stat.getLen(), is(7L));
assertThat(stat.getModificationTime(), is(lessThanOrEqualTo(System
.currentTimeMillis())));
assertThat(stat.getReplication(), is((short) 1));
assertThat(stat.getBlockSize(), is(64 * 1024 * 1024L));
assertThat(stat.getOwner(), is("tom"));
assertThat(stat.getGroup(), is("supergroup"));
assertThat(stat.getPermission().toString(), is("rw-r--r--"));
}

@Test
public void fileStatusForDirectory() throws IOException {
Path dir = new Path("/dir");
FileStatus stat = fs.getFileStatus(dir);
assertThat(stat.getPath().toUri().getPath(), is("/dir"));
assertThat(stat.isDir(), is(true));
assertThat(stat.getLen(), is(0L));
assertThat(stat.getModificationTime(), is(lessThanOrEqualTo(System
.currentTimeMillis())));
assertThat(stat.getReplication(), is((short) 0));
assertThat(stat.getBlockSize(), is(0L));
assertThat(stat.getOwner(), is("tom"));
assertThat(stat.getGroup(), is("supergroup"));
assertThat(stat.getPermission().toString(), is("rwxr-xr-x"));
}
}
如果文件不存在则会抛出FileNotFoundException。如果你仅对文件是否存在感兴趣，那么下面的方法更加适合：
public boolean exists(Path f) throws IOException
3.5.5.2. Listing Files
有时候需要获取一个目录里面的内容，这时候下面的方法就比较用户了，他能得到一个目录的FileStatus，方法有四个重载的函数。
public FileStatus[] listStatus(Path f) throws IOException
public FileStatus[] listStatus(Path f, PathFilter filter) throws IOException
public FileStatus[] listStatus(Path[] files) throws IOException
public FileStatus[] listStatus(Path[] files, PathFilter filter) throws IOException
当参数是一个文件的时候，返回的FileStatus的长度为1。当参数是目录的时候，返回数组的长度为0或者目录中所有文件的FileStatus.
Pathfilter起过滤作用。
下面的程序是以上方法的一个演示：
public class ListStatus {
public static void main(String[] args) throws Exception {
String uri = args[0];
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(uri), conf);
Path[] paths = new Path[args.length];
for (int i = 0; i < paths.length; i++) {
paths[i] = new Path(args[i]);
}
FileStatus[] status = fs.listStatus(paths);
Path[] listedPaths = FileUtil.stat2Paths(status);
for (Path p : listedPaths) {
System.out.println(p);
}
}
}
FileUtil.stat2Paths(status);将FileStatus转化成Path。
执行效果如下：
% hadoop ListStatus hdfs://localhost/ hdfs://localhost/user/tom
hdfs://localhost/user
hdfs://localhost/user/tom/books
hdfs://localhost/user/tom/quangle.txt
3.5.5.3. File pattern
Hadoop支持使用通配符来获取FileStatus的方法：
public FileStatus[] globStatus(Path pathPattern) throws IOException
public FileStatus[] globStatus(Path pathPattern, PathFilter filter) throws IOException
其中的pathPattern就可以含有通配符，通配符的具体说明见书61页。
3.5.5.4. PathFilter
PathFilter接口提供了更加灵活的扩充。
package org.apache.hadoop.fs;
public interface PathFilter {
boolean accept(Path path);
}
Filter只允许不符合PathFilter的文件通过。
下面是一个实现pathfilter接口的例子：
public class RegexExcludePathFilter implements PathFilter {
private final String regex;

public RegexExcludePathFilter(String regex) {
this.regex = regex;
}

public boolean accept(Path path) {
return !path.toString().matches(regex);
}
}
3.5.6. Deleting Data
public boolean delete(Path f, boolean recursive) throws IOException
如果f是一个文件或者空目录recursive的值就会被忽略。当一个目录不为空的时候：recursive为true时，目录将连同内部的内容都会被删除，否则抛出IOException异常。
3.5.7. Data Flow
3.5.7.1. Anatomy of a File Read

1） 客户端通过调用FileSystem对象的open()方法打开需要读取的文件，对HDFS来说是调用DistributedFileSystem
2） DistributedFileSystem通过RPC调用namenode确定文件的前几个block的位置。对于每一个block，namenode返回一含有那个block拷贝的datanode地址；接下来，datanode按照距离client的距离进行排序（确定距离的方法后面有介绍）。如果client本身就是一个datanode，那么就从本地datanode节点上读取数据。
DistributedFileSystem返回一个FSDataInputStream给客户端，让他从FSDataInputStream中读取数据。FSDataInputStream接着包装一个DFSInputStream，他用来管理datanode和namenode的I/O
3) client调用流的read()方法。
4) DFSInputStream开始的时候存放了前几个blocks的datanode的地址，这时候开始连接到最近datanode上。客户端反复调用read()方法，以流式方式从datanode读取数据。
5) 当读到block的结尾的时候，DFSInputStream会关闭到当前datanode的链接，然后查找下一个block的最好的datanode。这些操作对客户端都是透明的，客户端感觉到的是连续的流。（读取的时候就开始查找下一个块所在的地址）
6) 读取完成之后关闭FSDataInputStream

容错处理：
在读取期间，当client与datanode通信的时候如果发生错误的话，它会尝试读取下个紧接着的含有那个block的datanode。Client会记住发生错误datanode，这样它就不必在读取以后的块的时候再尝试这个datanode了。Client也验证从datanode传递过来的数据的checksum。如果错误的block被发现，它将在尝试从另一个datanode读取数据前被报告给namenode。

这个设计的一个重要方面是：客户端联系datanodes直接接收数据，并且客户端被namenode导向包含每块数据的最佳datanode。这样的设计可以使HDFS扩展而适应大量的客户端，因为数据传输线路是通过集群中的所有datanode的，namenode只需要相应块的位置查询服务即可（而namenode是将块的位置信息存放在内存中的，这样效率就非常高），namenode不需要提供数据服务，因为数据服务随着客户端的增加将很快成为瓶颈。

最短路径问题：
Hadoop计算路径是按照如下方式进行的：
1） 把网络看成树结构
2） 两个节点之间的距离=第一个节点到两个节点共同祖先节点的距离+第二个节点到两个节点共同祖先节点的距离
下面是一个例子：
distance(/d1/r1/n1, /d1/r1/n1) = 0 (processes on the same node)
distance(/d1/r1/n1, /d1/r1/n2) = 2 (different nodes on the same rack)
distance(/d1/r1/n1, /d1/r2/n3) = 4 (nodes on different racks in the same data center)
distance(/d1/r1/n1, /d2/r3/n4) = 6 (nodes in different data centers)

下面的图是书山的摘录，但是感觉有几处错误（如果我错了欢迎各位指出）：
R2上面的n1应该为n3，第二个datacenter上的n1应该为n4

下图为更加直白的表示，已经画成了树形，并且抽象出了两个数据中心共有的父节点

3.5.7.2. Anatomy of a File Write

写的过程如下：
1） client通过调用DistributedFileSystem的Create方法来请求创建文件
2） DistributedFileSystem通过对namenode发出rpc请求，在namenode的namespace里面创建一个新的文件，但是这时候并不关联任何的块。Namenode进行很多检查来保证不存在要创建的文件已经存在于文件系统中，还有检查是否有相应的权限来创建文件。如果这些检查都完成了，那么namenode将记录下来这个新文件的信息，否则文件创建失败，并且客户端会收到一个IOExpection。DistributedFileSystem返回一个FSDataOutputStream给客户端用来写入数据。和读的情形一样，FSDataOutputStream将包装一个DFSOutputStream用于和datanode及namenode通信。
3） 客户端开始写数据。DFSDataOutputStream把要写入的数据分成包（packet），并将它们写入到中间队列（data queue）中。Data queue中的数据由DataStreamer来读取。DataStreamer的职责是让namenode分配新的块——通过找出合适的datanodes——来存储作为备份而复制的数据。这些datanodes组成提个流水线，我们假设这个流水线是个三级流水线，那么里面将含有三个节点。DataStreamer将数据首先写入到流水线中的第一个节点，
4） 然后由第一个节点将数据包传送并写入到第二个节点，然后第二个将数据包传送并写入到第三个节点。
5） DFSOutputStream维护了一个内部关于packets的队列，里面存放等待被datanode确认无误的packets的信息。这个队列称为ack queue。一个packet的信息被移出本队列当且仅当packet被流水线中的所有节点都确认无误。

当正在写入数据的时候datanode发生错误的处理策略：
发现错误之后关闭流水线，然后将没有被确认的数据放到数据队列的开头，当前的块被赋予一个新的标识，这信息将发给namenode，以便在损坏的数据节点恢复之后删除这个没有被完成的块。然后从流水线中移除损坏的datanode。之后将这个块剩下的数据写入到剩下的两个节点中。Namenode注意到这个块的信息还没有被复制完成，他就在其他一个datanode上安排复制。接下来的block写入操作就和往常一样了。

尽管可能在写入数据的时候多个节点都出现故障，但是只要默认的一个节点（dfs.replication.min）被写入了，那么这个操作就会完成。因为数据块将会在集群间复制，直到复制完定义好的次数（dfs.replication，默认3份）

6） 当完成数据写入之后客户端调用流的close方法，在通知namenode完成写入之前，这个方法将flush残留的packets，并等待确认信息（acknowledgement）。
7） 因为先前已经存在DataStream请求namenode分配块这个操作，所以在这个阶段namenode会持有构成文件的块的信息。在block完成复制到最少的份数之后，namenode将成功返回。

备份文件的放置策略：
1） 第一份存放在客户端（如果客户端没在集群上，那么这个节点将被随机选择，尽管这样，系统也不会选择磁盘容量快满的，或者是比较忙的节点）
2） 第二份存放在与第一份不同机架的一个随机节点中
3） 第三份存放在与第二份相同的机架中，但是不在同一个节点
4） 接下来的就存放在集群中的随机节点中了，系统尽量避免在一个机架中存放多份备份文件。
3.5.7.3. Coherency Model
文件系统的连贯性模型描述了读写文件过程中的数据可见性。HDFS去掉了一些POSIX对性能的要求，所以一些操作可能与你的预想不大一致。
1） 在文件被创建之后，它在文件系统的名字空间中是可见的
Path p = new Path("p");
fs.create(p);
assertThat(fs.exists(p), is(true));
2） 但是任何没写入到文件的内容不保证可见，尽管你可能去flush流。所以文件看起来长度为0
Path p = new Path("p");
OutputStream out = fs.create(p);
out.write("content".getBytes("UTF-8"));
out.flush();
assertThat(fs.getFileStatus(p).getLen(), is(0L));
3） 当超过一个block的数据被写入之后，第一个block对reader将是可见的，接下来的也是一样：当前正在写的block总是不可见的，已经被写入的block是可见的。
4） HDFS通过FSDataOutputStream的sync()方法提供了一种强制使所有buffer同步到datanode方法。当sync()成功返回之后，HDFS保证sync之前的数据被持久化并且对所有reader可见。遇到client的crash时间，数据也不会丢失。
Path p = new Path("p");
FSDataOutputStream out = fs.create(p);
out.write("content".getBytes("UTF-8"));
out.flush();
out.sync();
assertThat(fs.getFileStatus(p).getLen(),is(((long) "content".length())));
这个行为 有点像unix系统的fsync系统调用，他把文件描述符的数据进行提交。
例如使用Java API来写本地文件，我们可能保证在flush stream和synchronizing之后数据是可见的：
FileOutputStream out = new FileOutputStream(localFile);
out.write("content".getBytes("UTF-8"));
out.flush(); // flush to operating system
out.getFD().sync(); // sync to disk
assertThat(localFile.length(), is(((long) "content".length())));
在HDFS中关闭一个文件其实是进行了一个隐含的sync()操作：
Path p = new Path("p");
OutputStream out = fs.create(p);
out.write("content".getBytes("UTF-8"));
out.close();
assertThat(fs.getFileStatus(p).getLen(), is(((long) "content".length())));
3.5.7.4. Consequences for application design
由于连贯性模型的原因，如果在写程序的时候没有调用sync函数，那么很有可能在客户端或者服务器出错的情况下丢失一个block的数据。对于很多程序来说这是不能接受的，所以要在程序中适时的调用sync()，例如在写入一定数量的记录或者一定量的数据之后就调用一下sync()。尽管sync()在设计的时候致力于不给HDFS增加太大的负荷，但是它确实要有些开销的，所以这样就有一个在数据的鲁棒性和吞吐量之间的权衡问题。一个可以接受的权衡是具有程序依赖性的，并且合适的值可以在测试系统的性能与不同的sync()调用频率之后被确定。
3.6. Parallel Copying with distcp
之前我们看到的HDFS访问模式都是单线程的访问。它可以通过指定文件通配符来做访问一批数据，但是对于高效、并行处理这些文件的时候你就必须自己写程序了。
Hadoop提供了一个非常有用的工具——distcp，来在Hadoop文件系统之间拷贝大量数据。
distcp的一个典型用途就是在两个HDFS集群之间传递数据。如果两个集群运行着不同版本的Hadoop，那么前缀hdfs需要加上：
% hadoop distcp hdfs://namenode1/foo hdfs://namenode2/bar
上面的代码将会把/foo目录连同他内部的文件从第一个集群拷贝到第二个集群的/bar目录。之后第二个集群将呈现出/bar/foo这样的目录结构，如果拷贝前bar目录不存在的话，它将会被先创建出来，源文件可以有多个，但是源文件的路径必须是绝对路径。
在默认情况下，distcp将会跳过已经存在的文件，但是在提供 –overwrite参数的情况下，存在的文件将会被覆盖。也可以选择-update参数来更新文件。

使用-overwrite或者-update选项改变了以前源路径和目标路径的使用方式，下面用一个例子解释：
假如在经过上面的拷贝操作之后，我们又改变了/foo目录下的一个文件的内容，更新的时候我们需要这样写：
% hadoop distcp -update hdfs://namenode1/foo hdfs://namenode2/bar/foo
Namenode2的/foo子目录需要加上。

除了上面介绍的选项之外还有很多选项可以控制distcp的行为，包括忽略失败，限制文件数量或者要拷贝的数据量等。调用distcp的时候不加参数可以看到这些选项的使用说明。

distcp是作为一个MapReduce job实现的。拷贝工作是并行运行在集群中的map节点完成的。没有reduce节点。每个文件由一个map来完成拷贝，并且distcp尝试通过分配给每个map近似相同分量的数据。（Each file is copied by a single map, and distcp tries to give each map approximately the same amount of data, by bucketing files into roughly equal allocations.）
map的数量由如下方式确定：
由于让每个map拷贝一个合理分量的数据来最小化task机构的开销是一个不错的注意，所以每个map至少拷贝256MB数据（除非数据量不够，这种情况下一个拷贝由一个map来完成）。例如：给一个1GB的文件分配4个map任务。当数据太大的时候限制map的数量来限制带宽和集群利用就有必要了。默认情况下，map的最大数量是20/tasktracker.例如拷贝1000GB文件的情况下，对于一个有100个节点的集群来说会启动2000个map（按照每个节点20个）。这样的情形下，每个节点将平均拷贝512MB的数据。可以启动的map数量可以通过调整distcp的-m参数来调整。例如-m 1000将分配1000个maps，每个拷贝1GB数据。

如果想在运行不同版本的HDFS集群之间拷贝使用hdfs协议运行distcp的话会产生错误。因为不同系统的rpc系统不兼容。为了补救，可以使用hftp从源文件中读取数据。但job就必须在拷贝的目标机器上运行，以便HDFS的rpc版本兼容。上面的例子可以写成下面的样子：
% hadoop distcp hftp://namenode1:50070/foo hdfs://namenode2/bar
注意：必须在uri中指定namenod的web端口号。这个端口的默认值是50070，由dfs.http.address属性值来决定。

3.6.1. Keeping an HDFS Cluster Balanced
当拷贝数据到HDFS的时候，考虑集群平衡很重要。当block在集群间均匀分布的时候性能最佳，所以要尽量保证distcp不要北打扰。
还拿刚才1000G的那个文件距离，如果指定-m为1的话，将只有一个map来进行拷贝工作——先不考虑速度问题和资源没有被充分利用的问题——这样就一位置每个block的第一个复制份将驻留在运行map的那个node上，直到硬盘被填满。第二和第三个复制品将分散在集群中，但是运行map的节点就不平衡了。通过启用超过集群中节点数量的map可以避免这个问题，也正是因为这样，最好是以默认是以默认的20个map每一个节点来运行distcp。
但并不可能总是阻止一个集群变得不平衡。可能你会想限制map的数量来使某些节点执行其他的任务。这种情况下，你可以使用一些均衡工具（balancer tool）来在接下来的时间中提高block在集群中的分布。
3.7. Hadoop Archives
HDFS存放小文件是低效大的，因为每个文件都存放在一个block中，而block的metadata保存在namenode的内存中。因此，大量的小文件会吃掉namenode的很多内存。注意：小文件并不需要占据比他自身更多的存储空间。1MB的文件储存在128MB的块中只占用1MB的磁盘空间，不占用128MB的空间。

Hadoop Archive（HAR）是一个文件打包工具，他在高效的将文件打包到HDFS block的同时，也可以减少namenode的内存使用量，并且仍旧允许客户端透明的访问文件。
HAR可以作为MapReduce的输入。
3.7.1. Using Hadoop Archives
HAR是使用archive工具打包一些文件创建的。Archive工具运行一个MapReduce job来并行处理输入文件。所以你需要在一个运行MapReduce的集群上使用它。
下面是一些要打包的文件：
% hadoop fs -lsr /my/files
-rw-r--r-- 1 tom supergroup 1 2009-04-09 19:13 /my/files/a
drwxr-xr-x - tom supergroup 0 2009-04-09 19:13 /my/files/dir
-rw-r--r-- 1 tom supergroup 1 2009-04-09 19:13 /my/files/dir/b
运行一下命令打包：
% hadoop archive -archiveName files.har /my/files /my
第一个参数要打包成的文件名（,har必须），第二个是要打包的文件，可以是目录，也可以有多个要打包的源，最后一个才参数是打包之后文件存放的位置。
运行上述命令之后，产生的.har文件信息如下：
% hadoop fs -ls /my
Found 2 items
drwxr-xr-x - tom supergroup 0 2009-04-09 19:13 /my/files
drwxr-xr-x - tom supergroup 0 2009-04-09 19:13 /my/files.har
% hadoop fs -ls /my/files.har
Found 3 items
-rw-r--r-- 10 tom supergroup 165 2009-04-09 19:13 /my/files.har/_index
-rw-r--r-- 10 tom supergroup 23 2009-04-09 19:13 /my/files.har/_masterindex
-rw-r--r-- 1 tom supergroup 2 2009-04-09 19:13 /my/files.har/part-0
上面的结果显示了：两个索引，一个part文件。对本例来说part文件只有一个。
Part文件包含了原始文件的内容，index用来索引这些数据。
使用har URI scheme与har文件交互：
% hadoop fs -lsr har:///my/files.har
drw-r--r-- - tom supergroup 0 2009-04-09 19:13 /my/files.har/my
drw-r--r-- - tom supergroup 0 2009-04-09 19:13 /my/files.har/my/files
-rw-r--r-- 10 tom supergroup 1 2009-04-09 19:13 /my/files.har/my/files/a
drw-r--r-- - tom supergroup 0 2009-04-09 19:13 /my/files.har/my/files/dir
-rw-r--r-- 10 tom supergroup 1 2009-04-09 19:13 /my/files.har/my/files/dir/b

以上都是在default FileSystem上面的操作，若是操作不同的FileSystem，可以用如下的形式：
% hadoop fs -lsr har://hdfs-localhost:8020/myfiles.har/my/files/dir
注意有一个hdfs指定FileSystem的scheme，localhost是主机地址，8082为端口
这与在目标机器上面运行：% hadoop fs -lsr har:///my/files.har效果是一样的

删除HAR文件：
%hadoop fs –rmr /my/files.har
3.7.2. Limitations
1） 创建的是归档文件，没有压缩功能，所以不会节省空间
2） 归档文件创建之后不能被修改，若要添加、删除文件的话，需要重新建立归档文件
3） 虽然HAR文件可以作为MapReduce的输入，但是InputFormat不支持将多个文件打包到一个MapReduce split中。所以处理大量的小文件，即使是在har文件中，都将是低效的。