hadoop筆記三：java訪問HDFS

1.HDFS的java訪問接口

1）org.apache.hadoop.conf.Configuration

讀取、解析配置文件(如core-site.xml/hdfs-default.xml/hdfs-site.xml等)，或添加配置的工具類。

2）org.apache.hadoop.fs.FileSystem

表明分佈式文件系統的一個實例，是一個通用的文件系統API，提供了不一樣文件系統的統一訪問方式。

3）org.apache.hadoop.fs.Path

是Hadoop文件系統中統一的文件或目錄描述，相似於java.io.File對本地文件系統的文件或目錄描述。

4）org.apache.hadoop.fs.FSDataInputStream

對Hadoop中數據輸入流的統一封裝

5）org.apache.hadoop.fs.FSDataInputStream

對Hadoop中數據輸出流的統一封裝

2.java訪問HDFS主要編程步驟

1）構建Configuration對象，讀取並解析相關配置文件

Configuration conf = new Configuration();

2）設置相關屬性

conf.set("fs.defaultFS","hdfs://192.168.1.113:8020");

3）獲取特定文件系統實例fs（以HDFS文件系統實例）

FileSystem fs=FileSystem.get(URI.create("hdfs://192.168.1.113:8020/user/root/input/words"), conf);

4）經過文件系統實例fs進行文件操做(以刪除文件實例)

fs.delete(new Path("/user/root/someWords.txt"));

3.操做實例

1）查詢HDFS集羣文件系統中一個文件，並打印文件內容

首先在文件系統中確保所要讀取的文件存在

java編碼讀取

package com.jf.hdfs;

import java.io.IOException;
import java.net.URI;
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 CatHdfsFile {

	private static String file = "hdfs://192.168.1.113:8020/user/root/input/words";

	public static void main(String[] args) throws IOException {
	
		//Configuration是讀取、解析配置文件(如core-site.xml/hdfs-default.xml/hdfs-site.xml等)，或添加配置的工具類
		Configuration conf = new Configuration();
		
		//FileSystem是一個通用的文件系統API，提供了不一樣文件系統的統一訪問方式
		FileSystem fs = FileSystem.get(URI.create(file), conf);
		
		//Path是Hadoop文件系統中統一的文件或目錄描述，相似於java.io.File對本地文件系統的文件或目錄描述
		Path path = new Path(file);
		
		//FSDataInputStream對Hadoop中數據輸入流的統一封裝
		FSDataInputStream is = fs.open(path);
		
		byte[] buff = new byte[1024];
		int length = 0;
		while ((length = is.read(buff)) != -1) {
			System.out.println(new String(buff, 0, length));
		}
	}
}

執行輸出結果

hello a
world b
hello c
jiafeng b
sean a
這是測試文件，
只是一個測試

4.java訪問HDFS中數據塊

1.相關java類介紹

Configured類：Configured類中實現了Configurable接口，實現了getConf和setConf方法能夠獲取Configuration對象。Configuration對象能夠獲取Hadoop的全部配置文件中的數據，還能夠經過使用命令行中使用-D（-D是一個標識）使用的變量以及值

// inherit javadoc
  @Override
  public void setConf(Configuration conf) {
    this.conf = conf;
  }

  // inherit javadoc
  @Override
  public Configuration getConf() {
    return conf;
  }

Tool接口：裏面有一個run方法，能夠經過ToolRunner來執行，Tool裏面的run方法。

public static int run(Configuration conf, Tool tool, String[] args) 
    throws Exception{
    if(conf == null) {
      conf = new Configuration();
    }
    GenericOptionsParser parser = new GenericOptionsParser(conf, args);
    //set the configuration back, so that Tool can configure itself
    tool.setConf(conf);
    
    //get the args w/o generic hadoop args
    String[] toolArgs = parser.getRemainingArgs();
    return tool.run(toolArgs);
  }

LocatedBlock類：存儲文件的數據塊對象，能夠獲取存儲文件的全部datanode。

ExtendedBlock類：經過LocatedBlock的getBlock()來獲取，該類主要能夠獲取數據塊的id和name。

2.java代碼訪問程序

package com.jf.hdfs;

import java.net.URI;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.client.HdfsDataInputStream;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FileBlockList extends Configured implements Tool {

	public int run(String[] arg0) throws Exception {
		// 獲取Configuration對象
		Configuration conf = getConf();
		// 經過-Dinput獲取參數
		String input = conf.get("input");
		// 建立文件系統對象
		FileSystem fs = FileSystem.get(URI.create(input), conf);
		// 獲取文件輸入流對象
		HdfsDataInputStream hdis = (HdfsDataInputStream) fs.open(new Path(input));
		// 獲取全部數據塊列表
		List<LocatedBlock> blocks = hdis.getAllBlocks();
		// 遍歷數據塊
		for (LocatedBlock locatedBlock : blocks) {
			// 存有該block的ID和名稱
			ExtendedBlock eBlock = locatedBlock.getBlock();
			System.out.println("id=" + eBlock.getBlockId() + ",name=" + eBlock.getBlockName() + ",blockSize="
					+ locatedBlock.getBlockSize() + ",startOffset=" + locatedBlock.getStartOffset());
			// 獲取該數據塊所在數據節點的信息
			DatanodeInfo[] datanodes = locatedBlock.getLocations();
			for (DatanodeInfo datanodeInfo : datanodes) {
				System.out.println("IPAddr=" + datanodeInfo.getIpAddr() + ",hostname=" + datanodeInfo.getHostName());
			}
		}
		return 0;
	}

	public static void main(String[] args) throws Exception {
		
		System.exit(ToolRunner.run(new FileBlockList(), args));
	}

}

打包以後放入客戶端執行該代碼程序：

hadoop-2.7.1/bin/hadoop jar my_hadoop-0.0.1-SNAPSHOT.jar com.jf.hdfs.FileBlockList -Dinput=/user/root/input/words

結果：

id=1073741855,name=blk_1073741855,blockSize=82,startOffset=0
IPAddr=192.168.1.113,hostname=hadoop

數據塊的偏移量是指一個數據塊距離一個文件開始的偏移位置。

5.java訪問HDFS集羣文件系統

1.相關類和方法

FileStatus類：封裝文件和目錄的文件系統元數據，包括文件長度，塊大小，複製，修改時間，全部權和許可信息。

FileSystem上的getFileStatus（）方法提供了一種獲取FileStatus的方法對象爲單個文件或目錄。

getAccessTime()  //上次訪問的時間
getOwner()   //文件的全部者
getGroup()   //文件的所屬者
getPath()  //獲得文件的路徑
getPermission()  //文件的權限
getReplication()  //文件的備份數

2.java代碼編程

package com.jf.hdfs;

import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ListFileStatus extends Configured implements Tool {

	private FileSystem fs;

	public int run(String[] args) throws Exception {
		Configuration conf = getConf();
		String input = conf.get("input");
		fs = FileSystem.get(URI.create(input), conf);
		// 封裝文件和目錄的文件系統元數據，包括文件長度，塊大小，複製，修改時間，全部權和許可信息
		FileStatus[] fileStatuses = fs.listStatus(new Path(input));
		for (FileStatus fileStatus : fileStatuses) {
			process(fileStatus);
		}
		return 0;
	}

	private void process(FileStatus status) throws Exception {
		if (status.isFile()) {// 爲文件時
			System.out.println("------------------");
			System.out.println("上次訪問時間=" + status.getAccessTime());
			System.out.println("文件全部者=" + status.getOwner());
			System.out.println("文件所屬者=" + status.getGroup());
			System.out.println("文件路徑=" + status.getPath());
			System.out.println("文件權限=" + status.getPermission());
			System.out.println("文件備份數=" + status.getReplication());
			System.out.println("------------------");
		} else if (status.isDirectory()) {// 判斷是文件夾時進行遞歸調用
			// 在HDFS中，當FileStatus對象表明一個目錄的時候沒有相應的方法來獲取該目錄下的全部文件要經過FileSystem類來獲取該目錄下的文件
			FileStatus[] fileStatus = fs.listStatus(status.getPath());
			for (FileStatus fileStatus2 : fileStatus) {
				process(fileStatus2);
			}
		}
	}

	public static void main(String[] args) throws Exception {
		System.exit(ToolRunner.run(new ListFileStatus(), args));
	}
}

3.執行結果

執行命令

hadoop-2.7.1/bin/hadoop jar my_hadoop-0.0.1-SNAPSHOT.jar com.jf.hdfs.ListFileStatus -Dinput=/user/root/input

輸出結果