【HDFS】Java_API使用

十里桃花舞丶

发布于 2021-09-10 11:18:44

4380

发布于 2021-09-10 11:18:44

文章被收录于专栏：桥路_大数据

Java API使用

环境初始化

首先完成Java开发环境准备，创建工程并导入开发所需的Jar包。之后在准备好的工程中完成以下步骤。

在IDE中新建一个类，类名为HDFSApp

在类中添加成员变量保存公共信息

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;

// 将代码中的{HDFS_HOST}:{HDFS_PORT}替换为HDFS的IP与端口，如192.168.31.41:9000
public class HDFSApp {
    public static final String HDFS_PATH="hdfs://{HDFS_HOST}:{HDFS_PORT}";
    FileSystem fileSystem = null;
    Configuration configuration = null;
}

在类中新增构造函数，初始化运行环境

public HDFSApp() throws Exception{
    this.configuration = new Configuration();
    this.fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "hadoop");
}

API基本使用

创建目录

任务：在HDFS上创建目录“/tmp/java_data”

// 添加方法mkdir()，方法中实现目录的创建
public void mkdir() throws Exception {
    fileSystem.mkdirs(new Path("/tmp/java_data"));
}

在main函数中执行测试：

// 创建Main函数，对方法进行测试
public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.mkdir();
}

回到shell工具中，使用shell命令查看是否执行成功。

hadoop fs -ls /tmp/

更改目录权限

任务：将HDFS目录“/tmp/java_data”的权限改为“rwxrwxrwx”

// 添加方法setPathPermission，方法中实现对目录的授权
public void setPathPermission() throws Exception {
        fileSystem.setPermission(new Path("/tmp/java_data"), new FsPermission("777"));
    }

在main函数中执行测试：

// 在Main函数中，对方法进行测试
public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.setPathPermission();
}

回到shell工具中，使用shell命令查看是否执行成功。

上传文件

任务：将本地文件“file.txt”上传到HDFS目录“/tmp/hdfs_data”目录中

// 在本地创建file.txt文件，文件中内容为hello word
// 添加方法copyFromLocalFile，方法中完成本地文件file.txt的上传
public void copyFromLocalFile() throws Exception {
        Path localPath = new Path("path to local file.txt");
        Path hdfsPath = new Path("/tmp/java_data/");
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }

在main函数中执行测试：

// 在Main函数中，对方法进行测试
public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
    	hdfsApp.copyFromLocalFile();
}

回到shell工具中，使用shell命令查看是否执行成功。

hadoop fs -ls /tmp/java_data

查看目录内容

任务：查看HDFS目录“/tmp/java_data”的内容。

// 添加方法listFiles，方法中查看“/tmp/java_data”目录下的内容
public void listFiles(String dir) throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path(dir));
        for(FileStatus fileStatus : fileStatuses) {
            String isDir = fileStatus.isDirectory() ? "文件夹" : "文件";
            short replication = fileStatus.getReplication();
            long len = fileStatus.getLen();
            String path = fileStatus.getPath().toString();
            System.out.println(isDir + "\t" + replication + "\t" + len + "\t" + path);
        }
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.listFiles("/tmp/java_data");
    }

查看文件内容

任务：查看HDFS文件“/tmp/java_data/file.txt”的内容。

// 添加方法cat，方法中实现对文件file.txt的查看
public void cat(String path) throws Exception {
        FSDataInputStream in = fileSystem.open(new Path(path));
        IOUtils.copyBytes(in, System.out, 1024);
        in.close();
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.cat("/tmp/java_data/file.txt");
    }

下载文件

任务：从HDFS中将“/tmp/java_data/file.txt”文件下载到本地

// 添加方法copyToLocalFile，方法中实现对文件file.txt的下载
public void copyToLocalFile() throws Exception {
        Path localPath = new Path("path to save file");
        Path hdfsPath = new Path("/tmp/java_data/file.txt");
        fileSystem.copyToLocalFile(hdfsPath, localPath);
    }

下载文件到本地，需要先将hadoop.dll文件拷贝到c:\windows\system32目录中，否则会报错java.io.IOException: (null) entry in command string: null chmod 0644。

链接: https://pan.baidu.com/s/10DJzC_341ILTb_Y6EshiVw 提取码: pun1 复制这段内容后打开百度网盘手机App，操作更方便哦 
--来自百度网盘超级会员v3的分享

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.copyToLocalFile();
    }

创建文件

任务：在HDFS “/tmp/java_data”目录下创建新文件word.txt，文件内容为hello hadoop。

// 添加create方法，在方法中实现word.txt的创建，并写入hello hadoop字符串
public void create() throws Exception {
        FSDataOutputStream output = fileSystem.create(new Path("/tmp/java_data/word.txt"));
        output.write("hello hadoop".getBytes());
        output.flush();
        output.close();
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.create();
        hdfsApp.cat("/tmp/java_data/word.txt");
    }

文件追加

任务：对“/tmp/java_data/word.txt”文件追加内容。

// 1. 在本地创建文件word_append.txt，内容为hello world append
// 2. 添加append方法，方法中实现对word.txt文件的追加
public void append() throws Exception {
        FSDataOutputStream output = fileSystem.append(new Path("/tmp/java_data/word.txt"));
        InputStream in = new BufferedInputStream(
                new FileInputStream(
                        new File("path to word_append.txt")));
        IOUtils.copyBytes(in, output, 4096);
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.append();
    }

因为hdfs会有一定的延迟，所以无法使用之前编写的cat方法立即查看结果，所以需要到命令行终端中使用shell命令查看。

hadoop fs -cat /tmp/java_data/word.txt

文件合并

任务：将 “/tmp/java_data/”目录下的file.txt文件合并到word.txt文件中。

// 添加方法concat，方法中将file.txt文件合并到word.txt文件中
public void concat() throws Exception {
        Path[] srcPath = {new Path("/tmp/java_data/file.txt")};
        Path trgPath = new Path("/tmp/java_data/word.txt");
        fileSystem.concat(trgPath,srcPath);
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.concat();
        hdfsApp.cat("/tmp/java_data/word.txt");
    }

文件改名

任务：将HDFS中的“/tmp/java_data/word.txt”改名为word_new.txt

// 添加方法rename，方法中将word.txt文件改名为word_new.txt
public void rename() throws Exception {
        Path oldPath = new Path("/tmp/java_data/word.txt");
        Path newPath = new Path("/tmp/java_data/word_new.txt");
        fileSystem.rename(oldPath, newPath);
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.rename();
        hdfsApp.listFiles("/tmp/java_data/");
    }

清空文件

任务：清空HDFS文件“/tmp/java_data/word_new.txt”内容。

// 添加方法truncate，方法中将文件word_new.txt清空
public void truncate() throws Exception {
        fileSystem.truncate(new Path("/tmp/java_data/word_new.txt"), 0);
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.truncate();
        hdfsApp.cat("/tmp/java_data/word_new.txt");
    }

删除文件

任务：将HDFS文件“/tmp/rest_data/word_new.txt”删除。

// 添加方法delete，方法中将文件word_new.txt删除
public void delete() throws Exception{
        fileSystem.delete(new Path("/tmp/java_data/word_new.txt"), true);
    }

在main函数中执行测试：

    // 在Main函数中，对方法进行测试
    public static void main(String[] args) throws Exception{
        HDFSApp hdfsApp = new HDFSApp();
        hdfsApp.delete();
        hdfsApp.listFiles("/tmp/java_data/");
    }