概要描述

TDH 生产集群一般都是开启了 kerberos 认证，本实验演示在开启了 kerberos 认证的环境下，如何通过 java 访问 hdfs 服务

以下开发环境以IDEA + Windows10 + jdk1.8 + Maven3.6.1为例

详细说明

获取 HDFS api 运行依赖的 jar 包

HDFS API 的依赖jar包都在如下列表中，jar 名称中 * 位置是包的版本号，这些 jar 的版本号根据您使用的 TDH 版本的不同会有所不同，以您集群上正在使用的包为准：

hadoop-hdfs-2.7.2-transwarp-6.2.0.jar
hadoop-hdfs-2.7.2-transwarp-6.2.0-tests.jar
hadoop-hdfs-nfs-2.7.2-transwarp-6.2.0.jar
annotations-13.0.jar
apacheds-jdbm1-2.0.0-M2.jar
asm-3.2.jar
commons-cli-1.2.jar
commons-codec-1.4.jar
commons-daemon-1.0.13.jar
commons-io-2.4.jar
commons-lang-2.6.jar
commons-logging-1.1.3.jar
dnw-1.0.7.jar
guardian-client-guardian-3.1.0.jar
guardian-common-guardian-3.1.0.jar
guava-11.0.2.jar
hdfs-plugin-transwarp-6.2.0.jar
htrace-core-3.1.0-incubating.jar
jackson-core-asl-1.9.13.jar
jackson-mapper-asl-1.9.13.jar
jersey-core-1.9.jar
jersey-server-1.9.jar
jetty-6.1.26.jar
jetty-util-6.1.26.jar
jsr305-3.0.0.jar
leveldbjni-all-1.8.jar
log4j-1.2.17.jar
netty-3.6.2.Final.jar
netty-all-4.0.23.Final.jar
plugin-common-guardian-3.1.0.jar
protobuf-java-2.5.0.jar
servlet-api-2.5.jar
slf4j-api-1.7.10.jar
swagger-annotations-1.5.9.jar
xercesImpl-2.9.1.jar
xml-apis-1.3.04.jar
xmlenc-0.52.jar
zookeeper-3.4.5-transwarp.jar

样例代码

package transwarp.support;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.*;

public class dfsClient {
    private static org.apache.hadoop.conf.Configuration getConf() {
        // 获取配置以及完成认证
        org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
        System.setProperty("java.security.krb5.conf", "D:\\JavaProject\\resource\\tdh621\\krb5.conf");
        conf.addResource("core-site.xml");
        conf.addResource("hdfs-site.xml");
        conf.addResource("yarn-site.xml");
//        conf.set("hadoop.security.authentication", "Kerberos");
        try {
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab("admin@TDH", "D:\\JavaProject\\resource\\tdh621\\admin.keytab");
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        return conf;
    }
    private static void CreateFile(String filename) throws Exception {
        // 通过Java API创建文件
        org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
        conf = getConf();
        String rootPath = "hdfs://nameservice1";
        Path p = new Path(rootPath + filename);
        FileSystem fs = p.getFileSystem(conf);
        fs.create(p);
        fs.close();
    }
    private static void CreateDir(String filename) throws Exception {
        // 创建目录
        org.apache.hadoop.conf.Configuration conf = getConf();
        String rootPath = "hdfs://nameservice1";
        Path p = new Path(rootPath + filename);
        FileSystem fs = p.getFileSystem(conf);
        boolean b = fs.mkdirs(p);
        System.out.println(b);
        fs.close();
    }
    private static void Delete(String filename) throws Exception {
        // 删除文件
        org.apache.hadoop.conf.Configuration conf = getConf();
        String rootPath = "hdfs://nameservice1";
        Path p = new Path(rootPath + filename);
        FileSystem fs = p.getFileSystem(conf);
        boolean b = fs.delete(p, true);
        System.out.println(b);
        fs.close();
    }
    public static void Downloader(String[] args) throws Exception {
        // 下载文件
        CreateFile("/tmp/fileapi3.txt");
    }
    private static void UploadFile(String localFile, String newPath) throws Exception {
        // 上传文件
        org.apache.hadoop.conf.Configuration conf = getConf();
        InputStream in = new BufferedInputStream(new FileInputStream(localFile));
        Path p = new Path(newPath);
        FileSystem fs = p.getFileSystem(conf);
        OutputStream out = fs.create(p);
        IOUtils.copyBytes(in, out, conf);
        fs.close();
        IOUtils.closeStream(in);
    }

    public static void appendFile(String[] args) {
        // 文件路径
        Path filePath = new Path("/tmp/fileapi3.txt");
        try (FileSystem fs = FileSystem.get(conf)) {
            // 检查文件是否存在
            if (!fs.exists(filePath)) {
                System.out.println("File does not exist. Please check the path.");
                return;
            }

            // 打开文件并追加内容
            try (FSDataOutputStream out = fs.append(filePath)) {
                out.writeUTF("\nNew data to append."); // 追加的内容
                System.out.println("Data appended successfully.");
            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("Error occurred while appending data to the file.");
        }
    }

    public static void main(String[] args) throws Exception {
        // 创建文件
        CreateFile("/tmp/fileapi3.txt");
        Delete("/tmp/fileapi3.txt");
        CreateDir("/tmp/tmp2/");
        UploadFile("D:\\JavaProject\\resource\\tdh621\\admin.keytab","/tmp/tmp3/");
        appendFile("D:\\JavaProject\\resource\\tdh621\\admin.keytab","/tmp/tmp3/");
    }

}