hdfs客戶端操作 · 大數據

[TOC] # 列出文件 ~~~ 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); } } } ~~~ # 本地copy到hdfs ![](https://box.kancloud.cn/6a7e0c629cef32984d4d1f5280e7bdb8_1296x114.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到本地 //第一個參數是否刪除.第4個參數表示需要不需要要用本地文件系統 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); fs.close(); } } ~~~ # 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(); //設置默認的文件系統 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(); } } ~~~ # 精準讀取 ~~~ public void testRandomAccess() throws IllegalArgumentException, IOException{ //先獲取一個文件的輸入流----針對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(); } ~~~ # 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); } } ~~~~