Hadoop 3.1.3 HDFS Java API 实战：10个核心文件操作与Shell命令对照实现
Hadoop 3.1.3 HDFS Java API 实战：10个核心文件操作与Shell命令对照实现
1. 环境准备与基础配置
在开始HDFS文件操作前，需要确保Hadoop环境已正确配置。以下是典型开发环境搭建步骤：
// 创建基础配置对象 Configuration conf = new Configuration(); // 设置HDFS访问地址（根据实际集群修改） conf.set("fs.defaultFS", "hdfs://namenode:8020"); // 获取文件系统实例 FileSystem fs = FileSystem.get(conf);对应的Shell环境检查命令：
# 检查HDFS服务状态 hdfs dfsadmin -report # 验证Java环境 java -version2. 文件上传操作对比
Java API实现（覆盖/追加策略）
// 上传文件（自动覆盖） public void uploadFile(Path localPath, Path hdfsPath, boolean overwrite) throws IOException { fs.copyFromLocalFile(false, overwrite, localPath, hdfsPath); } // 追加内容到现有文件 public void appendToFile(Path localPath, Path hdfsPath) throws IOException { FSDataOutputStream out = fs.append(hdfsPath); Files.copy(localPath, out); out.close(); }Shell命令对照
| 操作类型 | 命令示例 |
|---|---|
| 覆盖上传 | hdfs dfs -put -f local.txt /data/input |
| 追加内容 | hdfs dfs -appendToFile add.txt /data/existing.txt |
3. 文件下载与重命名机制
Java实现智能重命名
public void downloadWithRename(Path hdfsPath, Path localPath) throws IOException { if (Files.exists(localPath)) { int counter = 1; Path newPath = new Path(localPath + "." + counter); while (Files.exists(newPath)) { counter++; newPath = new Path(localPath + "." + counter); } fs.copyToLocalFile(hdfsPath, newPath); } else { fs.copyToLocalFile(hdfsPath, localPath); } }Shell命令方案
# 基础下载 hdfs dfs -get /data/file.txt ./local/ # 带重命名逻辑的脚本 if [ -f "./local/file.txt" ]; then suffix=$(date +%s) hdfs dfs -get /data/file.txt "./local/file_$suffix.txt" else hdfs dfs -get /data/file.txt ./local/ fi4. 文件内容查看与元数据获取
Java API元数据查询
public void printFileMetadata(Path hdfsPath) throws IOException { FileStatus status = fs.getFileStatus(hdfsPath); System.out.println("Path: " + status.getPath()); System.out.println("Permission: " + status.getPermission()); System.out.println("Size: " + status.getLen() + " bytes"); System.out.println("Modification Time: " + new Date(status.getModificationTime())); // 递归列出目录内容 if (status.isDirectory()) { RemoteIterator<LocatedFileStatus> iter = fs.listFiles(hdfsPath, true); while (iter.hasNext()) { printFileMetadata(iter.next().getPath()); } } }Shell命令对照表
| 信息类型 | Java API | Shell命令 |
|---|---|---|
| 文件内容 | FSDataInputStream | hdfs dfs -cat |
| 基础元数据 | FileStatus | hdfs dfs -ls |
| 递归列表 | listFiles(path, true) | hdfs dfs -ls -R |
| 块位置信息 | getFileBlockLocations | hdfs fsck -blocks |
5. 目录创建与删除操作
Java实现智能目录管理
// 创建目录（自动创建父目录） public void createDirectory(Path hdfsPath) throws IOException { if (!fs.exists(hdfsPath.getParent())) { fs.mkdirs(hdfsPath.getParent()); } fs.mkdirs(hdfsPath); } // 删除目录（可选递归） public void deleteDirectory(Path hdfsPath, boolean recursive) throws IOException { if (fs.exists(hdfsPath)) { fs.delete(hdfsPath, recursive); } }Shell命令参考
# 创建多级目录 hdfs dfs -mkdir -p /data/project/{input,output} # 交互式删除非空目录 hdfs dfs -rm -r -i /data/old_project6. 文件移动与重命名
Java跨节点移动实现
public void moveFile(Path src, Path dst) throws IOException { // 检查目标目录是否存在 if (!fs.exists(dst.getParent())) { fs.mkdirs(dst.getParent()); } // 执行移动操作（原子性） fs.rename(src, dst); }Shell移动操作对比
# 基础移动命令 hdfs dfs -mv /data/old/loc.txt /data/new/ # 跨集群移动方案 hdfs dfs -get /cluster1/data.txt - | hdfs dfs -put - /cluster2/data.txt7. 自定义输入流实现
扩展FSDataInputStream实现按行读取：
public class HDFSLineReader extends FSDataInputStream { private BufferedReader reader; public HDFSLineReader(InputStream in) { super(in); this.reader = new BufferedReader(new InputStreamReader(in)); } public String readLine() throws IOException { return reader.readLine(); } // 使用示例 public static void readFileByLine(FileSystem fs, Path file) throws IOException { try (HDFSLineReader reader = new HDFSLineReader(fs.open(file))) { String line; while ((line = reader.readLine()) != null) { System.out.println(line); } } } }8. 文件存在性检查与安全操作
Java安全检查模式
public void safeFileOperations(Path path) throws IOException { // 检查路径是否存在 if (!fs.exists(path)) { throw new FileNotFoundException(path.toString()); } // 检查是否为目录 if (fs.getFileStatus(path).isDirectory()) { throw new IllegalArgumentException("Path must be a file"); } // 检查读写权限 if (!fs.access(path, FsAction.READ_WRITE)) { throw new AccessControlException("No read/write permission"); } }Shell安全检查技巧
# 检查文件存在性 hdfs dfs -test -e /path/to/file && echo "Exists" # 验证目录空状态 hdfs dfs -count -q /path/to/dir | awk '{if($2==0) print "Empty"}'9. 高级特性：文件合并与压缩
Java多文件合并
public void mergeFiles(Path[] srcFiles, Path dstFile) throws IOException { try (FSDataOutputStream out = fs.create(dstFile)) { for (Path src : srcFiles) { try (FSDataInputStream in = fs.open(src)) { IOUtils.copyBytes(in, out, conf, false); } } } }Shell合并方案对比
# 本地合并后上传 cat part-* > combined.txt && hdfs dfs -put combined.txt /output/ # 直接HDFS合并 hdfs dfs -getmerge /input/part-* merged.txt10. 性能优化实践
Java缓冲区优化配置
// 创建带缓冲区的输出流（256KB缓冲区） public void writeWithBuffer(Path file, byte[] data) throws IOException { int bufferSize = 256 * 1024; // 256KB try (FSDataOutputStream out = fs.create(file, true, // overwrite bufferSize, (short)3, // replication 128 * 1024 * 1024)) { // block size out.write(data); } }Shell性能调优参数
# 设置副本数为2（默认3） hdfs dfs -setrep -w 2 /data/hot_files # 调整块大小（需在写入前设置） hadoop fs -Ddfs.blocksize=256M -put largefile /data/