一、HDFS客户端环境准备
1）根据自己电脑的操作系统拷贝对应的编译后的hadoop jar包到非中文路径
2）配置HADOOP_HOME环境变量
3）创建maven工程，引入依赖及日志
pom依赖：
<dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>RELEASE</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.8.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.7.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.7.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.7.2</version> </dependency> <dependency> <groupId>jdk.tools</groupId> <artifactId>jdk.tools</artifactId> <version>1.8</version> <scope>system</scope> <systemPath>D:/jdk/Java/jdk1.8.0_192/lib/tools.jar</systemPath> </dependency> </dependencies>
log4j.properties：
log4j.rootLogger=INFO, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spring.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
编写程序测试：
public class HDFSClient { public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException { // 1 获取文件系统 Configuration configuration = new Configuration(); // 配置在集群上运行 // configuration.set("fs.defaultFS", "hdfs://hadoop102:9000"); // FileSystem fs = FileSystem.get(configuration); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 创建目录 fs.mkdirs(new Path("/1108/daxian/lawrence")); // 3 关闭资源 fs.close(); System.out.println("over"); } }
二、HDFS客户端API操作
2.1、HDFS文件上传（测试参数优先级）
1）编写源代码
/** * 文件上传 */ @Test public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); // configuration.set("dfs.replication", "2"); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 上传文件 fs.copyFromLocalFile(new Path("E:\bigdata\HdfsClientDemo\src\main\resources\log4j.properties"), new Path("/log4j.properties")); // 3 关闭资源 fs.close(); System.out.println("over"); }
2）将hdfs-site.xml拷贝到项目的根目录下
<?xml version="1.0" encoding="UTF-8"?> <?xml-stylesheet type="text/xsl" href="configuration.xsl"?> <configuration> <property> <name>dfs.replication</name> <value>1</value> </property> </configuration>
3）参数优先级
参数优先级排序：（1）客户端代码中设置的值 >（2）ClassPath下的用户自定义配置文件 >（3）然后是服务器的默认配置
2.2、HDFS文件下载
1）编写源代码
/** * 文件下载 */ @Test public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "atguigu"); // 2 执行下载操作 // boolean delSrc 指是否将原文件删除 // Path src 指要下载的文件路径 // Path dst 指将文件下载到的路径 // boolean useRawLocalFileSystem 是否开启文件校验 fs.copyToLocalFile(false, new Path("/log4j.properties"), new Path("E:\bigdata\HdfsClientDemo\src\main\resources\banhua.txt"), true); // 3 关闭资源 fs.close(); System.out.println("over"); }
2.3、HDFS文件夹删除
/** * 文件删除 */ @Test public void testDelete() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 执行删除 fs.delete(new Path("/1108/"), true); // 3 关闭资源 fs.close(); System.out.println("over"); }
2.4、HDFS文件名更改
/** * 文件重命名 */ @Test public void testRename() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 修改文件名称 fs.rename(new Path("/log4j.properties"), new Path("/log4j1111.properties")); // 3 关闭资源 fs.close(); System.out.println("over"); }
2.5、HDFS文件详情查看
/** * 获取文件详情 * 查看文件名称、权限、长度、块信息 */ @Test public void testListFiles() throws IOException, InterruptedException, URISyntaxException { // 1获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 获取文件详情 RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true); while (listFiles.hasNext()) { LocatedFileStatus status = listFiles.next(); // 输出详情 // 文件名称 System.out.println(status.getPath().getName()); // 长度 System.out.println(status.getLen()); // 权限 System.out.println(status.getPermission()); // 分组 System.out.println(status.getGroup()); // 获取存储的块信息 BlockLocation[] blockLocations = status.getBlockLocations(); for (BlockLocation blockLocation : blockLocations) { // 获取块存储的主机节点 String[] hosts = blockLocation.getHosts(); for (String host : hosts) { System.out.println(host); } } System.out.println("-----------分割线----------"); } // 3 关闭资源 fs.close(); }
2.6、HDFS文件和文件夹判断
@Test
public void testListStatus() throws IOException, InterruptedException, URISyntaxException {
// 1 获取文件配置信息
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop");
// 2 判断是文件还是文件夹
FileStatus[] listStatus = fs.listStatus(new Path("/"));
for (FileStatus fileStatus : listStatus) {
// 如果是文件
if (fileStatus.isFile()) {
System.out.println("file:" + fileStatus.getPath().getName());
} else {
System.out.println("dir:" + fileStatus.getPath().getName());
}
}
// 3 关闭资源
fs.close();
}2.7、HDFS的I/O流操作
上面API操作HDFS系统都是框架封装好的，如果我们想自己实现上述API的操作该怎么实现呢？我们可以采用IO流的方式实现数据的上传和下载。
2.7.1、文件上传
/** * 把本地盘上的banhua.txt文件上传到HDFS根目录 */ @Test public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 创建输入流 FileInputStream fis = new FileInputStream(new File("d:/banhua.txt")); // 3 获取输出流 FSDataOutputStream fos = fs.create(new Path("/banhua.txt")); // 4 流对拷 IOUtils.copyBytes(fis, fos, configuration); // 5 关闭资源 IOUtils.closeStream(fos); IOUtils.closeStream(fis); fs.close(); }
2.7.2、文件下载
/** * 文件下载 */ @Test public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 获取输入流 FSDataInputStream fis = fs.open(new Path("/banhua.txt")); // 3 获取输出流 FileOutputStream fos = new FileOutputStream(new File("d:/banhuaaaaa.txt")); // 4 流的对拷 IOUtils.copyBytes(fis, fos, configuration); // 5 关闭资源 IOUtils.closeStream(fos); IOUtils.closeStream(fis); fs.close(); }
2.7.3、定位文件读取
分块读取HDFS上的大文件，比如根目录下的/hadoop-2.7.2.tar.gz
/** * 定位文件读取 * 先上传一个大文件 * hadoop fs -put /opt/software/hadoop-2.7.2.tar.gz / */ @Test public void readFileSeek1() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 获取输入流 FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz")); // 3 创建输出流 FileOutputStream fos = new FileOutputStream(new File("d:/hadoop-2.7.2.tar.gz.part1")); // 4 流的拷贝 byte[] buf = new byte[1024]; for (int i = 0; i < 1024 * 128; i++) { fis.read(buf); fos.write(buf); } // 5关闭资源 IOUtils.closeStream(fis); IOUtils.closeStream(fos); fs.close(); } @Test public void readFileSeek2() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "hadoop"); // 2 打开输入流 FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz")); // 3 定位输入数据位置 fis.seek(1024 * 1024 * 128); // 4 创建输出流 FileOutputStream fos = new FileOutputStream(new File("d:/hadoop-2.7.2.tar.gz.part2")); // 5 流的对拷 IOUtils.copyBytes(fis, fos, configuration); // 6 关闭资源 IOUtils.closeStream(fis); IOUtils.closeStream(fos); fs.close(); }
合并文件，在Window命令窗口中进入到目录D:，然后执行如下命令，对数据进行合并
type hadoop-2.7.2.tar.gz.part2 >> hadoop-2.7.2.tar.gz.part1
合并完成后，将hadoop-2.7.2.tar.gz.part1重新命名为hadoop-2.7.2.tar.gz。解压发现该tar包非常完整。
三、HDFS数据流
3.1、HDFS写数据流程
3.1.1、剖析文件写入
1）客户端通过Distributed FileSystem模块向NameNode请求上传文件，NameNode检查目标文件是否已存在，父目录是否存在。
2）NameNode返回是否可以上传。
3）客户端请求第一个 Block上传到哪几个DataNode服务器上。
4）NameNode返回3个DataNode节点，分别为dn1、dn2、dn3。
5）客户端通过FSDataOutputStream模块请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成。
6）dn1、dn2、dn3逐级应答客户端。
7）客户端开始往dn1上传第一个Block（先从磁盘读取数据放到一个本地内存缓存），以Packet为单位，dn1收到一个Packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个应答队列等待应答。
8）当一个Block传输完成之后，客户端再次请求NameNode上传第二个Block的服务器。（重复执行3-7步）。
3.1.2、网络拓扑-节点距离计算
在HDFS写数据的过程中，NameNode会选择距离待上传数据最近距离的DataNode接收数据。那么这个最近距离怎么计算呢？
节点距离：两个节点到达最近的共同祖先的距离总和。
类似于如下的网络拓扑图：
3.1.3、机架感知（副本存储节点选择）
机架感知说明：
http://hadoop.apache.org/docs/r2.7.2/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html#Data_Replication
For the common case, when the replication factor is three, HDFS’s placement policy is to put one replica on one node in the local rack, another on a different node in the local rack, and the last on a different node in a different rack.
对于hadoop2.7.2版本：
