Hadoop HDFS

1. Hadoop文件系统简介

Hadoop有一个抽象的文件系统概念，由FileSystem接口定义，有以下具体实现：

Local, HDFS, HFTP, HSFTP, HAR, hfs, FTP, S3(原生), S3(基于块)

2. Hadoop文件系统基本操作

Hadoop提供了命令行来操作文件系统。

查看帮助文档

hadoop fs -help

通过URI前缀来区分不同类型的文件系统，设置默认的文件系统

core-site.xml

<property>
  <name>fs.default.name</name>
  <value>hdfs://192.168.79.128:8020</value>
  <final>true</final>
</property>

这样，假如命令行中不指定文件系统则在该文件系统下，例如下面两个操作结果一直

hadoop fs -ls hdfs://192.168.79.128:8020/user/hadoop
hadoop fs -ls /user/hadoop

3. HDFS

Hadoop Distributed Filesystem，Hadoop分布式文件系统，以流式数据访问模式来存储超大文件。

-- 数据块

磁盘的数据块：磁盘进行数据读/写的最小单位。

文件系统通过磁盘块来管理文件系统中的块，该文件系统块的大小为磁盘块的整数倍。

HDFS上的文件也被划分为块大小的多个分块，默认为64MB，但是HDFS中小于一个块大小的文件不会占据整个块的空间。

hdfs-site.xml

<property>
  <name>dfs.block.size</name>
  <value>67108864</value>
</property>

-- namenode

管理文件系统的命令空间，维护着文件系统及整棵树内所有的文件和目录。

-- datanode

文件系统的工作节点，根据需要存储并检索数据块（受客户端或namenode调度），并且定期向namenode发送它们所存储的块的列表。

-- 容错机制

1）备份namenode的元数据

hdfs-site.xml 

<property>
  <name>dfs.name.dir</name>
  <value>/home/hadoop/hdfs/name,/home/hadoop/hdfs/bak/name</value>
  <final>true</final>
</property>

2）运行一个辅助namenode，定期通过编辑日志合并命名空间镜像，以防止编辑日志过大，状态滞后于namenode，所以在namenode全部失效时，难免会丢失部分数据。 

masters 

192.168.79.129

hdfs-site.xml 

<property>
  <name>dfs.checkpoint.dir</name>
  <value>/home/hadoop/hdfs/checkpoint</value>
  <final>true</final>
</property>

3）datanode数据冗余 

hdfs-site.xml 

<property>
  <name>dfs.replication</name>
  <value>3</value>
</property>

<property>
  <name>dfs.data.dir</name>
  <value>/home/hadoop/hdfs/data</value>
  <final>true</final>
</property>

4. JAVA API 

-- 读取：通过URL对象打开数据流，此时需要借助FsUrlStreamHandlerFactory来让JAVA程序能识别Hadoop的文件系统。

URLCat.java

package com.hadoop.study.fs;

import java.io.InputStream;
import java.net.URL;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;

public class URLCat {
	
	static {
		URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
	}
	
	public static void main(String[] args) throws Exception {
		InputStream in = null;
		try {
			in = new URL("hdfs://192.168.79.128/user/hadoop/output/part-r-00000").openStream();
			IOUtils.copy(in, System.out);
		} finally {
			IOUtils.closeQuietly(in);
		}
	}

}

注：URL.setURLStreamHandlerFactory(...)在一个JVM钟只能调用一次

-- 读取：FSDataInputStream.java

package com.hadoop.study.fs;

import java.net.URI;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class FileSystemDoubleCat {

	public static void main(String[] args) throws Exception {
		FileSystem fs = FileSystem.get(new URI("hdfs://192.168.79.128:8020"), new Configuration());
		FSDataInputStream in = null;
		try {
			in = fs.open(new Path("/user/hadoop/output/part-r-00000"));
			IOUtils.copy(in, System.out);
			in.seek(0);
			IOUtils.copy(in, System.out);
		} finally {
			IOUtils.closeQuietly(in);
		}
	}

}

-- 写入：FSDataOutputStream.java

package com.hadoop.study.fs;

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

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;

public class FileCopy {

	public static void main(String[] args) throws Exception {
		
		InputStream in = null;
		FSDataOutputStream out = null;
		FileSystem fs = FileSystem.get(new URI("hdfs://192.168.79.128:8020"), new Configuration());
		try {
			in = new BufferedInputStream(FileCopy.class.getResourceAsStream("/1902"));
			out = fs.create(new Path("/user/hadoop/input/1902"), true);
			IOUtils.copy(in, out);
		} finally {
			IOUtils.closeQuietly(in);
			IOUtils.closeQuietly(out);
		}
		
	}

}

运行此程序的时由于HDFS权限校验会抛出异常，可通过下述方式配置HDFS使得不进行权限校验

hdfs-site.xml

<property>
  <name>dfs.permissions</name>
  <value>true</value>
</property>

-- FileStatus 封装了文件系统中文件和目录的元数据(hadoop fs -ls中的信息)

-- PathFilter 和 pathPattern 用于过滤文件

5. 权限管理

与POSIX相似，提供了三类权限模式：只读(r)，写入(w)，可执行(x)

对文件来说x可以忽略，但是访问目录子项时需要该权限。

超级用户：namenode进程的标识，用户不会执行任何权限检查。

Kerberos：可以用来实现Hadoop的用户认证。

6. 网络拓扑

在海量数据处理中，其主要限制因素是节点之间数据的传输速率，但是要衡量节点之间的带宽很难实现。

Hadoop网络看成一棵树，两个节点之间的距离是它们到最近的共同祖先的距离总和。计算方式如下：

1）同一节点中的进程 0 (/d1/r1/n1, /d1/r1/n1)

2）同一机架上的不同节点 2 (/d1/r1/n1, /d1/r1/n2)

3）同一数据中心不同机架上的节点 4 (/d1/r1/n1, /d1/r2/n3)

4）不同数据中心的节点 6 (/d1/r1/n1, /d2/r3/n4)

-- 配置

默认情况下，网络是平铺的，仅有单一层次，所有节点都在同一数据中心的同一机器上。

core-site.xml

<property>
  <name>topology.script.file.name</name>
  <value></value>
  <description> The script name that should be invoked to resolve DNS names to
    NetworkTopology names. Example: the script would take host.foo.bar as an
    argument, and return /rack1 as the output.
  </description>
</property>

-- 作用

1）读取

namenode告知客户端每个块中最佳的datanode

2）复本布局

namenode如何选择在哪个datanode存储复本，这需要在可靠性，写入带宽和读取带宽之间进行权衡。

默认布局策略：

在运行客户端的节点上放1个复本

第2个复本放在与第1个不同且随机另外选择的机架节点上

第3个复本与第2个复本放在相同的机架，且随机选择另一个节点

其它复本放在集群中随机选择的节点上，不过系统会尽量避免在相同的机架上放太多复本

可以通过均衡器来调整

7. Hadoop存档 HAR文件系统

一个高效的文件存档工具，将文件存入HDFS块，在减少namenode内存使用的同时，还能允许对文件进行透明的访问。一旦创建，存档文件便不能再修改。

注：尽管HAR文件可以作为MapReduce的输入，但是InputFormat类并不知道文件已经存档，所以即使在HAR文件中处理许多小文件，也仍然是低效的。

-- 创建：通过运行一个MapReduce作业来处理。

hadoop archive -archiveName input.har -p /user/hadoop/input /user/hadoop

-- 查看：

直接查看

hadoop fs -ls /user/hadoop/input.har

结果

-rw-r--r--   2 hadoop supergroup          0 2014-02-25 22:38 /user/hadoop/input.har/_SUCCESS
-rw-r--r--   5 hadoop supergroup        202 2014-02-25 22:38 /user/hadoop/input.har/_index
-rw-r--r--   5 hadoop supergroup         21 2014-02-25 22:38 /user/hadoop/input.har/_masterindex
-rw-r--r--   2 hadoop supergroup    1777168 2014-02-25 22:37 /user/hadoop/input.har/part-0

列表显示了HAR文件的组成部分：两个索引文件以及部分文件的集合。通过索引可以找到包含在存档文件中的部分文件，它的起始点和长度。

通过HAR文件系统查看

hadoop fs -ls har:///user/hadoop/input.har

结果

-rw-r--r--   2 hadoop supergroup     888978 2014-02-25 19:02 /user/hadoop/input.har/1902
-rw-r--r--   2 hadoop supergroup     888190 2014-02-24 23:02 /user/hadoop/input.har/1901

-- 通过JAVA API读取

ArchiveCat.java

package com.hadoop.study.fs;

import java.io.InputStream;
import java.net.URI;

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

public class ArchiveCat {

	public static void main(String[] args) throws Exception {
		//URI: har://hdfs-192.168.79.128:8020/user/hadoop/input.har/ is an invalid Har URI since host==null.  
		//Expecting har://<scheme>-<host>/<path>.
		//FileSystem fs = FileSystem.get(new URI("har://hdfs-192.168.79.128:8020/user/hadoop/input.har/"), new Configuration());
		FileSystem fs = FileSystem.get(new URI("har://hdfs-hdmaster:8020/user/hadoop/input.har/"), new Configuration());
		InputStream in = null;
		try {
			in = fs.open(new Path("1901"));
			IOUtils.copy(in, System.out);
		} finally {
			IOUtils.closeQuietly(in);
		}
	}

}

8. 参考资料

Hadoop权威指南