Java API操作HDFS

Posted 2021-03-26 笙歌散尽，醉挽清风

package project.etl.core.util;

import java.io.FileNotFoundException;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;

/**
* 操作HDFS工具类
* @author snow
* @date: 2019-10-13 10:01:59
*/
public class HDFSUtil {
　　/**
　　* 获取默认文件系统
　　* @return
　　* @throws IOException
　　* @author snow
　　* @date: 2019-10-13 10:16:18
　　*/
　　public static FileSystem getFileSystem() throws IOException {
　　　　Configuration conf = new Configuration();
　　　　FileSystem fileSystem = FileSystem.get(conf);
　　　　return fileSystem;
　　}
　　/**
　　* 获取文件系统时需要验证
　　* @param user 用户名
　　* @param keytab 验证文件路径
　　* @return FileSystem
　　* @throws IOException
　　* @author snow
　　* @date: 2019-10-13 10:14:06
　　*/
　　public static FileSystem getFileSystem(String user,String keytab) throws IOException {
　　　　Configuration conf = new Configuration();
　　　　UserGroupInformation.loginUserFromKeytab(user, keytab);
　　　　FileSystem fileSystem = FileSystem.get(conf);
　　　　return fileSystem;
　　}
　　/**
　　* 向文件系统指定路径上传单个文件或文件夹
　　* @param fileSystem
　　* @param HDFSPath
　　* @param localPath
　　* @throws IOException
　　* @author snow
　　* @date: 2019-10-13 10:27:21
　　*/
　　public static void uploadFileToHDFS(FileSystem fileSystem,Path HDFSPath,Path localPath) throws IOException {
　　　　if(!fileSystem.exists(HDFSPath)) {
　　　　　　boolean bool = fileSystem.mkdirs(HDFSPath);
　　　　　　if(bool) {
　　　　　　　　System.out.println("创建文件夹成功！");
　　　　　　}else {
　　　　　　　　System.out.println("创建文件夹失败！");
　　　　　　}
　　　　}
　　　　fileSystem.copyFromLocalFile(localPath, HDFSPath);
　　}
　　/**
　　* 下载文件系统指定路径的单个文件或文件夹
　　* @param fileSystem
　　* @param HDFSPath
　　* @param localPath
　　* @author snow
　　* @throws IOException
　　* @date: 2019-10-13 11:02:54
　　*/
　　public static void loadHDFSFile(FileSystem fileSystem,Path HDFSPath,Path localPath) throws IOException {
　　　　fileSystem.copyToLocalFile(HDFSPath, localPath);
　　}
　　/**
　　* 删除文件系统文件或文件夹
　　* @param fileSystem
　　* @param HDFSPath
　　* @throws IOException
　　* @author snow
　　* @date: 2019-10-13 11:33:22
　　*/
　　public static void delHDFSFile(FileSystem fileSystem,Path HDFSPath) throws IOException {
　　　　if(fileSystem.exists(HDFSPath)) {
　　　　　　//禁止递归删除文件
　　　　　　fileSystem.delete(HDFSPath, false);
　　　　}
　　}
　　/**
　　* 查看文件系统指定路径下的文件及目录
　　* @param fileSystem
　　* @param HDFSPath
　　* @throws FileNotFoundException
　　* @throws IOException
　　* @author snow
　　* @date: 2019-10-13 12:03:12
　　*/
　　public static void checkHDFSFile(FileSystem fileSystem,Path HDFSPath) throws FileNotFoundException, IOException {
　　　　FileStatus[] fileStatuses = fileSystem.listStatus(HDFSPath);
　　　　for (FileStatus fileStatus : fileStatuses) {
　　　　　　String isDir = fileStatus.isDirectory() ? "文件夹：:" : "文件：";
　　　　　　short replication = fileStatus.getReplication();
　　　　　　String path = fileStatus.getPath().toString();
　　　　　　System.out.println(isDir + " " + replication + " " + path);
　　　　}
　　}
　　/**
　　* 关闭连接
　　* @param fileSystem
　　* @throws IOException
　　* @author snow
　　* @date: 2019-10-13 12:05:44
　　*/
　　public static void close(FileSystem fileSystem) throws IOException {
　　　　fileSystem.close();
　　}
}