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.txt3. 文件下载与重命名机制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()) { RemoteIteratorLocatedFileStatus iter fs.listFiles(hdfsPath, true); while (iter.hasNext()) { printFileMetadata(iter.next().getPath()); } } }Shell命令对照表信息类型Java APIShell命令文件内容FSDataInputStreamhdfs dfs -cat基础元数据FileStatushdfs dfs -ls递归列表listFiles(path, true)hdfs dfs -ls -R块位置信息getFileBlockLocationshdfs fsck -blocks5. 目录创建与删除操作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($20) 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.blocksize256M -put largefile /data/
