HDFS命令和java API

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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)。

1
2
3
4
5
6
7
8
9
10
创建文件目录
    public static void mkdir() throws IOException, URISyntaxException{
    
        //1、调用刚刚获得的HDFS文件系统，返回文件系统对象
        FileSystem fs = getFileSystem();
        //2、指定创建的文件系统路径
        fs.mkdirs(new Path("hdfs://test:9000/filetest"));
        //3、关闭文件流--------此处不关闭流的情况下，创建文件会正常运行，且不报错
        fs.close();
    }

1
2
3
4
5
6
7
8
9
10
11
12
//删除文件或者文件目录
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();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
  //获取目录下的所有文件
    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
         * */
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
//上传文件到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();
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
//从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();
    }    

1
2
3
4
5
6
7
8
9
10
11
12
13
14
//获取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());
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//查询某个文件在集群中的位置信息
    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]);
        }
    }