[DB] HDFS

体系架构

• NameNode
  • HDFS主节点、管理员
  • 接收客户端（命令行、Java程序）的请求：创建目录、上传、下载、删除数据
  • 管理和维护HDFS的日志和元信息
    • 日志文件（edits文件）
      • 二进制文件，记录客户端所有操作，同时体现HDFS的最新状态
      • $HADOOP_HOME/tmp/dfs/name/current
      • 日志查看器（edits viewer）：把edits转成文本（XML）格式
      • hdfs oev -i edits_inprogress_0000000000000000107 -o ~/a.xml
    • 元信息（fsimage文件）
      • 记录数据块的位置信息、数据块的冗余信息，不体现HDFS的最新状态
      • $HADOOP_HOME/tmp/dfs/name/current
      • image viewer，把fsimage文件转为文本或者xml
• DataNode
  • 数据节点
  • 按数据块保存数据库（1.x：64M，2.x：128M）
  • /root/training/hadoop-2.7.3/tmp/dfs/data/current/BP-419062579-192.168.157.111-1535553141546/current/finalized/subdir0/subdir0
  • 数据块冗余度设置原则：一般跟数据节点的个数一样，但是最大不要超过3
  • Hadoop 3.x以后，HDFS纠删码技术，大大的节约存储的空间（节约一半 ）
• SecondaryNameNode
  • 第二名称节点
  • 进行日志信息的合并
  • 由于edits文件记录了最新的状态信息，并且随着操作越来越多，edits就会越大
  • 把edits中的最新信息写到fsimage中
  • edits文件就可以清空
  • 通常和NameNode部署在一台机器上，提高下载速度
  • 什么时候合并？HDFS发出检查点时（checkpoint）
    • HDFS每隔60分钟产生一个检查点（fs.check.period）
    • edits文件达到64M（fs.check.size）

数据传输

• 数据上传
  • 请求上传数据 Distributed FileSystem.java
  • 创建 DFSClient.java
  • 建立RPC通信
  • 拿到NameNode代理对象NameNodeProxies（HA）
  • 请求创建文件元信息
  • 创建文件元信息
  • 将元信息返回给 Distributed FileSystem
  • 创建输出流 FSDataOutputStream
  • 上传数据到DataNode
  • 根据元信息，水平复制
• 数据下载
  • 请求
  • 创建
  • 建立RPC通信
  • 请求得到元信息
  • 查找元信息（先查缓存，再查fsimage）
  • 返回元信息
  • 创建输入流
  • 下载数据块
  • 把下载的数据块合成一个文件

高级特性

• 安全模式
  • 只读，正常运行时off
  • HDFS的自我保护机制，检查数据块副本率
  • 如果冗余度小于设定的副本率（DataNode坏掉），就水平复制
  • 在hdfs-default.xml中设定副本率
• 快照
  • 全部文件系统或某目录在某时刻的镜像，默认关闭
  • 启用目录的快照功能
  • 创建快照 -createSnapshot 目录 快照名称
  • 用于以下场景
    • 防止用户误操作/备份/测试/灾难恢复
  • 一般不建议使用，因为本来就有冗余，再生成新的冗余，太浪费空间
• 配额
  • HDFS为每个目录分配的大小空间
  • 名称配额
    • 设置该目录中最多存放的文件（目录）个数
  • 空间配额
    • 设置该目录中最大能够存放的文件大小
• 回收站
  • 默认禁用
  • 放入/trash
  • 回收站中文件可快速恢复
  • 可设置一个时间，超过后文件自动删除
• 用户权限管理
  • 功能较弱
  • 建议使用 Hadoop Kerberos

启动过程

• 网页->Startup Progress
• Loading fsimage
• Loading edits
• Saving checkpoint
• Safe mode

底层原理

• RPC（remote procedure call）
  • 远程过程调用（协议）
    • 在客户端调用服务器端的程序
    • 一个框架，调用者和被调用者运行在其中完成通信
    • 调用远程代码，需要实现调用者和被调用者间的连接与通信
    • 基于Client/Server进程间相互通信的一种同步通信形式
    • Client是请求服务的调用者，Server是执行Client的请求而被调用的程序
  • Hadoop用Java实现RPC
    • 客户端
    • 服务器端

MyRPCClient.java

package rpc.client;

import java.io.IOException;
import java.net.InetSocketAddress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;

import rpc.server.MyInterface;

public class MyRPCClient {

    public static void main(String[] args) throws IOException {
        // 使用Hadoop RPC框架调用Server端程序
        // 得到Server部署对象的代理对象
        MyInterface proxy = RPC.getProxy(MyInterface.class,
                     MyInterface.versionID, 
                     new InetSocketAddress("localhost",7788), 
                     new Configuration());
        // 使用代理对象调用Server程序
        String result = proxy.sayHello("Tom");
        System.out.println(result);
    }
}

MyInterface.java

package rpc.server;

import org.apache.hadoop.ipc.VersionedProtocol;

public interface MyInterface extends VersionedProtocol{
    // 定义版本号
    // 使用版本号进行签名
    public static long versionID = 1;
    
    // 定义业务方法
    public String sayHello(String name);
}

MyInterfaceImpl.java

package rpc.server;

import java.io.IOException;

import org.apache.hadoop.ipc.ProtocolSignature;

public class MyInterfaceImpl implements MyInterface{

    @Override
    public ProtocolSignature getProtocolSignature(
            String arg0, long arg1, int arg2)
            throws IOException {
        // 通过版本号定义签名信息
        return new ProtocolSignature(MyInterface.versionID,null);
    }

    @Override
    public long getProtocolVersion(String arg0, long arg1)
            throws IOException {
        // 返回版本号
        return MyInterface.versionID;
    }

    @Override
    public String sayHello(String name) {
        System.out.println("**********调用Server端**********");
        return "Hello " + name;
    }
}

MyRPCServer.java

package rpc.server;

import java.io.IOException;

import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;

public class MyRPCServer {
    public static void main(String[] args) throws HadoopIllegalArgumentException, IOException {
        // 利用Hadoop的RPC框架实现RPC Server
        
        // 使用RPC Builder构建
        RPC.Builder builder = new RPC.Builder(new Configuration());
        
        // 定义Server的参数
        builder.setBindAddress("localhost");
        builder.setPort(7788);
        
        // 部署程序
        builder.setProtocol(MyInterface.class);
        builder.setInstance(new MyInterfaceImpl());
        
        // 创建RPC Server
        Server server = builder.build();
        
        server.start();
    }
}

 

    

•  Java动态代理对象
  • 如果一个类的名字有$，表示这是一个代理对象
  • 是一种包装设计模式
  • 可以增强类的功能
  • 应用：数据库连接池
  • newProxyInstance 参数
    • ClassLoader 类加载器
    • Class<?>[ ] 真正对象实现的接口
    • InvocationHandler 实现接口来处理客户端调用

MyBusiness.java

package proxy;

public interface MyBusiness {
    public void method1();
    public void method2();
}

MyBusinessImpl.java

package proxy;

public class MyBusinessImpl implements MyBusiness {

    @Override
    public void method1() {
        System.out.println("*********method1*********");
    }

    @Override
    public void method2() {
        System.out.println("*********method2*********");
    }
}

TestMain.java

package proxy;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

public class TestMain {

    public static void main(String[] args) {
        // 创建对象
        MyBusiness obj = new MyBusinessImpl();
        // 创建代理对象
        MyBusiness proxy = (MyBusiness) Proxy.newProxyInstance(TestMain.class.getClassLoader(),
                                                                  obj.getClass().getInterfaces(),
                                                                  new InvocationHandler(){
            @Override
            public Object invoke(Object proxy,Method method,Object[] args)throws Throwable {
                if(method.getName().equals("method1")) {
                    //重写
                    System.out.println("*************代理对象中的method1*************");
                    return null;
                }else {
                    // 其他方法
                    return method.invoke(obj, args);
                }
            }
        });
        // 通过代理对象调用真正对象
        proxy.method1();
        proxy.method2();
    }
}

命令

• hdfs dfs
• -ls /：查看所有目录下的文件
• -ls /data：查看/data下的所有文件
• -mkdir /data：在hdfs上创建目录/data
• -cp：拷贝文件
• -rm：删除文件
• -get：复制文件到本地