hdfs客戶端操作 · JAVA

[TOC] # 用戶身份客戶端操作hdfs時是有一個用戶身份的,默認情況下,hdfs客戶端api會從jvm中獲取一個參數來作為自己的用戶身份,`-DHADOOP_USER_NAME=x`, x為用戶名稱\ 代碼中也可以設置,代碼優先級最高 # 列出文件 ~~~ package com.hdfs; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.LocatedFileStatus; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.RemoteIterator; import java.io.IOException; public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS","hdfs://master:9000"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //第一個參數是路徑,第二個參數是遞歸,false表示不需要 //獲取/下的文件 RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), false); //通過這個迭代器可以遍歷出我們hdfs文件系統根目錄下的文件 while (listFiles.hasNext()) { LocatedFileStatus fileStatus = listFiles.next(); Path path = fileStatus.getPath(); String filename = path.getName(); //只顯示文件,不顯示文件夾 System.out.println(filename); //塊大小 System.out.println(fileStatus.getBlockSize()); //權限 System.out.println(fileStatus.getPermission()); //組 System.out.println(fileStatus.getGroup()); //長度 System.out.println(fileStatus.getLen()); BlockLocation[] blockLocations = fileStatus.getBlockLocations(); for(BlockLocation bl : blockLocations) { //位置的角標 System.out.println("block-offset: " + bl.getOffset()); //塊的位置 String[] hosts = bl.getHosts(); for(String host:hosts) { System.out.println(host); } } } } } ~~~ # 判斷是否是文件 ~~~ Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME","root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //判斷是否是文件還是文件夾 FileStatus[] listStatus = fs.listStatus(new Path("/")); for(FileStatus fileStatus:listStatus) { //如果是文件 if(fileStatus.isFile()) { System.out.println(fileStatus.getPath().getName()); } } ~~~ # 本地copy到hdfs ![](images/screenshot_1523194120823.png) 發生這個問題,是你代碼執行的環境是一個用戶名和其他組,但是hdfs有自己的用戶名和組別,需要你這邊偽造下,不然就取你電腦上的用戶名和組別,就會發生權限問題代碼 ~~~ package com.hdfs; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.io.IOException; public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME","root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //把本地文件copy到hdfs fs.copyFromLocalFile(new Path("/Users/jdxia/Desktop/website/i.txt"), new Path("/")); fs.close(); } } ~~~ # hdfscopy到本地 copy到本地的話,也涉及到本地的一些權限代碼 ~~~ package com.hdfs; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.io.IOException; public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //把遠程的文件copy到本地 //第一個參數是否刪除原文件 //第四個參數表示需要不需要要用本地文件系統,開啟文件校驗 fs.copyToLocalFile(false, new Path("/i.txt"), new Path("/Users/jdxia/Desktop/website/hdfs"), true); fs.close(); } } ~~~ # windows開發注意建議在linux下進行hadoop應用的開發，不會存在兼容性問題。如在window上做客戶端應用開發，需要設置以下環境： * 用windows平臺下編譯的hadoop安裝包解壓一份到windows的任意一個目錄下 * 在window系統中配置HADOOP_HOME指向你解壓的安裝包目錄 * 在windows系統的path變量中加入HADOOP_HOME的bin目錄 # api流進行文件上傳 ~~~ package com.hdfs; 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; import java.io.FileInputStream; import java.io.IOException; public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //輸入流 FileInputStream in = new FileInputStream("/Users/jdxia/Desktop/website/hdfs/hello.txt"); //輸出流 Path path = new Path("/hello.txt"); FSDataOutputStream out = fs.create(path); //commons包的工具 IOUtils.copy(in,out); //或者這樣 //IOUtils.copyBytes(in,out, conf); //關閉資源 IOUtils.closeStream(in); IOUtils.closeStream(out); } } ~~~ # api流進行文件下載 ~~~ package com.hdfs; import org.apache.commons.io.IOUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置也可以這樣 //FileSystem fs = FileSystem.get(new URI("hdfs://master:9000"), conf, "root"); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); FSDataInputStream in = fs.open(new Path("/hello.txt")); FileOutputStream out = new FileOutputStream(new File("/Users/jdxia/Desktop/website/hdfs/hello.txt")); IOUtils.copy(in,out); fs.close(); } } ~~~ # 目錄操作 ## 創建目錄 ~~~ public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //目錄操作 fs.mkdirs(new Path("/nihao/hello")); fs.close(); } } ~~~ ## 刪除夾 ~~~ public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //目錄操作,如果是非空文件夾,第二個參數表示遞歸刪除,要給true //一直會把hello也刪除 fs.delete(new Path("/nihao/hello"),true); fs.close(); } } ~~~ ## 重命名 ~~~ public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //重命名文件或文件夾 fs.rename(new Path("/nihao"),new Path("/lihao")); fs.close(); } } ~~~ ## 查看目錄下的文件信息顯示文件不顯示文件夾 ~~~ package com.hdfs; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.*; import java.io.IOException; public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //第一個參數是路徑,第二個參數是遞歸,false表示不需要 //獲取/下的文件 RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true); //通過這個迭代器可以遍歷出我們hdfs文件系統根目錄下的文件 while (listFiles.hasNext()) { LocatedFileStatus fileStatus = listFiles.next(); System.out.println(fileStatus.getPath().getName()); System.out.println(fileStatus.getBlockSize()); System.out.println(fileStatus.getPermission()); System.out.println(fileStatus.getLen()); BlockLocation[] blockLocations = fileStatus.getBlockLocations(); for (BlockLocation bl : blockLocations) { System.out.println("block-length: "+bl.getLength()+"---"+"block-offset"+bl.getOffset()); String[] hosts = bl.getHosts(); for (String host:hosts) { System.out.println(host); } } System.out.println("------------"); } fs.close(); } } ~~~ ## 查看文件及文件夾信息 ~~~ package com.hdfs; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.*; import java.io.IOException; public class TestHDFS { public static void main(String[] args) throws IOException { Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME", "root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); FileStatus[] listStatus = fs.listStatus(new Path("/")); String flag = ""; for (FileStatus fstatus : listStatus) { if (fstatus.isFile()) { flag = "f-- "; } else { flag = "d-- "; } System.out.println(flag + fstatus.getPath().getName()); System.out.println(fstatus.getPermission()); } fs.close(); } } ~~~ # 精準讀取 ~~~ //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //先獲取一個文件的輸入流----針對hdfs上的 FSDataInputStream in = fs.open(new Path("/iloveyou.txt")); //可以將流的起始偏移量進行自定義 in.seek(22); //再構造一個文件的輸出流----針對本地的 FileOutputStream out = new FileOutputStream(new File("d:/iloveyou.line.2.txt")); //第三個參數表示要讀取多大,第四個參數表示要關閉這個流 IOUtils.copyBytes(in,out, 19L, true); ~~~ # 讀取指定block ~~~ public void testCat() throws IllegalArgumentException, IOException{ FSDataInputStream in = fs.open(new Path("/weblog/input/access.log.10")); //拿到文件信息 FileStatus[] listStatus = fs.listStatus(new Path("/weblog/input/access.log.10")); //獲取這個文件的所有block的信息 BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(listStatus[0], 0L, listStatus[0].getLen()); //第一個block的長度 long length = fileBlockLocations[0].getLength(); //第一個block的起始偏移量 long offset = fileBlockLocations[0].getOffset(); System.out.println(length); System.out.println(offset); //獲取第一個block寫入輸出流 // IOUtils.copyBytes(in, System.out, (int)length); byte[] b = new byte[4096]; FileOutputStream os = new FileOutputStream(new File("d:/block0")); while(in.read(offset, b, 0, 4096)!=-1){ os.write(b); offset += 4096; if(offset>length) return; }; os.flush(); os.close(); in.close(); } ~~~ # 一致性模型 ## 簡介 ![](https://box.kancloud.cn/a6e3b280aa9930553a2328f6b25c2d05_1160x538.png) 如果同時發生寫那邊不刷新的話,會等讀完了再寫可以設置寫一個讀一個 ## 例子刷新機制可以保證數據可靠性,影響性能默認是關閉后才可靠 ~~~ Configuration conf = new Configuration(); //設置默認的文件系統 conf.set("fs.defaultFS", "hdfs://master:9000"); //設置權限,也可以在vm那邊偽造 System.setProperty("HADOOP_USER_NAME","root"); //首先需要一個hdfs的客戶端對象 FileSystem fs = FileSystem.get(conf); //獲取輸出流,hdfs路徑 FSDataOutputStream fos = fs.create(new Path("/hello.txt")); //寫數據 fos.write("x".getBytes); //刷新,立即生效,沒有的話,不是立即生效 //可把這行去掉斷點調試看看 fos.hflush(); //關閉資源 IOUtils.closeStream(fos); fs.close(); ~~~ # hadoop的RPC框架使用 ## 協議 ClientNameNodeProtocal ~~~ package com.hadooprpc.protocal; public interface ClientNameNodeProtocal { //版本號 public static final long versionID = 1L; public String getMetaData(String path); } ~~~ ## server端 NameNode ~~~ package com.hadooprpc.server; import com.hadooprpc.protocal.ClientNameNodeProtocal; //服務被調用實現方 public class NameNode implements ClientNameNodeProtocal { //返回元數據信息 @Override public String getMetaData(String path) { //path是文件路徑,備份數是2,備份快blk_1,blk_2 //blk_1在master,slave節點上 //blk_2在master,slave節點上 return path + "2 {blk_1,blk_2} {blk_1:master,slave}{blk2:master.slave}"; } } ~~~ ServerPublisher ~~~ package com.hadooprpc.server; import com.hadooprpc.protocal.ClientNameNodeProtocal; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.ipc.RPC; import java.io.IOException; //服務發布 public class ServerPublisher { public static void main(String[] args) throws IOException { //所有rpc的配置都是通過new Configuration()來的,我們這給個空的配置 RPC.Builder builder = new RPC.Builder(new Configuration()); //setProtocol是設置協議 //setInstance設置實現類 builder.setBindAddress("localhost").setPort(8787).setProtocol(ClientNameNodeProtocal.class).setInstance(new NameNode()); //然后我們把他發布 RPC.Server server = builder.build(); //把服務啟動起來 server.start(); } } ~~~ ## client端 HDFSClient ~~~ package com.hadooprpc.client; import com.hadooprpc.protocal.ClientNameNodeProtocal; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.ipc.RPC; import java.io.IOException; import java.net.InetSocketAddress; public class HDFSClient { public static void main(String[] args) throws IOException { //這邊獲取namenode ClientNameNodeProtocal namenode = RPC.getProxy(ClientNameNodeProtocal.class, 1L, new InetSocketAddress("localhost", 8787), new Configuration()); //可以把遠程的namenode當成本地使用,這就是rpc的使用 //獲取這個path的元數據信息 String metaData = namenode.getMetaData("/a.txt"); System.out.println(metaData); } } ~~~~ ## rpc通信源碼 **HDFS相關** ~~~ ClientDatanodeProtocol : 一個客戶端和datanode之間的協議接口,用于數據塊恢復 ClientProtocol : client和Namenode交互的接口,所有控制流的請求均在這里,如創建文件,刪除文件等 DatanodeProtocol : Datanode與Namenode交互的接口,如心跳,blockreport等 NamenodeProtocol : SecondaryNode與Namenode交互的接口 ~~~ **Mapreduce相關** ~~~ InterDatanodeProtocol : Datanode內部交互的接口,用來更新block的元數據 InnerTrackerProtocol : TaskTracker與JobTracker交互的接口,功能與DatanodeProtocol相似 JobSubmissionProtocol : JobClient與JobTracker交互的接口,用來提交Job,獲得Job等與Job相關的操作 TaskUmbilicalProtocol : Task中子進程與母進程交互的接口,子進程即map,reduce等操作,母進程即TaskTracker,該接口可以回報子進程的運行狀態 ~~~