HDFS操作案例

本文目录导读：

1. HDFS Shell命令操作
2. Java API操作案例
3. 实用技巧

我来为您提供详细的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的常见操作,您可以根据实际需求调整路径和参数，如需了解更多特定场景的实现，请告诉我您的具体需求。