掌握HDFS的Java API接口访问

1 /**
2 * 获取文件系统
3 *
4 * @return FileSystem
5 */
6 public static FileSystem getFileSystem() {
7 //读取配置文件
8 Configuration conf = new Configuration();
9 // 文件系统
10 FileSystem fs = null;
11
12 String hdfsUri = HDFSUri;
13 if(StringUtils.isBlank(hdfsUri)){
14 // 返回默认文件系统 如果在 Hadoop集群下运行，使用此种方法可直接获取默认文件系统
15 try {
16 fs = FileSystem.get(conf);
17 } catch (IOException e) {
18 logger.error("", e);
19 }
20 }else{
21 // 返回指定的文件系统,如果在本地测试，需要使用此种方法获取文件系统
22 try {
23 URI uri = new URI(hdfsUri.trim());
24 fs = FileSystem.get(uri,conf);
25 } catch (URISyntaxException | IOException e) {
26 logger.error("", e);
27 }
28 }
29
30 return fs;
31 }

1 /**
2 * 创建文件目录
3 *
4 * @param path
5 */
6 public static void mkdir(String path) {
7 try {
8 // 获取文件系统
9 FileSystem fs = getFileSystem();
10
11 String hdfsUri = HDFSUri;
12 if(StringUtils.isNotBlank(hdfsUri)){
13 path = hdfsUri + path;
14 }
15
16 // 创建目录
17 fs.mkdirs(new Path(path));
18
19 //释放资源
20 fs.close();
21 } catch (IllegalArgumentException | IOException e) {
22 logger.error("", e);
23 }
24 }

1 /**
2 * 删除文件或者文件目录
3 *
4 * @param path
5 */
6 public static void rmdir(String path) {
7 try {
8 // 返回FileSystem对象
9 FileSystem fs = getFileSystem();
10
11 String hdfsUri = HDFSUri;
12 if(StringUtils.isNotBlank(hdfsUri)){
13 path = hdfsUri + path;
14 }
15
16 // 删除文件或者文件目录 delete(Path f) 此方法已经弃用
17 fs.delete(new Path(path),true);
18
19 // 释放资源
20 fs.close();
21 } catch (IllegalArgumentException | IOException e) {
22 logger.error("", e);
23 }
24 }

1 /**
2 * 根据filter获取目录下的文件
3 *
4 * @param path
5 * @param pathFilter
6 * @return String[]
7 */
8 public static String[] ListFile(String path,PathFilter pathFilter) {
9 String[] files = new String[0];
10
11 try {
12 // 返回FileSystem对象
13 FileSystem fs = getFileSystem();
14
15 String hdfsUri = HDFSUri;
16 if(StringUtils.isNotBlank(hdfsUri)){
17 path = hdfsUri + path;
18 }
19
20 FileStatus[] status;
21 if(pathFilter != null){
22 // 根据filter列出目录内容
23 status = fs.listStatus(new Path(path),pathFilter);
24 }else{
25 // 列出目录内容
26 status = fs.listStatus(new Path(path));
27 }
28
29 // 获取目录下的所有文件路径
30 Path[] listedPaths = FileUtil.stat2Paths(status);
31 // 转换String[]
32 if (listedPaths != null && listedPaths.length > 0){
33 files = new String[listedPaths.length];
34 for (int i = 0; i < files.length; i++){
35 files[i] = listedPaths[i].toString();
36 }
37 }
38 // 释放资源
39 fs.close();
40 } catch (IllegalArgumentException | IOException e) {
41 logger.error("", e);
42 }
43
44 return files;
45 }

1 /**
2 * 文件上传至 HDFS
3 *
4 * @param delSrc
5 * @param overwrite
6 * @param srcFile
7 * @param destPath
8 */
9 public static void copyFileToHDFS(boolean delSrc, boolean overwrite,String srcFile,String destPath) {
10 // 源文件路径是Linux下的路径，如果在 windows 下测试，需要改写为Windows下的路径，比如D://hadoop/djt/weibo.txt
11 Path srcPath = new Path(srcFile);
12
13 // 目的路径
14 String hdfsUri = HDFSUri;
15 if(StringUtils.isNotBlank(hdfsUri)){
16 destPath = hdfsUri + destPath;
17 }
18 Path dstPath = new Path(destPath);
19
20 // 实现文件上传
21 try {
22 // 获取FileSystem对象
23 FileSystem fs = getFileSystem();
24 fs.copyFromLocalFile(srcPath, dstPath);
25 fs.copyFromLocalFile(delSrc,overwrite,srcPath, dstPath);
26 //释放资源
27 fs.close();
28 } catch (IOException e) {
29 logger.error("", e);
30 }
31 }

1 /**
2 * 从 HDFS 下载文件
3 *
4 * @param srcFile
5 * @param destPath
6 */
7 public static void getFile(String srcFile,String destPath) {
8 // 源文件路径
9 String hdfsUri = HDFSUri;
10 if(StringUtils.isNotBlank(hdfsUri)){
11 srcFile = hdfsUri + srcFile;
12 }
13 Path srcPath = new Path(srcFile);
14
15 // 目的路径是Linux下的路径，如果在 windows 下测试，需要改写为Windows下的路径，比如D://hadoop/djt/
16 Path dstPath = new Path(destPath);
17
18 try {
19 // 获取FileSystem对象
20 FileSystem fs = getFileSystem();
21 // 下载hdfs上的文件
22 fs.copyToLocalFile(srcPath, dstPath);
23 // 释放资源
24 fs.close();
25 } catch (IOException e) {
26 logger.error("", e);
27 }
28 }

1 /**
2 * 获取 HDFS 集群节点信息
3 *
4 * @return DatanodeInfo[]
5 */
6 public static DatanodeInfo[] getHDFSNodes() {
7 // 获取所有节点
8 DatanodeInfo[] dataNodeStats = new DatanodeInfo[0];
9
10 try {
11 // 返回FileSystem对象
12 FileSystem fs = getFileSystem();
13
14 // 获取分布式文件系统
15 DistributedFileSystem hdfs = (DistributedFileSystem)fs;
16
17 dataNodeStats = hdfs.getDataNodeStats();
18 } catch (IOException e) {
19 logger.error("", e);
20 }
21 return dataNodeStats;
22 }

1 /**
2 * 查找某个文件在 HDFS集群的位置
3 *
4 * @param filePath
5 * @return BlockLocation[]
6 */
7 public static BlockLocation[] getFileBlockLocations(String filePath) {
8 // 文件路径
9 String hdfsUri = HDFSUri;
10 if(StringUtils.isNotBlank(hdfsUri)){
11 filePath = hdfsUri + filePath;
12 }
13 Path path = new Path(filePath);
14
15 // 文件块位置列表
16 BlockLocation[] blkLocations = new BlockLocation[0];
17 try {
18 // 返回FileSystem对象
19 FileSystem fs = getFileSystem();
20 // 获取文件目录
21 FileStatus filestatus = fs.getFileStatus(path);
22 //获取文件块位置列表
23 blkLocations = fs.getFileBlockLocations(filestatus, 0, filestatus.getLen());
24 } catch (IOException e) {
25 logger.error("", e);
26 }
27 return blkLocations;
28 }

1 /**
2 * 文件重命名
3 *
4 * @param srcPath
5 * @param dstPath
6 */
7 public boolean rename(String srcPath, String dstPath){
8 boolean flag = false;
9 try {
10 // 返回FileSystem对象
11 FileSystem fs = getFileSystem();
12
13 String hdfsUri = HDFSUri;
14 if(StringUtils.isNotBlank(hdfsUri)){
15 srcPath = hdfsUri + srcPath;
16 dstPath = hdfsUri + dstPath;
17 }
18
19 flag = fs.rename(new Path(srcPath), new Path(dstPath));
20 } catch (IOException e) {
21 logger.error("{} rename to {} error.", srcPath, dstPath);
22 }
23
24 return flag;
25 }

1 /**
2 * 判断目录是否存在
3 *
4 * @param srcPath
5 * @param dstPath
6 */
7 public boolean existDir(String filePath, boolean create){
8 boolean flag = false;
9
10 if (StringUtils.isEmpty(filePath)){
11 return flag;
12 }
13
14 try{
15 Path path = new Path(filePath);
16 // FileSystem对象
17 FileSystem fs = getFileSystem();
18
19 if (create){
20 if (!fs.exists(path)){
21 fs.mkdirs(path);
22 }
23 }
24
25 if (fs.isDirectory(path)){
26 flag = true;
27 }
28 }catch (Exception e){
29 logger.error("", e);
30 }
31
32 return flag;
33 }