HDFS命令和java API 实验步骤
搭建Hadoop伪分布式模式、或者完全分布式模式Windows里安装配置JDK+Eclipse+Maven在Eclipse里新建Maven Project，新建包，新建class编程上述4个Java程序虚拟机的namenode主机上，启动Hadoop在eclipse里运行上述4个Java程序使用hdfs的shell命令查看运行结果使用web console查看运行结果 Linux的shell命令 常用的linux的shell命令： cd /aa 转到/aa目录 pws 显示当前目录 ll /aa 显示/aa目录 mkdir /aa 新建/aa目录 rm /aa/out.txt 删除/aa目录下的out.txt文件 cat /aa/out.txt 查看/aa目录下的out.txt文件的内容 vi /aa/out.txt 编辑/aa目录下的out.txt文件的内容 ifconfig 查看IP地址 ip addr 查看IP地址 hdfs的shell命令 hdfs的shell命令，有三种形式，用前面两个。 hdfs dfs √ hadoop fs √ hadoop dfs × 常用的hdfs的shell命令： hdfs dfs -ls / 查看hdfs根目录下文件和目录 hdfs dfs -ls -R / 查看hdfs根目录下包括子目录在内的所有文件和目录 hdfs dfs -mkdir /aa/bb 在hdfs的/aa目录下新建/bb目录 hdfs dfs -rm -r /aa/bb 删除hdfs的/aa目录下的/bb目录 hdfs dfs -rm /aa/out.txt 删除hdfs的/aa目录下的out.txt文件 hdfs dfs -put anaconda-ks.cfg /aa 把本地文件上传到hdfs hdfs dfs -copyFromLocal a.txt / 把本地文件上传到hdfs hdfs dfs -get /bb.txt bbcopy.txt 从hdfs下载文件到本地 hdfs dfs -copyToLocal /bb.txt bbcopy.txt 从hdfs下载文件到本地 1、获取 HDFS 文件系统 public static FileSystem getFileSystem() throws IOException, URISyntaxException{ //1、获取配置文件 Configuration conf = new Configuration();// FileSystem fs = FileSystem.get(conf);//集群环境下直接获取默认文件系统 //2、指定的文件系统地址 URI path = new URI(“hdfs://test:9000”); //3、返回指定的文件系统地址 ------本地测试使用此方法 FileSystem fs = FileSystem.get(path, conf); //4、此处关闭文件系统流会报错，导致之后的操作无法进行 /** * fs.close(); * Exception in thread “main” java.io.IOException: Filesystem closed at org.apache.hadoop.hdfs.DFSClient.checkOpen(DFSClient.java:795) at org.apache.hadoop.hdfs.DFSClient.primitiveMkdir(DFSClient.java:2743) at org.apache.hadoop.hdfs.DFSClient.mkdirs(DFSClient.java:2724) at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:870) at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:866) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.mkdirsInternal(DistributedFileSystem.java:866) at org.apache.hadoop.hdfs.DistributedFileSystem.mkdirs(DistributedFileSystem.java:859) at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:1817) at com.dajiangtai.hdfsJavaAPI.FileTest.mkdir(FileTest.java:33) at com.dajiangtai.hdfsJavaAPI.FileTest.main(FileTest.java:46)
* * */ return fs; } 如果放到 hadoop 集群上面运行，获取文件系统可以直接使用 FileSystem.get(conf)。2、创建文件目录 //创建文件目录 public static void mkdir() throws IOException, URISyntaxException{
//1、调用刚刚获得的HDFS文件系统，返回文件系统对象 FileSystem fs = getFileSystem(); //2、指定创建的文件系统路径 fs.mkdirs(new Path("hdfs://test:9000/filetest")); //3、关闭文件流--------此处不关闭流的情况下，创建文件会正常运行，且不报错 fs.close(); }3、删除文件或文件目录 //删除文件或者文件目录 public static void rmdir() throws Exception {
//返回FileSystem对象 FileSystem fs = getFileSystem(); //删除文件或者文件目录 delete(Path f) 此方法已经弃用 fs.delete(new Path("hdfs://cloud004:9000/middle/weibo"),true); //释放资源 fs.close();} 4、获取目录下的所有文件 //获取目录下的所有文件 public static void getAllFile() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); //2、获取该目录下的文件名称—FileStatus获取文件的元数据 FileStatus[] listStatus = fs.listStatus(new Path(“hdfs://test:9000/filetest/”)); //3.获取文件的path路径 Path[] stat2Paths = FileUtil.stat2Paths(listStatus); //4、循环遍历路径，输出结果 for(Path path: stat2Paths){ System.out.println(path); } fs.close(); /**打印结果 * hdfs://test:9000/filetest/file0.txt hdfs://test:9000/filetest/file1.txt hdfs://test:9000/filetest/file2.txt hdfs://test:9000/filetest/file3.txt hdfs://test:9000/filetest/file4.txt hdfs://test:9000/filetest/file5.txt * / } 5、文件上传至 HDFS //上传文件到hdfs public static void copyToHdfs() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); / * Exception in thread “main” java.io.FileNotFoundException: File does not exist: /home/hadoop/app/hadoop-2.6.0/filetest/copyToHdfs.txt * * Exception in thread “main” java.lang.IllegalArgumentException: Pathname /C:/Users/yizhijing/Desktop/llllllll/file1.txt from C:/Users/yizhijing/Desktop/llllllll/file1.txt is not a valid DFS filename. * */ //本地文件路径,并且集群中同文件目录下不存在同名文件的情况下 //2、获取本地文件路径------在windows环境下就是wondows路径，在Linux环境下就是linux路径eg:/home/hadoop/app/hadoop-2.6.0/filetest/file.txt Path path = new Path(“D://file.txt”); //3、获取集群文件路径 Path path2 = new Path(“hdfs://test:9000/filetest”); //4、调用文件系统的方法上传文件 fs.copyFromLocalFile(path, path2); //5、关闭文件流 fs.close(); } 6、从 HDFS 下载文件 //从hdfs下载文件到到本地windows系统 public static void copyToLocal() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); //2、指定下载的hdfs文件路径 Path path = new Path(“hdfs://test:9000/filetest/file.txt”); //3、指定本地接收的文件路径 Path path2 = new Path(“F://软件all/微信/WeChat Files/WeChat Files/l1067021152/Files”); //4、调用文件系统的下载文件系统方法 fs.copyToLocalFile(path, path2); //5、关闭文件流 fs.close(); } 7、获取 HDFS 集群节点信息 //获取hdfs节点信息 public static void getNodeMessage() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); //2、获取分布式文件系统对象 DistributedFileSystem hdfs = (DistributedFileSystem)fs; //3、获取分布式系统所有节点信息 DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats(); //4、打印节点 for(int i = 0;i<dataNodeStats.length;i++){ System.out.println(“datanode_”+i+“name:”+dataNodeStats[i].getHostName()); } } 8、查找某个文件在 HDFS 集群的位置 //查询某个文件在集群中的位置信息 public static void getFileMessage() throws IOException, URISyntaxException{ //1、获取分布式文件系统 FileSystem fs = getFileSystem(); //2、获取某文件的路径 Path path = new Path(“hdfs://test:9000/filetest/file.txt”); //3、获取某文件的目录 FileStatus fileStatus = fs.getFileStatus(path); //4、获取文件块位置列表 BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen()); //5、循环输出块信息 for (int i = 0; i < fileBlockLocations.length; i++) { String[] hosts = fileBlockLocations[i].getHosts(); System.out.println(“block”+i+“location”+hosts[0]); } }
上述 Java API对 HDFS 的操作方法，完整代码如下所示：import java.io.IOException; import java.net.URI; import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.BlockLocation; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FileUtil; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
public class FileTest { //获取HDFS文件系统 public static FileSystem getFileSystem() throws IOException, URISyntaxException{ //1、获取配置文件 Configuration conf = new Configuration();
// FileSystem fs = FileSystem.get(conf);//集群环境下直接获取默认文件系统 //2、指定的文件系统地址 URI path = new URI(“hdfs://test:9000”); //3、返回指定的文件系统地址 ------本地测试使用此方法 FileSystem fs = FileSystem.get(path, conf); /** * fs.close(); * Exception in thread “main” java.io.IOException: Filesystem closed at org.apache.hadoop.hdfs.DFSClient.checkOpen(DFSClient.java:795) at org.apache.hadoop.hdfs.DFSClient.primitiveMkdir(DFSClient.java:2743) at org.apache.hadoop.hdfs.DFSClient.mkdirs(DFSClient.java:2724) at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:870) at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:866) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.mkdirsInternal(DistributedFileSystem.java:866) at org.apache.hadoop.hdfs.DistributedFileSystem.mkdirs(DistributedFileSystem.java:859) at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:1817) at com.dajiangtai.hdfsJavaAPI.FileTest.mkdir(FileTest.java:33) at com.dajiangtai.hdfsJavaAPI.FileTest.main(FileTest.java:46)
* * */ return fs; } //创建文件目录 public static void mkdir() throws IOException, URISyntaxException{ //1、调用刚刚获得的HDFS文件系统，返回文件系统对象 FileSystem fs = getFileSystem(); //2、指定创建的文件系统路径 fs.mkdirs(new Path("hdfs://test:9000/filetest")); for(int i=0;i<6;i++){ Path path = new Path("hdfs://test:9000/filetest/file"+i+".txt"); fs.create(path); } //3、关闭文件流 fs.close(); } //删除文件操作 public static void rmFile() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); //删除指定的文件系统路径 fs.delete(new Path("hdfs://test:9000/filetest"),true); //3、关闭文件流 fs.close(); } //获取目录下的所有文件 public static void getAllFile() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); //2、获取该目录下的文件名称---FileStatus获取文件的元数据 FileStatus[] listStatus = fs.listStatus(new Path("hdfs://test:9000/filetest/")); //3.获取文件的path路径 Path[] stat2Paths = FileUtil.stat2Paths(listStatus); //4、循环遍历路径，输出结果 for(Path path: stat2Paths){ System.out.println(path); } fs.close(); /**打印结果 * hdfs://test:9000/filetest/file0.txthdfs://test:9000/filetest/file1.txt hdfs://test:9000/filetest/file2.txt hdfs://test:9000/filetest/file3.txt hdfs://test:9000/filetest/file4.txt hdfs://test:9000/filetest/file5.txt * */ }
//上传文件到hdfs public static void copyToHdfs() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); /* * Exception in thread "main" java.io.FileNotFoundException: File does not exist: /home/hadoop/app/hadoop-2.6.0/filetest/copyToHdfs.txt * * Exception in thread "main" java.lang.IllegalArgumentException: Pathname /C:/Users/yizhijing/Desktop/llllllll/file1.txt from C:/Users/yizhijing/Desktop/llllllll/file1.txt is not a valid DFS filename. * */ //本地文件路径,并且集群中同文件目录下不存在同名文件的情况下 //2、获取本地文件路径 Path path = new Path("D://file.txt"); //3、获取集群文件路径 Path path2 = new Path("hdfs://test:9000/filetest"); //4、调用文件系统的方法上传文件 fs.copyFromLocalFile(path, path2); //5、关闭文件流 fs.close(); } //从hdfs下载文件到到本地windows系统 public static void copyToLocal() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); //2、指定下载的hdfs文件路径 Path path = new Path("hdfs://test:9000/filetest/file.txt"); //3、指定本地接收的文件路径 Path path2 = new Path("F://软件all/微信/WeChat Files/WeChat Files/l1067021152/Files"); //4、调用文件系统的下载文件系统方法 fs.copyToLocalFile(path, path2); //5、关闭文件流 fs.close(); } //获取hdfs节点信息 public static void getNodeMessage() throws IOException, URISyntaxException{ //1、获取文件系统对象 FileSystem fs = getFileSystem(); //2、获取分布式文件系统对象 DistributedFileSystem hdfs = (DistributedFileSystem)fs; //3、获取分布式系统所有节点信息 DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats(); //4、打印节点 for(int i = 0;i<dataNodeStats.length;i++){ System.out.println("datanode_"+i+"name:"+dataNodeStats[i].getHostName()); } } //查询某个文件在集群中的位置信息 public static void getFileMessage() throws IOException, URISyntaxException{ //1、获取分布式文件系统 FileSystem fs = getFileSystem(); //2、获取某文件的路径 Path path = new Path("hdfs://test:9000/filetest/file.txt"); //3、获取某文件的目录 FileStatus fileStatus = fs.getFileStatus(path); //4、获取文件块位置列表 BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen()); //5、循环输出块信息 for (int i = 0; i < fileBlockLocations.length; i++) { String[] hosts = fileBlockLocations[i].getHosts(); System.out.println("block"+i+"location"+hosts[0]); } } public static void main(String[] args) throws IOException, URISyntaxException { // getFileSystem(); // mkdir(); // rmFile(); // getAllFile(); // copyToHdfs(); // copyToLocal(); // getNodeMessage(); getFileMessage(); } }运行程序 以上 Java API 操作 HDFS 的各种方法，在本地测试完成后，可以根据自己实际情况，稍加修改Path路径就可以直接放到 Hadoop 环境下运行。大致需要以下几步： 第一步：我们使用 myEclipse 将 FileTest.java 类打包为 FileTest.jar。因为这里不涉及第三方 jar 包，而且 Hadoop 集群上面已经存在 hadoop 所需要的jar包，所以我们只需要打包 FileTest.java 即可。 FileTest.jar 第二步：在hadoop用户下，通过 xshell 客户端，将 test.jar 上传至 hadoop服务器本地目录/home/hadoop/app/hadoop-2.6.0/filetest/的下面。 [hadoop@test djt]$ ls FileTest.jar 第3步：我们需要切换到 hadoop 的安装目录bin下来运行 FileTest.jar 文件，否则无法找到需要的 jar包来执行程序。 [hadoop@testhadoop-2.2.0-x64]$ hadoop jar /home/hadoop/djt/FileTest.jar
