本文目录导读：
我来为您提供详细的HDFS操作案例,涵盖常用命令和Java API操作。
HDFS Shell命令操作
基础文件操作
# 创建目录 hdfs dfs -mkdir /user/hadoop hdfs dfs -mkdir -p /user/hadoop/test/data # 查看文件列表 hdfs dfs -ls / hdfs dfs -ls -R /user # 递归显示 # 上传文件到HDFS hdfs dfs -put /local/file.txt /user/hadoop/ hdfs dfs -copyFromLocal /local/file.txt /user/hadoop/ # 下载文件到本地 hdfs dfs -get /user/hadoop/file.txt /local/ hdfs dfs -copyToLocal /user/hadoop/file.txt /local/ # 查看文件内容 hdfs dfs -cat /user/hadoop/file.txt hdfs dfs -tail /user/hadoop/file.txt # 查看末尾数据 # 查看文件大小 hdfs dfs -du -h /user/hadoop hdfs dfs -du -s /user/hadoop # 汇总大小
文件管理操作
# 复制文件 hdfs dfs -cp /source/path /target/path # 移动文件 hdfs dfs -mv /source/path /target/path # 删除文件 hdfs dfs -rm /user/hadoop/file.txt hdfs dfs -rm -r /user/hadoop/dir # 递归删除目录 hdfs dfs -rm -skipTrash /path # 不经过回收站直接删除 # 修改文件权限 hdfs dfs -chmod 755 /user/hadoop/file.txt hdfs dfs -chown hadoop:hadoop /user/hadoop/file.txt # 查看文件信息 hdfs dfs -stat '%b %o %r %y %n' /user/hadoop/file.txt hdfs fsck /user/hadoop/file.txt # 检查文件健康状态
Java API操作案例
环境配置
<!-- Maven依赖 -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.3.4</version>
</dependency>
文件上传下载
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
public class HdfsOperation {
private static FileSystem fs;
// 初始化连接
public static void init() throws IOException {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://localhost:9000");
fs = FileSystem.get(conf);
}
// 上传文件
public static void uploadFile(String localPath, String hdfsPath) throws IOException {
Path src = new Path(localPath);
Path dst = new Path(hdfsPath);
fs.copyFromLocalFile(false, true, src, dst);
System.out.println("文件上传成功: " + hdfsPath);
}
// 下载文件
public static void downloadFile(String hdfsPath, String localPath) throws IOException {
Path src = new Path(hdfsPath);
Path dst = new Path(localPath);
fs.copyToLocalFile(false, src, dst, true);
System.out.println("文件下载成功: " + localPath);
}
}
目录操作
public class DirectoryOperation {
// 创建目录
public static void createDirectory(String path) throws IOException {
Path dir = new Path(path);
if (fs.mkdirs(dir)) {
System.out.println("目录创建成功: " + path);
}
}
// 递归列出目录
public static void listAllFiles(String path) throws IOException {
Path dir = new Path(path);
RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(dir, true);
while (iterator.hasNext()) {
LocatedFileStatus status = iterator.next();
System.out.println("文件: " + status.getPath()
+ " | 大小: " + status.getLen()
+ " | 副本数: " + status.getReplication());
BlockLocation[] blocks = status.getBlockLocations();
for (BlockLocation block : blocks) {
System.out.println(" 块位置: " + Arrays.toString(block.getHosts()));
}
}
}
// 判断文件/目录是否存在
public static boolean exists(String path) throws IOException {
return fs.exists(new Path(path));
}
}
操作
public class FileContentOperation {
// 读取文件内容（流式）
public static void readFileByStream(String hdfsPath) throws IOException {
Path path = new Path(hdfsPath);
FSDataInputStream in = fs.open(path);
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
String line = null;
while ((line = reader.readLine()) != null) {
System.out.println(line);
}
reader.close();
}
// 写入文件内容
public static void writeFile(String hdfsPath, String content) throws IOException {
Path path = new Path(hdfsPath);
FSDataOutputStream out = fs.create(path, true); // true表示覆盖
out.write(content.getBytes());
out.flush();
out.close();
System.out.println("内容写入成功");
}
// 高效读取（使用IOUtils）
public static void readFileByIOUtils(String hdfsPath) throws IOException {
Path path = new Path(hdfsPath);
FSDataInputStream in = fs.open(path);
ByteArrayOutputStream out = new ByteArrayOutputStream();
IOUtils.copyBytes(in, out, 4096);
String result = new String(out.toByteArray());
System.out.println(result);
IOUtils.closeStream(in);
IOUtils.closeStream(out);
}
}
完整操作示例
public class HdfsCompleteExample {
public static void main(String[] args) throws IOException {
// 1. 初始化
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://localhost:9000");
fs = FileSystem.get(conf);
// 2. 创建目录
String dirPath = "/user/test/data";
fs.mkdirs(new Path(dirPath));
System.out.println("目录创建成功");
// 3. 上传文件
String localFile = "/Users/hadoop/input.txt";
String hdfsPath = dirPath + "/input.txt";
fs.copyFromLocalFile(new Path(localFile), new Path(hdfsPath));
// 4. 查看文件状态
FileStatus status = fs.getFileStatus(new Path(hdfsPath));
System.out.println("文件大小: " + status.getLen());
System.out.println("修改时间: " + new Date(status.getModificationTime()));
// 5. 读取文件
FSDataInputStream in = fs.open(new Path(hdfsPath));
BufferedReader br = new BufferedReader(new InputStreamReader(in));
String line;
while ((line = br.readLine()) != null) {
System.out.println("读取内容: " + line);
}
br.close();
// 6. 修改权限
fs.setPermission(new Path(hdfsPath), new FsPermission((short) 0644));
System.out.println("权限修改成功");
// 7. 删除文件
fs.delete(new Path(hdfsPath), false);
System.out.println("文件删除成功");
fs.close();
}
}
批量处理操作
public class BatchOperation {
// 批量上传
public static void batchUpload(String localDir, String hdfsDir) throws IOException {
File localFile = new File(localDir);
File[] files = localFile.listFiles();
for (File file : files) {
if (file.isFile()) {
String localPath = file.getAbsolutePath();
String hdfsPath = hdfsDir + "/" + file.getName();
fs.copyFromLocalFile(new Path(localPath), new Path(hdfsPath));
System.out.println("上传: " + file.getName());
}
}
}
// 过滤文件
public static void listWithFilter(String hdfsPath, String regex) throws IOException {
Path dir = new Path(hdfsPath);
FileStatus[] statuses = fs.listStatus(dir, new PathFilter() {
@Override
public boolean accept(Path path) {
return path.getName().matches(regex);
}
});
for (FileStatus status : statuses) {
System.out.println("匹配文件: " + status.getPath().getName());
}
}
}
实际应用场景示例
public class HdfsApplication {
// 日志文件分析
public static void analyzeLogs(String logPath) throws IOException {
Path file = new Path(logPath);
FSDataInputStream in = fs.open(file);
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
Map<String, Integer> errorCount = new HashMap<>();
String line;
while ((line = reader.readLine()) != null) {
if (line.contains("ERROR") || line.contains("Exception")) {
String errorType = extractErrorType(line);
errorCount.merge(errorType, 1, Integer::sum);
}
}
reader.close();
System.out.println("日志分析结果:");
errorCount.forEach((k, v) -> System.out.println(k + ": " + v + "次"));
}
// 数据备份
public static void backupData(String sourcePath, String backupPath) throws IOException {
Path src = new Path(sourcePath);
Path dst = new Path(backupPath);
try {
fs.copyToLocalFile(src, dst);
Path temp = new Path(backupPath + "_temp");
fs.rename(dst, temp);
fs.rename(src, dst);
fs.rename(temp, src);
} catch (IOException e) {
System.out.println("备份失败: " + e.getMessage());
}
}
}
实用技巧
性能优化建议
// 使用Buffer提升IO效率
public static void efficientRead(String path) throws IOException {
int bufferSize = 128 * 1024; // 128KB
Path file = new Path(path);
FSDataInputStream in = fs.open(file, bufferSize);
byte[] buffer = new byte[bufferSize];
int readLength = -1;
while ((readLength = in.read(buffer)) != -1) {
// 处理数据
}
}
错误处理
public static void safeOperation(String path) {
try {
// 操作代码
if (!fs.exists(new Path(path))) {
throw new FileNotFoundException("文件不存在: " + path);
}
} catch (FileNotFoundException e) {
System.err.println("文件未找到: " + e.getMessage());
} catch (IOException e) {
System.err.println("IO异常: " + e.getMessage());
} finally {
// 资源清理
try {
if (fs != null) fs.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
这些案例涵盖了HDFS的常见操作,您可以根据实际需求调整路径和参数，如需了解更多特定场景的实现，请告诉我您的具体需求。
