大数据系列（5）——hdfs的学习

2024年4月18日 198次阅读 来源: EVAO_大个子

1. hdfs(分布式文件系统)

1.1 分布式文件系统

​ 数据集的大小超过一台独立的计算机的存储能力时,就要通过网络中的多个机器来存储数据集,把管理网络中多台计算机组成的文件系统,称为分布式文件系统

1.2 hdfs的特点

  • 分布式

    • 数据量越来越多，在一个操作系统管辖的范围存不下了，那么就分配到更多的操作系统管理的磁盘中，但是不方便管理和维护，因此迫切需要一种系统来管理多台机器上的文件，这就是分布式文件管理系统 ,

  • 高可用

    • 副本机制

  • 通透性

    • 实际上是通过网络来访问文件的动作，由程序与用户看来，就像是访问本地的磁盘一般

1.3 hdfs的体系架构

  • namenode

    • 名称节点
    • 文件系统的管理节点
    • 维护着整个文件系统的文件目录树
    • 接收用户的请求

  • datanode

    • 数据节点
    • 存储block(一个block在hadoop1.x的版本中64mb,在hadoop2.x的版本中是128mb)

1.4 hdfs的设计

  • 流式数据的访问

    • 一次写入多次读取

  • 商用硬件

    • hadoop不需要运行在昂贵的商业机器上(ibm的小型机等),只需要普通的机器即可

  • 低时间延时的数据访问

    • 要求几十毫秒获取响应结果的应用数据不能使用hdfs来存储
    • 虽然hdfs不能解决低延迟的访问,但是基于hdfs的hbase能解决延迟问题

  • 大量的小文件

    • 每个文件在namenode中,存储文件目录信息,block信息,约占150byte
    • hdfs不适合存储小文件

  • 多用户写入,任意修改文件

    • 存储在hdfs中的文件只能有一个写入者(writer)
    • 只能在文件末尾追加数据,不能在任意位置修改文件

1.4 block的大小规划

  • block: 数据块

    • 大数据集存储的基本单位
    • block在hadoop1.x的版本中64mb,在hadoop2.x的版本中是128mb

    • 为什么会有以上的设计

      • 硬盘有个寻址时间(10ms)
      • 寻址时间占传输时间的1%
      • 硬盘的读取速率一般为100mb/s

1.5 secondary namenode

  • 合并edits与fsimage

  • 合并的时机

    • 3600s
    • 64mb

2. hdfs的操作

2.1 图形化操作

2.2 shell操作

2.3 API操作

3. hdfs的操作(图形界面)

3.1 hdfs的启动流程

  1. 进入安全模式
  2. 加载fsimage
  3. 加载edits
  4. 保存检查点(融合fsimage和edits文件,生成新的fsimage)
  5. 退出安全模式

3.2 通过浏览器访问

​ http://namenode:50070

4. hdfs的操作(shell操作)

  • hdfs dfs
  • hadoop fs

5. hdfs的操作(API操作)

5.1 依赖POM

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.6.4</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>2.6.4</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>2.6.4</version>
</dependency>

5.2 hdfs读写文件

import org.apache.commons.compress.utils.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Test;
public class HdfsTest {
    /**
     * 写文件操作
     */
    @Test
    public void testWriteFile() throws Exception {
        //创建配置对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://uplooking01:8020");
        //创建文件系统对象
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("/test002.txt");
        FSDataOutputStream fsDataOutputStream = fs.create(path, true);
        fsDataOutputStream.write("hello".getBytes());
        fsDataOutputStream.flush();
        fsDataOutputStream.close();
    }

    /**
     * 读文件操作
     */
    @Test
    public void testReadFile() throws Exception {
        //创建配置对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://uplooking01:8020");
        //创建文件系统对象
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path("/test002.txt");
        FSDataInputStream fsDataInputStream = fs.open(path);
        IOUtils.copy(fsDataInputStream, System.out);
    }


    /**
     * 上传文件操作
     */
    @Test
    public void testuploadFile() throws Exception {
        //创建配置对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://uplooking01:8020");
        //创建文件系统对象
        FileSystem fs = FileSystem.get(conf);
        Path fromPath = new Path("file:///f:/test01.txt");
        Path toPath = new Path("/test01.txt");
        fs.copyFromLocalFile(false, fromPath, toPath);
    }

    /**
     * 下载文件操作
     */
    @Test
    public void testdownloadFile() throws Exception {
        //创建配置对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://uplooking01:8020");
        //创建文件系统对象
        FileSystem fs = FileSystem.get(conf);
        Path fromPath = new Path("/test01.txt");
        Path toPath = new Path("file:///f:/test01.txt");
        fs.copyToLocalFile(false, fromPath, toPath);
    }


    /**
     * 下载文件操作
     */
    @Test
    public void testOtherFile() throws Exception {
        //创建配置对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://uplooking01:8020");
        //创建文件系统对象
        FileSystem fs = FileSystem.get(conf);
//        BlockLocation[] blockLocations = fs.getFileBlockLocations(new Path("/test01.txt"), 0, 134217730);
//        System.out.println(blockLocations);
        FileStatus[] listStatus = fs.listStatus(new Path("/test01.txt"));
        System.out.println(listStatus);
    }
}

3. hdfs的高级操作

回滚edits: hdfs dfsadmin -rollEdits

进入安全模式: hdfs dfsadmin -safemode | enter | leave| get| wait

融合edits和fsimage: hdfs dfsadmin -saveNamespace:

查看fsimage: hdfs oiv -i -o -p

查看edits: hdfs oev -i -o -p

4. hdfs中的配额管理

  • 目录配额

    • 设置目录配额

      • hdfs dfsadmin -setQuota n dir
      • n:指的是目录配额的个数,如果个数为1,则不能存放任何文件,如果为2则只能放一个文件,以此类推.

    • 清除目录配额

      • hdfs dfsadmin -clrQuota dir

  • 空间配额

    • 设置空间配额

      • hdfs dfsadmin -setSpaceQuota n dir

        • n:指空间的大小

    • 清除空间配额

      • hdfs dfsadmin -clrSpaceQuota dir

5. 获取配置

​ hdfs getconf -confKey keyname

6. hadoop中的RPC

  • RPC(Remote Procedure Call)——远程过程调用协议
  • 它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议

  • 设计目的:

    • 调用远程的方法和调用本地方法一样方便

6.1 编写RPC服务端

定义协议

/**
 * 定义协议
 */
public interface IHelloService extends VersionedProtocol {
    public long versionID = 123456798L;//定义协议的版本
    public String sayHello(String name);//协议的具体条目
}

定义RPC的服务器实例类

/**
 * 实例类,实现了协议的类
 */
public class HelloServiceImpl implements IHelloService {
    @Override
    public String sayHello(String name) {
        System.out.println("==================" + name + "==================");
        return "hello" + name;
    }

    @Override
    public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
        return versionID;
    }

    @Override
    public ProtocolSignature getProtocolSignature(String protocol, long clientVersion, int clientMethodsHash) throws IOException {
        return new ProtocolSignature();
    }
}

定义RPC程序的启动程序

public class MyRpcServer {
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        RPC.Server server = new RPC.Builder(conf)
            .setBindAddress("172.16.4.3")//配置主机
            .setPort(8899)//配置端口
            .setProtocol(IHelloService.class)//配置协议
            .setInstance(new HelloServiceImpl())//配置实例,可以配置多个
            .build();
        server.start();
        System.out.println("RPC服务器启动成功....");
    }
}

6.2 编写RPC客户端

定义协议

/**
 * 定义协议
 */
public interface IHelloService extends VersionedProtocol {
    public long versionID = 123456798L;//定义协议的版本
    public String sayHello(String name);//协议的具体条目
}

定义客户端启动程序

Configuration conf = new Configuration();
ProtocolProxy<IHelloService> proxy = RPC.getProtocolProxy(IHelloService.class, IHelloService.versionID, new InetSocketAddress("172.16.4.3", 8899), conf);
IHelloService helloService = proxy.getProxy();
String ret = helloService.sayHello("xiaoming");
System.out.println(ret);

7. 独立启动namenode datanode

hadoop-daemon.sh start namenode

hadoop-daemon.sh start datanode

hadoop-daemon.sh start secondarynamenode

yarn-daemon.sh start resourcemanager

yarn-daemon.sh start nodemanager

8. 节点的服役和退役

  • 动态的添加节点,不需要停止整个集群
  • hdfs中维护着一个白名单和一个黑名单

8.1 节点服役

==在namenode中操作==

hdfs-site.xm

<!-- 白名单-->
<property>
    <name>dfs.hosts</name>
    <value>/opt/hadoop/etc/hadoop/dfs.include</value>
</property>

创建白名单文件

/opt/hadoop/etc/hadoop/dfs.include

​ uplooking03

​ uplooking04

​ uplooking05

​ uplooking06

刷新节点:

hdfs dfsadmin -refreshNodes

8.1 节点退役

  • 从白名单移除
  • 添加到黑名单
  • 刷新节点
  • 从黑名单移除
  • 停止datanode进程
    原文作者：EVAO_大个子
    原文地址: https://segmentfault.com/a/1190000019410477
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞