第十二章：HDFS客戶端操作

05-03

第十二章：HDFS客戶端操作

（作者：memories）

一、HDFS客戶端環境準備

1、根據自己電腦的操作系統拷貝對應的編譯後的hadoop jar包到非中文路徑（例如：D:Develophadoop-2.7.2）。

百度雲鏈接：https://pan.baidu.com/s/1EpnuaTg_XFXujFRWT4D2Jg 密碼：xxx0

2、配置HADOOP_HOME環境變數

3、配置Path環境變數

4、創建一個Maven工程HdfsClientDemo

5、打開工程文件項目，找到pom.xml導入相應的依賴

<groupId>junit</groupId>

<artifactId>junit</artifactId>

<version>RELEASE</version>

</dependency>

<groupId>org.apache.logging.log4j</groupId>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-common</artifactId>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-hdfs</artifactId>

</dependency>

</dependencies>

6、在src/main/java下創建包名：com.memories.hdfs

7、在com.memories.hdfs中創建HdfsClient類

public class HdfsClient{

@Test

public void testMkdirs() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件系統（選擇hadoop的配置文件）

Configuration configuration = new Configuration();

// 配置在集群上運行

// configuration.set("fs.defaultFS", "hdfs://hadoop101:9000");

// FileSystem fs = FileSystem.get(configuration);

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 創建目錄

fs.mkdirs(new Path("/xiaomo/nihao"));

// 3 關閉資源

fs.close();

}

8、執行程序

運行時需要配置用戶名稱

客戶端去操作hdfs時，是有一個用戶身份的。默認情況下，hdfs客戶端api會從jvm中獲取一個參數來作為自己的用戶身份：-DHADOOP_USER_NAME=memories，memories為用戶名稱。

8）注意：如果eclipse列印不出日誌，在控制台上只顯示

1.log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).

2.log4j:WARN Please initialize the log4j system properly.

3.log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

需要在項目的src/main/resources目錄下，新建一個文件，命名為「log4j.properties」，在文件中填入

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender

log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

log4j.appender.logfile=org.apache.log4j.FileAppender

log4j.appender.logfile.File=target/spring.log

log4j.appender.logfile.layout=org.apache.log4j.PatternLayout

log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

二、HDFS的API操作

1、HDFS文件上傳（測試參數優先順序）

1）編寫源代碼

@Test

public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {

// 1 獲取文件系統

Configuration configuration = new Configuration();

configuration.set("dfs.replication", "2");

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 上傳文件

fs.copyFromLocalFile(new Path("e:/hello.txt"), new Path("/hello.txt"));

// 3 關閉資源

fs.close();

System.out.println("over");

}

2）將hdfs-site.xml拷貝到項目的根目錄下

<?xml version="1.0" encoding="UTF-8"?>

<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>

<name>dfs.replication</name>

</property>

</configuration>

3）參數優先順序

參數優先順序排序：（1）客戶端代碼中設置的值 >（2）classpath下的用戶自定義配置文件 >（3）然後是伺服器的默認配置

2、HDFS文件下載

@Test

public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 執行下載操作

// boolean delSrc 指是否將原文件刪除

// Path src 指要下載的文件路徑

// Path dst 指將文件下載到的路徑

// boolean useRawLocalFileSystem 是否開啟文件校驗

fs.copyToLocalFile(false, new Path("/hello1.txt"), new Path("e:/hello1.txt"), true);

// 3 關閉資源

fs.close();

}

3、HDFS文件夾刪除

@Test

public void testDelete() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 執行刪除

fs.delete(new Path("/xiaomo/"), true);

// 3 關閉資源

fs.close();

}

4、HDFS文件名更改

@Test

public void testRename() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 修改文件名稱

fs.rename(new Path("/hello.txt"), new Path("/hello6.txt"));

// 3 關閉資源

fs.close();

}

5、HDFS文件詳情查看

查看文件名稱、許可權、長度、塊信息

@Test

public void testListFiles() throws IOException, InterruptedException, URISyntaxException{

// 1獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 獲取文件詳情

RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

while(listFiles.hasNext()){

LocatedFileStatus status = listFiles.next();

// 輸出詳情

// 文件名稱

System.out.println(status.getPath().getName());

// 長度

System.out.println(status.getLen());

// 許可權

System.out.println(status.getPermission());

// z組

System.out.println(status.getGroup());

// 獲取存儲的塊信息

BlockLocation[] blockLocations = status.getBlockLocations();

for (BlockLocation blockLocation : blockLocations) {

// 獲取塊存儲的主機節點

String[] hosts = blockLocation.getHosts();

for (String host : hosts) {

System.out.println(host);

}

System.out.println("----------------分割線-----------");

}

6、HDFS文件和文件夾判斷

@Test

public void testListStatus() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件配置信息

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 判斷是文件還是文件夾

FileStatus[] listStatus = fs.listStatus(new Path("/"));

for (FileStatus fileStatus : listStatus) {

// 如果是文件

if (fileStatus.isFile()) {

System.out.println("f:"+fileStatus.getPath().getName());

}else {

System.out.println("d:"+fileStatus.getPath().getName());

}

// 3 關閉資源

fs.close();

}

三、HDFS的I/O流操作

1、HDFS文件上傳

@Test

public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException {

// 1 獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 創建輸入流

FileInputStream fis = new FileInputStream(new File("e:/hello.txt"));

// 3 獲取輸出流

FSDataOutputStream fos = fs.create(new Path("/hello4.txt"));

// 4 流對拷

IOUtils.copyBytes(fis, fos, configuration);

// 5 關閉資源

IOUtils.closeStream(fis);

IOUtils.closeStream(fos);

}

2、HDFS文件下載

1）需求：從HDFS上下載文件到本地e盤上。

2）編寫代碼：

// 文件下載

@Test

public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 獲取輸入流

FSDataInputStream fis = fs.open(new Path("/hello1.txt"));

// 3 獲取輸出流

FileOutputStream fos = new FileOutputStream(new File("e:/hello1.txt"));

// 4 流的對拷

IOUtils.copyBytes(fis, fos, configuration);

// 5 關閉資源

IOUtils.closeStream(fis);

IOUtils.closeStream(fos);

fs.close();

}

3、定位文件讀取

1）下載第一塊

@Test

public void readFileSeek1() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 獲取輸入流

FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));

// 3 創建輸出流

FileOutputStream fos = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part1"));

// 4 流的拷貝

byte[] buf = new byte[1024];

for(int i =0 ; i < 1024 * 128; i++){

fis.read(buf);

fos.write(buf);

}

// 5關閉資源

IOUtils.closeStream(fis);

IOUtils.closeStream(fos);

}

2）下載第二塊

@Test

public void readFileSeek2() throws IOException, InterruptedException, URISyntaxException{

// 1 獲取文件系統

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "memories");

// 2 打開輸入流

FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));

// 3 定位輸入數據位置

fis.seek(1024*1024*128);

// 4 創建輸出流

FileOutputStream fos = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part2"));

// 5 流的對拷

IOUtils.copyBytes(fis, fos, configuration);

// 6 關閉資源

IOUtils.closeStream(fis);

IOUtils.closeStream(fos);

}

3）合併文件

在window命令窗口中執行

type hadoop-2.7.2.tar.gz.part2 >> hadoop-2.7.2.tar.gz.part1