- Kerberos集群CDH版本为：5.11.2，OS版本为：Redhat7.2
- 非Kerberos集群版本为：CDH5.13，OS版本为：CentOS7.2
- Maven依赖
org.apache.hadoop hadoop-client2.6.0-cdh5.11.2 org.apache.hadoop hadoop-common2.6.0-cdh5.11.2
创建访问HDFS集群的Keytab文件（非Kerberos集群可跳过此步）
[ec2-user@ip-186-31-22-86 keytab]$ sudo kadmin.local Authenticating as principal mapred/admin@CLOUDERA.COM with password. kadmin.local: listprincs fayson* fayson@CLOUDERA.COM kadmin.local: xst -norandkey -k fayson.keytab fayson@CLOUDERA.COM ... kadmin.local: exit [ec2-user@ip-186-31-22-86 keytab]$ ll total 4 -rw------- 1 root root 514 Nov 28 10:54 fayson.keytab [ec2-user@ip-186-31-22-86 keytab]$
- 获取集群krb5.conf文件，内容如下（非Kerberos集群可跳过此步）
includedir /etc/krb5.conf.d/
[logging]
default = FILE:/var/log/krb5libs.log
kdc = FILE:/var/log/krb5kdc.log
admin_server = FILE:/var/log/kadmind.log
[libdefaults]
dns_lookup_realm = false
ticket_lifetime = 24h
renew_lifetime = 7d
forwardable = true
rdns = false
default_realm = CLOUDERA.COM
#default_ccache_name = KEYRING:persistent:%{uid}
[realms]
CLOUDERA.COM = {
kdc = ip-186-31-22-86.ap-southeast-1.compute.internal
admin_server = ip-186-31-22-86.ap-southeast-1.compute.internal
}
- 配置hosts文件，确保本地开发环境与集群所有节点通且端口均放通（如8020等）
- 由于Fayson这里使用的是AWS环境所以hostname与外网的ip对应，这里会导致一个问题在向集群put数据文件时会失败，如果开发环境和HDFS都属于内网环境则不会有这个问题。
- 后面Fayson又找了一台AWS的Windows机器执行代码才能成功执行。
54.169.247.56 ip-186-31-22-86.ap-southeast-1.compute.internal 54.169.143.46 ip-186-31-26-102.ap-southeast-1.compute.internal 54.255.193.103 ip-186-31-21-45.ap-southeast-1.compute.internal 54.169.51.79 ip-186-31-26-80.ap-southeast-1.compute.internal
- 通过CM下载HDFS客户端配置
- HDFS API工具类
- 这里Fayson将HDFS的一些常用方法作为一个工具类独立出来，下面无论是Kerberos环境还是非Kerberos环境都可以直接引用，也为后期其他的项目工程开发提供便利。
- 该工具类主要是HDFS的一些常用操作，包括：创建文件，上传文件，删除文件，创建目录，读取HDFS文件等。
package com.cloudera.hdfs.utils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
public class HDFSUtils {
public static Configuration initConfiguration(String confPath) {
Configuration configuration = new Configuration();
System.out.println(confPath + File.separator + "core-site.xml");
configuration.addResource(new Path(confPath + File.separator + "core-site.xml"));
configuration.addResource(new Path(confPath + File.separator + "hdfs-site.xml"));
return configuration;
}
public static void createFile(FileSystem fs, String dst, String contents) {
try {
Path path = new Path(dst);
FSDataOutputStream fsDataOutputStream = fs.create(path);
fsDataOutputStream.write(contents.getBytes());
fsDataOutputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
public static void uploadFile(FileSystem fs, String src, String dst) {
try {
Path srcPath = new Path(src); //原路径
Path dstPath = new Path(dst); //目标路径
//调用文件系统的文件复制函数,前面参数是指是否删除原文件，true为删除，默认为false
fs.copyFromLocalFile(false,srcPath, dstPath);
//打印文件路径
System.out.println("------------list files------------"+"n");
FileStatus[] fileStatus = fs.listStatus(dstPath);
for (FileStatus file : fileStatus) {
System.out.println(file.getPath());
}
} catch (IOException e) {
e.printStackTrace();
}
}
public static void rename(FileSystem fs, String oldName,String newName) {
try {
Path oldPath = new Path(oldName);
Path newPath = new Path(newName);
boolean isok = fs.rename(oldPath, newPath);
if(isok){
System.out.println("rename ok!");
}else{
System.out.println("rename failure");
}
} catch (IOException e) {
e.printStackTrace();
}
}
public static void delete(FileSystem fs, String filePath) {
try {
Path path = new Path(filePath);
boolean isok = fs.deleteonExit(path);
if(isok){
System.out.println("delete ok!");
}else{
System.out.println("delete failure");
}
} catch (IOException e) {
e.printStackTrace();
}
}
public static void mkdir(FileSystem fs,String path) {
try {
Path srcPath = new Path(path);
if (fs.exists(srcPath)) {
System.out.println("目录已存在");
return;
}
boolean isok = fs.mkdirs(srcPath);
if (isok) {
System.out.println("create dir ok!");
} else {
System.out.println("create dir failure");
}
} catch (IOException e) {
e.printStackTrace();
}
}
public static void readFile(FileSystem fs, String filePath) {
try {
Path srcPath = new Path(filePath);
InputStream in = null;
in = fs.open(srcPath);
IOUtils.copyBytes(in, System.out, 4096, false); //复制到标准输出流
} catch (IOException e) {
e.printStackTrace();
}
}
}
- 非Kerberos环境
- 代码示例
package com.cloudera.hdfs.nonekerberos;
import com.cloudera.hdfs.utils.HDFSUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import java.io.File;
import java.io.IOException;
public class NoneKBHDFSTest {
private static String confPath = System.getProperty("user.dir") + File.separator + "hdfsdemo" + File.separator + "conf";
public static void main(String[] args) {
//初始化HDFS Configuration 配置
Configuration configuration = HDFSUtils.initConfiguration(confPath);
try {
FileSystem fileSystem = FileSystem.get(configuration);
//创建目录
// HDFSUtils.mkdir(fileSystem, "/fayson");
//创建文件
HDFSUtils.createFile(fileSystem, "/fayson/test.txt", "123testaaaaaaaaaa");
//文件重命名
HDFSUtils.rename(fileSystem, "/fayson/test.txt", "/fayson/fayson.txt");
//查看文件
HDFSUtils.readFile(fileSystem, "/fayson/fayson.txt");
//删除文件
HDFSUtils.delete(fileSystem, "/fayson/fayson.txt");
fileSystem.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
- Kerberos环境
- 代码示例
package com.cloudera.hdfs.kerberos;
import com.cloudera.hdfs.utils.HDFSUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.security.UserGroupInformation;
import java.io.File;
import java.io.IOException;
public class KBHDFSTest {
private static String confPath = System.getProperty("user.dir") + File.separator + "hdfsdemo" + File.separator + "kb-conf";
public static void main(String[] args) {
//初始化HDFS Configuration 配置
Configuration configuration = HDFSUtils.initConfiguration(confPath);
//初始化Kerberos环境
initKerberosENV(configuration);
try {
FileSystem fileSystem = FileSystem.get(configuration);
//创建目录
HDFSUtils.mkdir(fileSystem, "/test");
//上传本地文件至HDFS目录
// HDFSUtils.uploadFile(fileSystem, "/Volumes/Transcend/keytab/schema.xml", "/test");
//文件重命名
HDFSUtils.rename(fileSystem, "/test/item.csv", "/test/fayson.csv");
//查看文件
HDFSUtils.readFile(fileSystem, "/test/fayson.csv");
//删除文件
HDFSUtils.delete(fileSystem, "/test/fayson.csv");
fileSystem.close();
} catch (IOException e) {
e.printStackTrace();
}
}
public static void initKerberosENV(Configuration conf) {
System.setProperty("java.security.krb5.conf", "/Volumes/Transcend/keytab/krb5.conf");
System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
// System.setProperty("sun.security.krb5.debug", "true");
try {
UserGroupInformation.setConfiguration(conf);
UserGroupInformation.loginUserFromKeytab("fayson@CLOUDERA.COM", "/Volumes/Transcend/keytab/fayson.keytab");
System.out.println(UserGroupInformation.getCurrentUser());
} catch (IOException e) {
e.printStackTrace();
}
}
}
在进行本地开发时，必须将集群的hostname及IP配置在本地的hosts文件中（如果使用DNS服务则可以不配置hosts文件），否则无法与集群互通，确保本地客户端与集群的端口是放通的。
大数据视频推荐：
CSDN
大数据语音推荐：
企业级大数据技术应用
大数据机器学习案例之推荐系统
自然语言处理
大数据基础
人工智能：深度学习入门到精通
