當(dāng)數(shù)據(jù)量很大，一臺物理機(jī)無法容納時(shí)，我們就需要將數(shù)據(jù)存儲(chǔ)到由網(wǎng)絡(luò)連接的若干臺機(jī)器上，這就是所謂的分布式文件系統(tǒng)。hadoop 使用 hdfs 作為自己的分布式文件系統(tǒng)。目前 hdfs 不適合低延遲訪問，也不適合存儲(chǔ)大量的小文件以及隨機(jī)寫入（hdfs 中只能在文件末尾進(jìn)行寫操作）。
1 hdfs 的概念
硬盤有自己的塊大小，表示其可以讀寫的最小數(shù)據(jù)量。文件系統(tǒng)塊的大小為若干倍的硬盤大小。
hdfs 上的文件也以塊來劃分，默認(rèn)為 128M，為了容錯(cuò)，每個(gè) block 通常在其他節(jié)點(diǎn)有若干個(gè)備份，一個(gè)節(jié)點(diǎn)出問題了可以使用別的節(jié)點(diǎn)的數(shù)據(jù)。
使用下面的命令可以查看 hdfs 的信息：
hdfs fsck / -files -blocks
2 namenode 與 datanode
hdfs 集群包括兩種節(jié)點(diǎn)：
- namenode：是集群的管理者，記錄著文件系統(tǒng)樹以及所有文件和目錄的元數(shù)據(jù)。namenode 還記錄著組成文件的 block 的位置（在每次系統(tǒng)啟動(dòng)時(shí)建立）
- datanode：存儲(chǔ)數(shù)據(jù)并提供數(shù)據(jù)，定時(shí)向 namenode 同步自己存儲(chǔ)的 block 的信息
經(jīng)常被讀取的文件的相關(guān) block 會(huì)被緩存在 datanode 的內(nèi)存中，用戶也可以指定哪些文件的 block 應(yīng)當(dāng)被緩存。
3 命令行接口
在學(xué)習(xí) hadoop 時(shí)，我們使用偽分布式的方法啟動(dòng) hadoop 集群。在 **core-site.xml ** 中可以通過設(shè)置下面的屬性來配置默認(rèn)的文件系統(tǒng)。
<property>
<name>fs.defaultFS</name>
<value>hdfs://localhost:9000</value>
</property>
基礎(chǔ)的操作有：
//將本地文件系統(tǒng)的數(shù)據(jù)復(fù)制到 hdfs 中，由于已經(jīng)配置了 fs.defaultFS 可以省略 hdfs://localhost:9000
hdfs dfs -copyFromLocal /test/data/* /test1/input
//使用相對路徑會(huì)將數(shù)據(jù)復(fù)制到用戶目錄下 我這里是 /user/hadoop
hdfs dfs -copyFromLocal /test/data/* input
//從hdfs copy 到本地
hdfs dfs -copyToLocal test.txt text.txt
//創(chuàng)建文件夾
hdfs dfs -mkdir books
//列出所有文件
hdfs dfs -ls .
//print 文件的內(nèi)容
hdfs dfs -cat /user/hadoop/books
hdfs 的文件也有 r w x 的概念，當(dāng)然這里的 x 是指是否可以訪問一個(gè)目錄下面的文件。hdfs 中的文件也有所屬 user 以及 group 的概念，使用當(dāng)前調(diào)用者進(jìn)程的 user 和 group 來標(biāo)識，運(yùn)行 namenode 的用戶有 “root” 權(quán)限。
hadoop 有一個(gè)抽象的文件系統(tǒng)，hdfs 是其實(shí)現(xiàn)之一，通過 uri 的方式，可以選擇不同的文件系統(tǒng)，比如hdfs
hdfs dfs -ls file:///home/hadoop
4 java api
使用 URL 就可以訪問 hdfs 文件，不過要進(jìn)行一點(diǎn)小處理，設(shè)置 UrlStreamHandlerFactory：
不方便設(shè)置 UrlStreamHandlerFactory 時(shí)，可以使用 FileSystem Api：
public class ReadFile {
public static void main(String[] args) throws IOException {
if (args.length != 1) {
System.out.println("Usage: ReadFile path/to/file");
System.exit(-1);
}
String filePath = args[0];
if (!filePath.startsWith("/")) {
filePath = "/" + filePath;
}
filePath = "hdfs://localhost:9000" + filePath;
// conf 會(huì)根據(jù) classpath 中的 core-site.xml 創(chuàng)建
Configuration conf = new Configuration();
//創(chuàng)建 fileSystem 實(shí)例
FileSystem fileSystem = FileSystem.get(URI.create(filePath), conf);
InputStream is = null;
try {
//使用 open 打開 InputStream
is = fileSystem.open(new Path(filePath));
IOUtils.copyBytes(is, System.out, 4096, false);
} finally {
IOUtils.closeStream(is);
}
}
}
其中，F(xiàn)ileSystem.open 返回的是一個(gè) FSDataInputStream 對象，可以使用 seek 調(diào)整讀取的位置：
另外，其還實(shí)現(xiàn)了 PositionedReadable 接口，可以從指定位置讀取指定的數(shù)據(jù)：
public interface PositionedReadable {
/**
* Read upto the specified number of bytes, from a given
* position within a file, and return the number of bytes read. This does not
* change the current offset of a file, and is thread-safe.
*/
public int read(long position, byte[] buffer, int offset, int length)
throws IOException;
/**
* Read the specified number of bytes, from a given
* position within a file. This does not
* change the current offset of a file, and is thread-safe.
*/
public void readFully(long position, byte[] buffer, int offset, int length)
throws IOException;
/**
* Read number of bytes equal to the length of the buffer, from a given
* position within a file. This does not
* change the current offset of a file, and is thread-safe.
*/
public void readFully(long position, byte[] buffer) throws IOException;
}
使用 FileSystem api 也可以進(jìn)行寫操作：
String localFile = args[0];
String hdfsFile = args[1];
if (!hdfsFile.startsWith("/")) {
hdfsFile = "/" + hdfsFile;
}
hdfsFile = "hdfs://localhost:9000" + hdfsFile;
InputStream in = null;
FSDataOutputStream out = null;
try {
in = new BufferedInputStream(new FileInputStream(localFile));
FileSystem fileSystem = FileSystem.get(URI.create(hdfsFile), new Configuration());
out = fileSystem.create(new Path(hdfsFile));
IOUtils.copyBytes(in, out, 4096, true);
} catch (Exception e) {
e.printStackTrace();
}
通過 FileSystem.getStatus 可以查看文件的狀態(tài)。
通過 FileSystem.listStatus 可以獲取文件夾下的所有文件的信息。
通過 FileSystem.globStatus 可以按照一定的 pattern 查看
通過 PathFilter 可以在globStatus 加入更細(xì)粒度的控制:
fs.globStatus(new Path("/2007/*/*"), new RegexExcludeFilter("^.*/2017/12/31$"));
刪除數(shù)據(jù)使用 FileSystem 的 delete 方法進(jìn)行。
文件讀取的過程中，從 namenode 獲取 block 的位置，再去對應(yīng)的 datanode 讀取數(shù)據(jù)：
文件寫入數(shù)據(jù)的過程中，先在 namenode 創(chuàng)建元數(shù)據(jù)，然后向一個(gè) datanode 寫入數(shù)據(jù)，該 datanode 會(huì)同時(shí)向其他 datanode 同步數(shù)據(jù)。
數(shù)據(jù)寫入時(shí)，應(yīng)該在需要的地方調(diào)用輸出流的 sync，以便于將緩存寫入文件系統(tǒng)中，注意 close 時(shí)也會(huì)調(diào)用一次 sync。
5 distcp
使用 distcp 可以并發(fā)的進(jìn)行數(shù)據(jù)復(fù)制，個(gè)人認(rèn)為功能可以與 rsync 類比。
hadoop@millions-server:/usr/local/hadoop/etc/hadoop$ hdfs dfs -ls testdata
Found 1 items
-rw-r--r-- 1 hadoop supergroup 288374 2017-03-07 23:22 testdata/synthetic_control.data
而后輸入：
hadoop distcp -update testdata/synthetic_control.data testdata/synthetic_control2.data
可以看到：
hadoop@millions-server:/usr/local/hadoop/etc/hadoop$ hdfs dfs -ls testdata
Found 2 items
-rw-r--r-- 1 hadoop supergroup 288374 2017-03-07 23:22 testdata/synthetic_control.data
-rw-r--r-- 1 hadoop supergroup 288374 2017-05-06 23:09 testdata/synthetic_control2.data
再來同步一個(gè)文件夾：
hadoop distcp testdata testdata2
hdfs dfs -ls testdata2
Found 2 items
-rw-r--r-- 1 hadoop supergroup 288374 2017-05-06 23:14 testdata2/synthetic_control.data
-rw-r--r-- 1 hadoop supergroup 288374 2017-05-06 23:14 testdata2/synthetic_control2.data
刪除 testdata 中的 synthetic_control2.data：
hdfs dfs -rm testdata/synthetic_control2.data
再進(jìn)行一次帶 -update 和 -delete 的 distcp，可以看到，多余的文件被刪除了：
hadoop distcp -update -delete testdata testdata2
hdfs dfs -ls testdata2
Found 1 items
-rw-r--r-- 1 hadoop supergroup 288374 2017-05-06 23:14 testdata2/synthetic_control.data
