Hadoop之HDFS详解（二）

Hadoop之HDFS详解（二）

第三章 HDFS 的 API 操作

1. 客户端环境准备

置 Path 环境变量

IDEA 中创建一个 Maven 工程 HdfsClientDemo，并导入相应的依赖坐标+日志添加

org.apache.hadoop hadoop-client 3.1.3 junit junit 4.12 org.slf4j slf4j-log4j12 1.7.30

在项目的 src/main/resources 目录下，新建一个文件，命名为“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

建 HdfsClient 类

public class HdfsClient { @Test public void testMkdirs() throws IOException, URISyntaxException, InterruptedException { // 1 获取文件系统Configuration configuration = new Configuration(); // FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration,"root"); // 2 创建目录 fs.mkdirs(new Path("/xiyou/huaguoshan/")); // 3 关闭资源 fs.close(); }}

行程序即可

2.HDFS 的 API 案例实操

1.HDFS 文件上传

@Testpublic void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); configuration.set("dfs.replication", "2"); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, "root"); // 2 上传文件 fs.copyFromLocalFile(new Path("d:/sunwukong.txt"), new Path("/xiyou/huaguoshan")); // 3 关闭资源 fs.close();｝

注：参数优先级 参数优先级排序：（1）客户端代码中设置的值 >（2）ClassPath 下的用户自定义配置文 件 >（3）然后是服务器的自定义配置（xxx-site.xml）>（4）服务器的默认配置（xxx-default.xml）

2.HDFS 文件下载

@Testpublic void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException{ // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, "root"); // 2 执行下载操作 // boolean delSrc 指是否将原文件删除 // Path src 指要下载的文件路径 // Path dst 指将文件下载到的路径 // boolean useRawLocalFileSystem 是否开启文件校验 fs.copyToLocalFile(false, new Path("/xiyou/huaguoshan/sunwukong.txt"), new Path("d:/sunwukong2.txt"), true); // 3 关闭资源 fs.close();}

3.HDFS 文件更名和移动

@Testpublic void testRename() throws IOException, InterruptedException, URISyntaxException{// 1 获取文件系统Configuration configuration = new Configuration();FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, "root"); // 2 修改文件名称fs.rename(new Path("/xiyou/huaguoshan/sunwukong.txt"), new Path("/xiyou/huaguoshan/meihouwang.txt"));// 3 关闭资源fs.close();}

4.HDFS 文件详情查看

@Testpublic void testListFiles() throws IOException, InterruptedException, URISyntaxException {// 1 获取文件系统Configuration configuration = new Configuration();FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, "root");// 2 获取文件详情RemoteIterator listFiles = fs.listFiles(new Path("/"), true);while (listFiles.hasNext()) {LocatedFileStatus fileStatus = listFiles.next();System.out.println("========" + fileStatus.getPath() + "=========");System.out.println(fileStatus.getPermission());System.out.println(fileStatus.getOwner());System.out.println(fileStatus.getGroup());System.out.println(fileStatus.getLen());System.out.println(fileStatus.getModificationTime());System.out.println(fileStatus.getReplication());System.out.println(fileStatus.getBlockSize());System.out.println(fileStatus.getPath().getName());// 获取块信息BlockLocation[] blockLocations = fileStatus.getBlockLocations();System.out.println(Arrays.toString(blockLocations));}// 3 关闭资源fs.close();}

6.HDFS 文件和文件夹判断

@Testpublic void testListStatus() throws IOException, InterruptedException, URISyntaxException{ // 1 获取文件配置信息 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:8020"), configuration, "root"); // 2 判断是文件还是文件夹 FileStatus[] listStatus = fs.listStatus(new Path("/")); for (FileStatus fileStatus : listStatus) { // 如果是文件 if (fileStatus.isFile()) { System.out.println("f:"+fileStatus.getPath().getName()); }else { System.out.println("d:"+fileStatus.getPath().getName()); } } // 3 关闭资源 fs.close();}

第四章 HDFS 的读写流程

1.HDFS 写数据流程

客户端通过 Distributed FileSystem 模块向 NameNode 请求上传文件，NameNode 检查目标文件是否已存在，父目录是否存在。NameNode 返回是否可以上传。客户端请求第一个 Block 上传到哪几个 DataNode 服务器上。NameNode 返回 3 个 DataNode 节点，分别为 dn1、dn2、dn3。客户端通过 FSDataOutputStream 模块请求 dn1 上传数据，dn1 收到请求会继续调用 dn2，然后 dn2 调用 dn3，将这个通信管道建立完成。dn1、dn2、dn3 逐级应答客户端。客户端开始往 dn1 上传第一个 Block（先从磁盘读取数据放到一个本地内存缓存）， 以 Packet 为单位，dn1 收到一个 Packet 就会传给 dn2，dn2 传给 dn3；dn1 每传一个 packet 会放入一个应答队列等待应答。当一个 Block 传输完成之后，客户端再次请求 NameNode 上传第二个 Block 的服务 器。（重复执行 3-7 步）。

2.HDFS 读数据流程

客户端通过 DistributedFileSystem 向 NameNode 请求下载文件，NameNode 通过查 询元数据，找到文件块所在的 DataNode 地址。挑选一台 DataNode（就近原则，然后随机）服务器，请求读取数据。DataNode 开始传输数据给客户端（从磁盘里面读取数据输入流，以 Packet 为单位 来做校验）。客户端以 Packet 为单位接收，先在本地缓存，然后写入目标文件。

第五章 NameNode 和 SecondaryNameNode

1.NN 和 2NN 工作机制

第一阶段：NameNode 启动

一次启动 NameNode 格式化后，创建 Fsimage 和 Edits 文件。如果不是第一次启动，直接加载编辑日志和镜像文件到内存。

户端对元数据进行增删改的请求。ameNode 记录操作日志，更新滚动日志。ameNode 在内存中对元数据进行增删改。

第二阶段：Secondary NameNode 工作

Secondary NameNode 询问 NameNode 是否需要 CheckPoint。直接带回 NameNode 是否检查结果。

econdary NameNode 请求执行 CheckPoint。ameNode 滚动正在写的 Edits 日志。滚动前的编辑日志和镜像文件拷贝到 Secondary NameNode。econdary NameNode 加载编辑日志和镜像文件到内存，并合并。成新的镜像文件 fsimage.chkpoint。贝 fsimage.chkpoint 到 NameNode。ameNode 将 fsimage.chkpoint 重新命名成 fsimage。

2.Fsimage 和 Edits 解析

Fsimage和Edits概念

Fsimage文件：HDFS文件系统元数据的一个永久性的检查点，其中包含HDFS文件系统的所有目 录和文件inode的序列化信息。Edits文件：存放HDFS文件系统的所有更新操作的路径，文件系统客户端执行的所有写操作首先 会被记录到Edits文件中。seen_txid文件保存的是一个数字，就是最后一个edits_的数字每 次NameNode启动的时候都会将Fsimage文件读入内存，

3.CheckPoint 时间设置

通常情况下，SecondaryNameNode 每隔一小时执行一次

hdfs-default.xml

dfs.namenode.checkpoint.period 3600s

一分钟检查一次操作次数，当操作次数达到 1 百万时，SecondaryNameNode 执行一次。

dfs.namenode.checkpoint.txns 1000000操作动作次数 dfs.namenode.checkpoint.check.period 60s 1 分钟检查一次操作次数

第六章 DataNode

1. DataNode 工作机制

一个数据块在 DataNode 上以文件形式存储在磁盘上，包括两个文件，一个是数据 本身，一个是元数据包括数据块的长度，块数据的校验和，以及时间戳。DataNode 启动后向 NameNode 注册，通过后，周期性（6 小时）的向 NameNode 上 报所有的块信息。

DN 向 NN 汇报当前解读信息的时间间隔，默认 6 小时；

dfs.blockreport.intervalMsec21600000Determines block reporting interval in milliseconds.

DN 扫描自己节点块信息列表的时间，默认 6 小时

dfs.datanode.directoryscan.interval21600sInterval in seconds for Datanode to scan datadirectories and reconcile the difference between blocks in memory and on the disk.Support multiple time unit suffix(case insensitive), as describedin dfs.heartbeat.interval.

心跳是每 3 秒一次，心跳返回结果带有 NameNode 给该 DataNode 的命令如复制块 数据到另一台机器，或删除某个数据块。如果超过 10 分钟没有收到某个 DataNode 的心跳， 则认为该节点不可用。

集群运行中可以安全加入和退出一些机器。

2.数据完整性

DataNode 节点保证数据完整性的方法

当 DataNode 读取 Block 的时候，它会计算 CheckSum。如果计算后的 CheckSum，与 Block 创建时值不一样，说明 Block 已经损坏。Client 读取其他 DataNode 上的 Block。常见的校验算法 crc（32），md5（128），sha1（160）DataNode 在其文件创建后周期验证 CheckSum。

3.掉线时限参数设置

dfs.namenode.heartbeat.recheck-interval 300000 dfs.heartbeat.interval 3

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。