HDFS补充点知识

川长思鸟来 2024-04-18 05:32 136阅读 0赞

一、HDFS分块抽象的好处

1 文件大小可以大于任意一个磁盘的容量，块并不需要存储在同一个磁盘上

2 抽象块作为存储单元，简化存储子系统的设计

  1. datanode将块作为处理对象，能存储多少块也能计算出
  2. namenode管理元数据

3 数据备份提高容错能力和可用性

HDFS块为什么这么大?

最小化寻址开销

块足够大时，磁盘传输速率会远远大于寻址时间，则传输时间更依赖于磁盘传输速率

二、 HDFS处理大量小文件

大量小文件的缺点：

  1. 占用namenode内存
  2. 处理时增加map任务数量，增加寻址次数

1 使用HAR(Hadoop Archives)，构建在其他文件系统之上用于文件存档的文件系统

Hadoop存档文件系统通常将HDFS中的多个文件打包成一个存档文件，减少namenode内存的使用

hadoop archive命令创建HAR文件

2 使用sequencefile

3 运行时使用CombinerFileInputFormat

点此查看对小文件的处理

三、HDFS的容错机制

1 远程备份组成元数据持久状态的文件

将持久状态写入本地磁盘的同时，写入一个远程的网络文件系统NFS，操作是实时且原子性的

2 运行secondary namenode(辅助namenode)

辅助namenode主要是合并edits和fsimage，并保存合并后的fsimage

namenode出现故障时，启用辅助namenode，因为辅助namenode上的fsimage未包含最近的edits文件，所以一般是将远程NFS上的元数据信息拷贝到辅助namenode上作为主namenode运行
具体见HDFS学习总结的checkpoint的工作机制

四、文件权限

  1. hadoop fs -ls /countout

启用权限控制

  1. dfs.permission.enabled

修改文件权限

  1. hadoop fs -chmod 777 /wordcount

文件模式

drwx: 目录 可读 可写 可执行

-r-x: 文件 可读 不可写 可执行

五、高可用相关

故障切换failover controller

  1. 管理将active namenode转换为standby namenode的过程
  2. 默认使用zookeeper保证只有一个active namenode
  3. 每个namenode上运行一个failover controller，监控宿主namenode状态(使用心跳机制)并在active namenode失效时进行故障切换
  4. 使用配置文件实现故障切换控制

规避fencing

  1. 为保证之前的active namenode确实停止运行，高可用使用“规避”
  2. QJM：仅允许一个namenode向编辑日志中写入数据，设置一个SSH规避命令用于杀死namenode进程

  3. NFS过滤器：不能实现只有一个namenode写入数据

    撤销访问权限

    屏蔽网络端口

    断电

高可用实现了什么

一个新的namenode满足以下条件才可以开始为集群提供服务，导致冷启动时间需要30min，甚至更长

  1. 将命名空间映像加载到内存中
  2. 重演编辑日志
  3. 接收到足够多datanode的数据块报告并退出安全模式

高可用HDFS架构实现：

  1. 使用高可用的共享存储实现编辑日志的共享
  2. datanode需要同时向两个namenode发送数据块处理报告，因为文件与数据块的映射关系存储在namenode内存中，而不是磁盘上
  3. 使用故障切换和规避机制处理namenode的失效问题，这对用户是透明的
  4. 备用namenode包含了辅助namenode的功能，并为活动namenode的命名空间设置周期性检查点

结果：
因为备用namenode内存中也持有最新的状态：包含最新的编辑日志条目和最新的数据块映射信息，所以只需要几十秒到一分钟就可以实现任务接管

六、java API实现流式数据访问

  1. package HDFS;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.*;
  5. import org.junit.Test;
  7. import java.io.IOException;
  8. import java.net.URISyntaxException;
  11. /**
  12.  * Unit test for simple App.
  13.  */
  14. public class AppTest {
  16.     @Test
  17.     /**
  18.      *   HDFS获取文件系统
  19.      */
  20.     public void initHDFS() throws Exception {
  22. // 1 创建配置信息对象
  23.         Configuration configuration = new Configuration();
  24. // 2 获取文件系统
  25.         FileSystem fs = FileSystem.get(configuration);
  26. // 3 打印文件系统
  27.         System.out.println(fs.toString());
  29.     }
  31.     @Test
  32.     /**
  33.      * HDFS文件上传(测试参数优先级)
  34.      */
  35.     public void testCopyFromLocalFIle() throws URISyntaxException, IOException, InterruptedException {
  37.         //1 获取文件系统
  38.         Configuration configuration = new Configuration();
  39.         //configuration.set("dfs.replication","2");
  40.         FileSystem fs = FileSystem.get(configuration);
  41.         //2 上传文件
  42.         fs.copyFromLocalFile(new Path("D:/222.txt"),new Path("/idea.txt"));
  43.         //3 关闭资源
  44.         fs.close();
  45.         System.out.println("over");
  46.     }
  48.     /**
  49.      * HDFS文件的下载
  50.      */
  51.     @Test
  52.     public void testCopyToLocalFile() throws URISyntaxException, IOException, InterruptedException {
  54.         // 1 获取文件系统
  55.         Configuration configuration = new Configuration();
  56.         FileSystem fs = FileSystem.get(configuration);
  57.         // 2 执行下载操作
  58.         //boolean delSrc 指是否将原文件删除
  59.         //Path src 指要下载的文件路径
  60.         //Path dst 指将文件下载到的路径
  61.         //boolean useRawLocalFileSystem 是否开启文件校验
  62.         fs.copyToLocalFile(false,new Path("/wordcount.txt"),new Path("D:/wordcount.txt"),true);
  63.         //3 关闭资源
  64.         fs.close();
  65.         System.out.println("over");
  66.     }
  67.     @Test
  68.     /**
  69.      * HDFS文件目录的创建
  70.      */
  71.     public void  testMkdir() throws URISyntaxException, IOException, InterruptedException {
  73.         //1.获取文件系统
  74.         Configuration configuration = new Configuration();
  75.         FileSystem fs = FileSystem.get(configuration);
  76.         //2.创建目录
  77.         fs.mkdirs(new Path("/Test/Mkdir/is/success"));
  78.         //3.关闭资源
  79.         fs.close();
  80.     }
  82.     /**
  83.      * HDFS文件的删除
  84.      * @throws IOException
  85.      */
  86.     @Test
  87.     public void testDelete() throws IOException, URISyntaxException, InterruptedException {
  89.         //1.获取文件系统
  90.         Configuration configuration = new Configuration();
  91.         FileSystem fs = FileSystem.get(configuration);
  93.         //2.执行删除
  94.         fs.delete(new Path("/Test"),true);// 后面的布尔选项是递归删除
  95.         //3.关闭资源
  96.         fs.close();
  97.     }
  99.     /**
  100.      * HDFS 文件名的修改
  101.      * @throws URISyntaxException
  102.      * @throws IOException
  103.      * @throws InterruptedException
  104.      */
  105.     @Test
  106.     public void testRename() throws URISyntaxException, IOException, InterruptedException {
  108.         //1 获取文件系统
  109.         Configuration configuration = new Configuration();
  110.         FileSystem fs = FileSystem.get(configuration);
  111.         //2 修改文件名称
  112.         fs.rename(new Path("/test"),new Path("/test.txt"));
  113.         //3. 关闭资源
  114.         fs.close();
  115.     }
  116.     @Test
  117.     public void testListFiles() throws URISyntaxException, IOException, InterruptedException {
  119.         //1 获取文件系统
  120.         Configuration configuration = new Configuration();
  121.         FileSystem fs = FileSystem.get(configuration);
  122.         long s = fs.getStatus().getCapacity();
  123.         System.out.println(s);
  124.         //2 获取文件详情
  125.         /**
  126.          * 思考:为什么返回迭代器不是list之类的容器
  127.          * 集合占内存多 一次全部拿取
  128.          * 迭代器一次拿一个
  129.          */
  130.         RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"),true);
  132.         while (listFiles.hasNext()){
  134.             LocatedFileStatus status = listFiles.next();
  135.             //输出详情
  136.             //文件名称
  137.             System.out.println("文件名:"+status.getPath().getName());
  138.             //长度
  139.             System.out.println("文件大小"+status.getLen());
  140.             //获取组
  141.             System.out.println("所属组:"+status.getGroup());
  142.             //权限
  143.             System.out.println("权限:"+status.getPermission());
  145.             //获取存储的块信息
  146.             BlockLocation[] blockLocations = status.getBlockLocations();
  148.             for (BlockLocation blockLocation:blockLocations){
  150.                 //获取块的存储的主机节点
  151.                 String[] hosts = blockLocation.getHosts();
  152.                // String[] ip = blockLocation.getNames();
  153.                 for (String host:hosts){
  155.                     System.out.println("获取块的存储的主机节点:"+host);
  156.                 }
  157.               //
  158.             }
  159.             System.out.println("-------------------------");
  160.         }
  162.         fs.close();
  163.     }
  164. }

七、副本放在哪里？？

副本放哪里？
考量因素：可靠性，写入带宽，读取带宽

权衡冗余性和带宽损耗

多个副本：第一个副本放在运行客户端的节点上，第二个副本放在不同的机架的节点上，第三个副本与第一个副本放在相同机架上，但在不同节点上，其他副本随机放在集群中的节点

八、客户端和与之交互的HDFS，namenode，datanode之间的数据流

  1. 文件读取

  2. 客户端调用FileSystem的open()方法打开文件，对于HDFS，是一个DistributedFileSystem的实例

  3. DistributedFileSystem的实例通过RPC调用namenode，namenode根据映射关系返回存有文件块副本的datanode地址，datanode根据与客户端的距离排序(网络拓扑)，如果本身也是datanode且存有文件块，则客户端从本地获取副本
  4. DistributedFileSystem返回一个FSDataInputStream对象给客户端以便读取数据，FSDataInputStream封装了DFSInputStream，这个类管理者datanode和namenode的I/O，客户端对DFSInputStream反复调用read()方法，将数据从datanode传输到客户端
  5. 块读取完毕后，DFSInputStream关闭与该datanode的连接，寻找下一个块所在的最佳datanode，重复读取，一直到读取完毕，调用FSDataInputStream的close()

DFSInputStream的作用

  1. 管理namenode和datanode的I/O
  2. 存储namenode告知的块位置
  3. 如果遇到与datanode通信错误，则连接另一个最近的有副本的datanode，并记录故障节点，不再与它通信

  4. 校验datanode块的完整性，如果有问题，告诉namenode，并从其他节点读取副本

  5. 文件写入

  6. 客户端使用DistributedFileSystem对象调用create()创建一个新文件

  7. 对namenode进行RPC调用在文件系统命名空间新建文件，namenode执行检查确认文件不存在且客户端有新建文件权限
  8. DistributedFileSystem向客户端返回一个FSDataOutputStream对象，FSDataOutputStream封装了一个DFSOutputStream对象,负责与namenode和datanode通信

  9. 写入数据时

    DFSOutputStream将数据分为数据包，写入“数据队列”(data queue)

    DataStreamer挑选出适合存储副本的一组datanode，并向namenode请求分配新的数据块

    存储副本的datanode形成管线，DataStreamer将数据包流式传输到第一个datanode，这个datanode通过管线将数据包传递给下一个datanode，依次到结束

    DFSOutputStream维护确认队列(ack queue)，收到所有datanode确认信息后，数据包从确认队列删除

  10. 客户端完成数据写入后，调用FSDataOutputStream的close()方法，将剩余数据包全部写入datanode管线，并等待确认信息，最后告知namenode文件写入完成

如果datanode在写入时发生故障

  1. 关闭管线，把队列中的所有数据包(共同组成一个块)添加回数据队列的最前端，确保下游的datanode不漏掉任何一个数据包
  2. 将存储了正确数据块信息的datanode添加标识，把标识传给namenode，故障节点在恢复后删除存储的部分数据块，在管线添加一个新的datanode，替换故障datanode

九、HDFS 一致模型和distcp

  1. 一致模型

新建文件后，能在文件系统的命名空间立即可见

写入文件的内容不能保证立即可见，因为正在写入的块对reader不可见，会不一致

hflush()：FSDataInputStream的方法，HDFS保证目前写入的数据都到达datanode的写入管线并对所有reader可见，确保数据在内存中，不确保在已经写入磁盘；HDFS的close()方法已经包含执行hflush()

hsync()：确保数据已经写入磁盘，提交文件描述符的缓冲数据

重要性：

如果不使用一致模型，系统故障时可能会丢失数据块。

一致模型提供了鲁棒性，但增加HDFS负载，吞吐量受到影响，所以要根据实际找到合适的调用频率

  1. distcp

Hadoop自带程序，并行地将数据复制进出Hadoop，作为一个MR作业实现，并行运行map，没有reducer，-m可以修改运行map的数量

  1. hadoop distcp file1 file2
  2. hadoop disctp dir1 dir2
  3. //如果dir2不存在，则新建dir2
  4. //如果dir2存在，则形成dir2/dir1
  5. //可以指定多个来源dir
  6. hadoop distcp -update dir1 dir2
  7. hadoop distcp -overwrite dir1 dir2

HDFS集群间传送数据

  1. hdfs webhdfs HttpFs
  3. hadoop distcp -update -delete -p hdfs://namenode1/foo hdfs://namenode2/foo
  5. hadoop distcp webhdfs://namenode1:50070/foo webhdfs://namenode2:50070/foo

保持HDFS集群均衡

复制数据可能影响集群的均衡性，最好先使用默认的每个节点20个map运行distcp运行，还可以使用之后介绍的均衡器工具改善均匀程度

例如-m设置为1，则首先选择运行map的当前节点存储副本，其他副本分散在集群中，这导致运行节点总会有副本，不均衡

发表评论

表情：
评论列表 （有 0 条评论，136人围观）

还没有评论，来说两句吧...

相关阅读

    相关 HDFS补充点知识

    一、HDFS分块抽象的好处 1 文件大小可以大于任意一个磁盘的容量，块并不需要存储在同一个磁盘上 2 抽象块作为存储单元，简化存储子系统的设计 1. datan...

    川长思鸟来川长思鸟来/ 0 赞/ 137 阅读