对HDFS进行文件操作的若干代码_头哥hdfs文件操作代码

import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class Optation {
    /**
     * HDFS 文件访问API
     */
    public static String hdfsUrl="hdfs://localhost:9000";
    /**
     * create HDFS folder 创建一个文件夹
     * @param dirPath
     * @return
     */
    public static void createDir(String dirPath) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path path = new Path(dirPath);
        hdfs.mkdirs(path);
        hdfs.close();
    }
    /**
     * delete HDFS folder 删除一个文件夹
     * @param dirPath
     * @return
     */
    public static void deleteDir(String dirPath) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path path = new Path(dirPath);
        hdfs.delete(path);
        hdfs.close();
    }
    /**
     * create a file 创建一个文件
     * @param filePath
     * @return
     */
    public static void createFile(String filePath) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path path = new Path(filePath);
        FSDataOutputStream out =hdfs.create(path);
        out.close();
        hdfs.close();
    }
    /**
     * rename a file 重命名一个文件
     * @param oldPath
     * @param newPath
     * @return
     */
    public static void renameFile(String oldPath,String newPath) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path path = new Path(oldPath);
        Path nPath = new Path(newPath);
        hdfs.rename(path, nPath);
        hdfs.close();
    }
    /**
     * delete a file 删除一个文件
     * @param hadoopFile
     * @return isDeleted
     */
    public static boolean deleteFile(String hadoopFile) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path path = new Path(hadoopFile);
        boolean isDeleted =hdfs.delete(path);
        hdfs.close();
        return isDeleted;
    }
    /**
     * upload a local file 上传文件
     * @param localPath
     * @param hadoopPath
     * @return
     */
    public static void uploadLocalFile(String localPath,String hadoopPath) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path src = new Path(localPath);
        Path dst =new Path(hadoopPath);
        hdfs.copyFromLocalFile(src, dst);
        hdfs.close();
    }
    /**
     * 读取文件于字节缓冲数组
     * @param hadoopFile
     * @return buffer
     */
    public static byte[] readFile(String hadoopFile) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path path = new Path(hadoopFile);
        if(hdfs.exists(path)){
            FSDataInputStream in = hdfs.open(path);
            FileStatus stat = hdfs.getFileStatus(path);
            byte[] buffer = new byte[Integer.parseInt(String.valueOf(stat.getBlockSize()))];
            in.readFully(0,buffer);
            in.close();
            hdfs.close();
            return buffer;
        }
        else{
            throw new Exception("the file is not found.");
        }
    }
    /**
     * list files under folder 列出文件夹中所有文件
     * @param hadoopPath
     * @return fileString
     */
    public static String listFiles(String hadoopPath) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path dst = new Path(hadoopPath);
        FileStatus[] files = hdfs.listStatus(dst);
        String fileString = "";
        for(FileStatus file:files){
            System.out.println(file.getPath().toString());
            fileString +=file.getPath().toString();
        }
        hdfs.close();
        return fileString;
    }
    /**
     * list block info of file 查找文件所在的数据块
     * @param hadoopPath
     * @return blockString
     */
    public static String getBlockInfo(String hadoopPath) throws Exception{
        Configuration conf = new Configuration();
        FileSystem hdfs =FileSystem.get(URI.create(hdfsUrl),conf);
        Path dst = new Path(hadoopPath);
        FileStatus fileStatus = hdfs.getFileStatus(dst);
        BlockLocation[] blkloc =hdfs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
        String blockString = "";
        for(BlockLocation loc:blkloc){
            for(int i=0;i<loc.getHosts().length;i++){
                System.out.println(loc.getHosts()[i]);
                // blockString +=loc.getHosts()[i]+" ";
            }
            hdfs.close();
            return blockString;
        }
    }
}
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
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155