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

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进程