大数据学习day11------hbase_day01----1. zk的监控机制，2动态感知服务上下线案例 3.HDFS-HA的高可用基本的工作原理 4. HDFS-HA的配置详解 5. HBASE（简介，安装，shell客户端，java客户端）

2022-10-31 22:47:29

1. ZK的监控机制

1.1 监听数据的变化

（1）监听一次

public class ChangeDataWacher {

    public static void main(String[] args) throws Exception {

        // 连接并获取zk客户端的对象

        ZooKeeper zk = new ZooKeeper("feng01:2181,feng02:2181,feng03:2181", 2000, null);

        zk.getData("/user", new Watcher() {

            // 当事件触发时会执行这个方法

            @Override

            public void process(WatchedEvent event) {

                System.out.println("事件的类型:"+"路径"+event.getPath());

            }

        }, null);

        Thread.sleep(Integer.MAX_VALUE);

    }

}

当在服务器上set /user hang（改变节点数据）会出现如下变化（只能监听一次，再次改变节点数据，监听不到）

（2）多次监听

public class ChangeDataWacher {

    public static void main(String[] args) throws Exception {

        // 连接并获取zk客户端的对象

        ZooKeeper zk = new ZooKeeper("feng01:2181,feng02:2181,feng03:2181", 2000, null);

        zk.getData("/user", new Watcher() {

            // 当事件触发时会执行这个方法

            @Override

            public void process(WatchedEvent event) {

                try {

                    System.out.println("事件的类型:"+"路径"+event.getPath());

                    byte[] data = zk.getData("/user", this, null);

                    System.out.println(new String(data));

                } catch (KeeperException | InterruptedException e) {

                    e.printStackTrace();

                }

            }

        }, null);

        Thread.sleep(Integer.MAX_VALUE);

    }

}

1.2 监听节点的变化（getChildren）

public class ChangeNodeWacher {

    public static void main(String[] args) throws Exception {

        // 连接并获取zk客户端对象

        ZooKeeper zk = new ZooKeeper("feng01:2181,feng02:2181,feng03:2181", 2000, null);

        zk.getChildren("/user", new Watcher() {

            // 当事件触发时，会执行此方法

            @Override

            public void process(WatchedEvent event) {

                System.out.println("事件的类型是:"+event.getType()+"路径"+event.getPath());

                try {

                    List<String> children = zk.getChildren("/user",this, null);

                    for (String node : children) {

                        System.out.println(node);

                    }

                } catch (KeeperException | InterruptedException e) {

                    e.printStackTrace();

                }

            }

        }, null);

        Thread.sleep(Integer.MAX_VALUE);

    }

}

当在zookeeper服务端增删节点时，如下

会出现如下监听情况

2. 动态感知服务上下线案例

分布式服务端：

public class DistributeServer {

    ZooKeeper zk = null;

    // 所有server节点的父节点，要监控的节点

    String  parentPath = "/servers" ;

    /**

     * 初始化zk客户端对象

     * @throws Exception

     */

    public void init() throws Exception{

        zk = new ZooKeeper("feng01:2181,feng02:2181,feng03:2181", 2000, new Watcher() {

            @Override

            public void process(WatchedEvent event) {

                System.out.println("zk客户端连接成功");

            }

        });

    }

    /**

     * 向zk集群注册服务

     * @param hostName

     * @throws Exception

     * @throws InterruptedException

     */

    public void registerServer(String hostName) throws Exception, InterruptedException {

        zk.create(parentPath+"/server", hostName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);

        System.out.println(hostName+"上线了。。。。。");

    }

    /**

     * 正在进行的服务任务

     * @throws Exception

     */

    public void service() throws Exception {

        System.out.println("正在处理服务。。。。。。");

        Thread.sleep(Integer.MAX_VALUE);

    }

    public static void main(String[] args) throws Exception {

        DistributeServer ds = new  DistributeServer() ;

        ds.init();

        ds.registerServer(args[0]);

        ds.service();

    }

}

运行Run as----->Run Configuration。。。。(输入参数feng01)，运行结果如下：

分布式客户端

public class DistributeClient {

    ZooKeeper zk = null;

    String parentPath = "/servers" ;

    /**

     * 初始化zk客户端对象

     * @throws Exception

     */

    public void init() throws Exception {

        zk = new ZooKeeper("feng01:2181,feng02:2181,feng03:2181", 2000, new Watcher() {

            @Override

            public void process(WatchedEvent event) {

                System.out.println("zk客户端连接成功");

            }

        });

    }

    public void getHost() throws Exception, InterruptedException {

        zk.getChildren(parentPath, new Watcher() {

            @Override

            public void process(WatchedEvent event) {

                List<String> list = new ArrayList<>();

                try {

                    // 当节点发生变化时，重新获取子节点，并将子节点的值放入list

                    List<String> children = zk.getChildren(parentPath, this);

                    for (String node : children) {

                        byte[] data = zk.getData(parentPath+"/"+node, null, null);

                        list.add(new String(data));

                    }

                    // 获取子节点的值

                    System.out.println("正在服务的机器有:"+list);

                } catch (KeeperException | InterruptedException e) {

                    e.printStackTrace();

                }

            }

        });

        Thread.sleep(Integer.MAX_VALUE);

    }

    public static void main(String[] args) throws Exception {

        DistributeClient client = new DistributeClient();

        client.init();

        client.getHost();

    }

}

3.hdp的高可用基本原理的工作流程

原理图

可看博客 https://www.cnblogs.com/zsql/p/11560372.html

　　在hadoop2.0中（在2.0以下是由一个namenode）通常由两个NameNode组成，一个处于active状态，另一个处于standby状态。Active NameNode对外提供服务，而Standby NameNode则不对外提供服务，仅同步active namenode的状态，以便能够在它失败时快速进行切换。

　namenode是整个hdfs的核心，如果namenode出现单点故障了，那么整个hdfs文件系统也不能提供服务，所以hadoop2.x对hdfs提供了高可用的方案，即Hadoop HA。hdfs的高可用提供了两种方案，一种是基于QJM（Quorum Journal Manager）的，一种是基于NFS的，我们用的一般都是基于QJM的，所以这里也是讲基于QJM的高可用，高可用用来解决NameNode单点故障的问题。解决的方法是在HDFS集群中设置多个NameNode节点，但是多个namenode会产生如下问题：

（1）如何保证多个namenode的元数据一致

　　这里以两台namenode机器为例，一台状态为active（活跃状态），另一台为standby状态（备用状态）。DataNode只会将心跳信息和Block汇报信息发给活跃的NameNode。为了保证两个namenode的数据同步，hadoop中引入了journalNode(JN, 日志管理系统)。若是直接在活跃的namenode中记录日志文件，其就必须提供下载服务，供standby状态的namenode进行下载，然后进行信息的同步，但这样无疑增加了active状态的namenode的工作压力，所以就有了JN。为了防止JN出现单点故障问题，其也是一个集群，当客户端进行相关操作时（如mkdir），该操作就会被记录到该日志文件管理系统中，处于standby状态的datanode就会将之下载下来，下载得到日志信息会被FsImage对象读取并序列化得到image文件，同时该镜像（image）文件会被定期发送给active的namenode，从而实现两者的元数据一致

　　注意：有关JN

　　journal系统和namenode的本地磁盘都会记录日志信息，因为写journal是网络传输，而本地磁盘更快更安全。namenode重启时，加载本地日志回复元数据也更快，JN主要是为了提供一个可靠的元数据同步渠道，好让standby namenode囊在active namenode在挂掉后，也能成功取到元数据（standby只从JN中获取日志信息，原因主要是缓解namenode工作压力，如上所说）

（2）多个namenode如何进行状态切换（ZKFC）

ZKFC（zk Failover Controller）是一个新组件，它是一个ZooKeeper客户端，其能监视和管理NameNode的状态。运行NameNode的每台机器也运行ZKFC

运行状况监视 ：

　　ZKFC定期使用运行状况检查命令对其本地NameNode进行ping操作。只要NameNode及时响应健康状态，ZKFC就认为该节点是健康的。如果节点已崩溃，冻结或以其他方式进入不健康状态，则运行状况监视器会将其标记为运行状况不佳。

ZooKeeper会话管理：

　　当本地NameNode运行正常时，ZKFC在ZooKeeper中保持会话打开。如果本地NameNode处于活动状态，它还拥有一个特殊的“锁定”znode。此锁使用ZooKeeper对“临时”节点的支持; 如果会话过期，将自动删除锁定节点。

基于ZooKeeper的选举：

　　如果本地NameNode是健康的，并且ZKFC发现没有其他节点当前持有锁znode，它将自己尝试获取锁。如果成功，那么它“赢得了选举”，并负责运行故障转移以使其本地NameNode处于活动状态。故障转移过程类似于上述手动故障转移：首先，需对先前的活跃namenode进行隔离（防止出现脑裂的情况），然后本地NameNode转换为活动状态。

此处基于Zookeeper的选举需注意：若将standby状态的namenode切换为活跃的namenode时，需要确保先前的活跃的namenode被杀死，在将standby状态的namenode切换为active前，其会向原先的机器发送kill 命令（kill -9 namenode(Pid)）,确保原先活跃的namenode没在工作，否则两个namenode都处于活跃状态的话就会出现争抢共享资源的情况（脑裂）

4. HDFS-HA的配置详解

4.0 说明：

自己总共使用虚拟机创建了4台机器feng01，feng02，feng03，feng04，各个机器的组件有无如下表（下面自己只配置HDFS HA）

注意： Datanode和nodemanage尽量配置在同一台机器（Nodemanage是用来处理任务（Task）的，而Task是用来处理的数据的，若是nodemanage需要处理的数据就在本地的datamanage中，本地处理效率高）

4.1 前期准备（前面笔记有讲以下这些配置）

linux主机名域名映射
ip配置防火墙关闭
集群的免密配置
每个机器的免密配置
zk集群正常启动
hadoop的安装

4.2 修改hadoop中的core-site.xml文件

<configuration>

<!-- 指定hdfs的nameservice为doit11 -->

<property>

<name>fs.defaultFS</name>

<value>hdfs://doit11/</value>

</property>

<!-- 指定hadoop临时目录 -->

<property>

<name>hadoop.tmp.dir</name>

<value>/myha/hdpdata/</value>

</property>

<!-- 指定zookeeper地址 -->

<property>

<name>ha.zookeeper.quorum</name>

<value>doit01:2181,doit02:2181,doit03:2181</value>

</property>

</configuration>

4.3 修改hdfs-site.xml

<configuration>

<!--指定hdfs的nameservice为doit11，需要和core-site.xml中的保持一致 -->

<property>

<name>dfs.nameservices</name>

<value>doit11</value>

</property>

<!-- doit11下面有两个NameNode，分别是nn1，nn2 -->

<property>

<name>dfs.ha.namenodes.doit11</name>

<value>nn1,nn2</value>

</property>

<!-- nn1的RPC通信地址 -->

<property>

<name>dfs.namenode.rpc-address.doit11.nn1</name>

<value>doit01:9000</value>

</property>

<!-- nn1的http通信地址 -->

<property>

<name>dfs.namenode.http-address.doit11.nn1</name>

<value>doit01:50070</value>

</property>

<!-- nn2的RPC通信地址 -->

<property>

<name>dfs.namenode.rpc-address.doit11.nn2</name>

<value>doit02:9000</value>

</property>

<!-- nn2的http通信地址 -->

<property>

<name>dfs.namenode.http-address.doit11.nn2</name>

<value>doit02:50070</value>

</property>

<!-- 指定NameNode的edits元数据在JournalNode上的存放位置 -->

<property>

<name>dfs.namenode.shared.edits.dir</name>

<value>qjournal://doit01:8485;doit02:8485;doit03:8485/doit11</value>

</property>

<!-- 指定JournalNode在本地磁盘存放数据的位置 -->

<property>

<name>dfs.journalnode.edits.dir</name>

<value>/myha/journaldata</value>

</property>

<!-- 开启NameNode失败自动切换 -->

<property>

<name>dfs.ha.automatic-failover.enabled</name>

<value>true</value>

</property>

<!-- 配置失败自动切换实现方式 -->

<property>

<name>dfs.client.failover.proxy.provider.doit11</name>

<value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>

</property>

<!-- 配置隔离机制方法，多个机制用换行分割，即每个机制暂用一行-->

<property>

<name>dfs.ha.fencing.methods</name>

<value>

sshfence

shell(/bin/true)

</value>

</property>

<!-- 使用sshfence隔离机制时需要ssh免登陆 -->

<property>

<name>dfs.ha.fencing.ssh.private-key-files</name>

<value>/root/.ssh/id_rsa</value>

</property>

<!-- 配置sshfence隔离机制超时时间 -->

<property>

<name>dfs.ha.fencing.ssh.connect-timeout</name>

<value>30000</value>

</property>

</configuration>

4.4 将配置好的hadoop拷贝到其他节点中去

　　scp -r /usr/apps/hadoop-2.8.5 feng02:/usr/apps/

4.5 启动zookeeper集群

　　 zkServer.sh start(分别在feng01,feng02,feng03上启动)

　　查看状态（zkServer.sh status）：一个leader，两个follower

4.6 手动启动journalnode（分别在feng01，feng02，feng03）

　　hadoop-daemon.sh start journalnode

　　运行jps命令检验，feng01、feng02、feng03上多了JournalNode进程

4.7 格式化namenode

　　在feng01上执行命令：hdfs namenode -formatc

　　格式化后会在根据core-site.xml中的hadoop.tmp.dir配置生成个文件，这里自己配置的是/myha/hdpdata，同时将这个文件拷贝至另一台namenode机器上（feng02）

4.8 格式化ZKFC（在feng01上进行）

　　hdfs zkfc -formatZK

4.9 启动HDFS（在deng01上执行）

　　start-dfs.sh

到此为止HDFS HA就算配置完成了，接下来是验证配置是否成功

（1）访问http://feng01:50070，可得NameNode 'feng01:9000' (active)

　　访问http://feng02:50070 , 可得NameNode 'feng02:9000' (standby)

（2）向hdfs上传一个文件

　　hadoop fs -put /etc/profile /profile

（3）kill掉active的NameNode

（4）这个时候访问http://feng02:50070 ，会发现，namenode编程active。同时在hdfs的根目录下能发现profile文件

（5）手动启动刚才kill掉的namenode

　　hadoop-daemon.sh start namenode

（6）继续访问http://feng01:50070，可得NameNode 'feng01:9000' (standby)

这样就表示HDFS HA配置成功

5. Hbase（具体见HBASE教程）

5.1 简介　　

（1）mysql 关系型数据库等传统数据库有如下瓶颈

并发量有限, 当并发量大的时候，简单的单表查询就有可能很慢
存储的数据条数有限, 如果数据量非常大查询很慢

（2）HBase是一个分布式数据库系统（在关系型和菲关系型数据库nosql之间），其特点是高可靠性、高性能、面向列、可伸缩

高可靠性：数据的可靠和服务（集群）的可靠性。原因：hbase4是基于HDFS存储数据的（hdfs会有多个数据副本）　
高性能：在上亿条数据中查询结果控制在几十或者几百毫秒以内(分布式存储和分布式运算) , 将数据上亿条数据存储在不同的机器上 , 在查找额时候快速的找到元数据找到对应的这条数据所在的机器节点 (快速定位)
可伸缩：存储的伸缩性分布式的数据库系统基于hdfs(横向扩容) 运算机器(为我们提供服务的机器可以添加或者减少)

（3）存储的方式：

　　本质：数据在磁盘中存储大量的key-value数据

　　一个完整的ket由三段组成：行键：列族名：列明（rowkey:columnFamily:qualifier），如下

　 RowKey：用来表示唯一一行记录的主键，HBase的数据是按照RowKey的字典顺序进行全局排序的，所有的查询都只能依赖于这一个排序维度。

Hbase中的数据没有类型约束的（所谓没有类型，其实就是byte[]）

f　在物理上，hbase的这些key-value是按“完整key”的字典顺序有序存储的

（4）面向列，如下图

HBASE的物理数据存储是按列族分开存储的，所以hbase被称之为列式存储数据库。在对应的HDFS中，一张表对应一个文件夹，下面有表的列族的子文件夹（如上图的f1，f2）

注意：列族不要定的太多（若按region来查询数据的话，就需要跨多个子文件夹来查询数据），列族名尽可能短

5.2 安装

5.2.1 前提：

HDFS可以启动
zk集群启动
时间同步

（1）连接时间服务器，自动同步时间，需要安装ntpdata（yum -y install ntpdate.x86_64）,安装前检查本地yum源（yum list | grep ntpdate）

　　连接时间服务器： ntpdate 0.asia.pool.ntp.org

（2）手动设置

　　date -s ‘2019-10-11 16:00:00’

5.2.2 安装过程：

（1）上传解压：

tar -zxvf  hbase-2.0.4-bin.tar.gz  -C  /usr/apps/

（2）hbase配置

修改如下内容（vi /usr/apps/hbase-2.0.4/conf/hbase-env.sh）

export JAVA_HOME=/usr/apps/jdk1.8.0_141/

export HBASE_MANAGES_ZK=false  // 表示禁用hbase自带的zk

hbase-site.xml修改内容：

<configuration>

<!-- 指定hbase在HDFS上存储的路径 -->

<property>

<name>hbase.rootdir</name>

<value>hdfs://linux01:9000/hbase</value>

</property>

<!-- 指定hbase是分布式的 -->

<property>

<name>hbase.cluster.distributed</name>

<value>true</value>

</property>

<!-- 指定zk的地址，多个用“,”分割 -->

<property>

<name>hbase.zookeeper.quorum</name>

<value>linux91:2181,linux02:2181,linux03:2181</value>

</property>

</configuration>

regionservers 配置,表示启动机器的regionserver的设置(注：通常应该将regionserver配置为datanode相同的server上以实现本地存储，提升性能)

linux01

linux02

linux03

（3）集群分发

scp -r hbase-2.0.4/ linux02:/usr/apps

scp -r hbase-2.0.4/ linux03:/usr/apps

(4)环境变量配置（vi /etc/profile）

（4）启动

启动方式一：

hbase-daemon.sh start master

hbase-daemon.sh start regionserver

启动方式:二：

start-hbase.sh  一键启动

（5）页面访问

启动成功后，可以通过“host:port”的方式来访问HBase管理页面，例如：

5.3 shell客户端

　　见文档

5.4 java客户端

　　见文档

码农公寓

相关文章