首页 > 其他分享> 文章详细

ZooKeeper, HBase, Azkaban

2021-10-16 23:31:32 阅读：177 来源： 互联网

标签：info -- ZooKeeper Azkaban lagou path HBase 数据节点

文章目录

ZooKeeper
HBase
Azkaban

ZooKeeper

一、ZooKeeper 简介

分布式协调服务

主要⽤用来解决分布式集群中应⽤用系统的一致性问题

ZooKeeper 本质上是⼀个分布式的⼩文件存储系统, 基于类似于⽂文件系统的⽬目录树⽅方式的数据存储，并且可以对树中的节点进⾏行行有效管理理。
ZooKeeper 提供给客户端监控存储在zk内部数据的功能, 达到基于数据的集群管理理

统一命名服务(dubbo)、分布式配置管理理(solr的配置集中管理理)、分布式消息队列列 (sub/pub)、分布式锁、分布式协调

1.1 ZooKeeper 架构

Leader

Leader 不是手动指定的，而是 Follower 选举出来的

核心组件，事务请求(写操作) 的唯一处理者

Follower

处理客户端非事务(读操作) 请求

转发事务请求给 Leader

Observer

针对访问量大的 ZooKeeper 集群，可以增加 Observer

观察 ZooKeeper 集群的最新状态变化，并将这些状态同步

对于非事务请求，可以独立处理

对于事务请求，会转发给 Leader 服务器

不会参与任何形式的投票，只提供非事务服务

不影响群事务处理能力的前提下，提升集群的非事务处理能力

1.2 ZooKeeper 特点

Leader + 多个 Follower 的集群
Leader 负责发起投票和决议，更新系统状态
Follower 用于接受客户请求，并向客户端返回结果，参与投票
集群中有半数节点存活，集群就能正常服务
因为全局数据一致，每个 server 保存一份相同的数据副本
更新请求顺序执行
数据更新原子性，一次数据更新，要么成功，要么失败

二、环境搭建

解压

tar -zxvf zookeeper-3.4.14.tar.gz -C ../servers/

修改配置文件, 创建 data 和 log 目录

添加 myid 配置
启动 zk
查看启动情况
集群启动停止脚本

三、数据结构和监听机制

ZooKeeper 数据模型 Znode

数据信息保存在若干数据节点 Znode 上，是 ZooKeeper 中最小的数据单位

Znode 组成 Znode 树的命名空间

3.1 Znode 类型

persistent 持久性节点
ephemeral 临时性节点
sequential 顺序性节点

持久性节点，直到删除操作才会被清除

临时节点，不能有子节点，其生命周期和客户端会话绑定在一起，会话结束后被清除

持久顺序节点，持久的节点，其节点名会有一个表示顺序的数字后缀

临时顺序节点，临时的节点，其节点名会有一个表示顺序的数字后缀

事务 ID

事务 - 对物理和抽象的应用状态上的操作集合

通常指的是数据库事务，一般包含了⼀系列对数据库有序的读写操作

但是，事务是指能够改变 ZooKeeper 服务器器状态的操作

或更新操作，⼀般包括数据节点创建与删除、数据节点内容更新等操作

zk 中的事务指的是对zk服务器状态改变的操作(create, update data,更新字节点); zk 对这些事务操作都会编号，这个编号是⾃增长的被称为ZXID

3.2 Znode 状态信息

#使⽤用bin/zkCli.sh 连接到zk集群

[zk: localhost:2181(CONNECTED) 2] get /zookeeper
cZxid = 0x0
ctime = Wed Dec 31 19:00:00 EST 1969 mZxid = 0x0
mtime = Wed Dec 31 19:00:00 EST 1969 pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

ZNode 节点内容包括两部分: 节点数据内容和节点状态信息。此处数据内容是空

cZxid 就是 Create ZXID，表示节点被创建时的事务 ID。

ctime 就是 Create Time，表示节点创建时间。

mZxid 就是 Modified ZXID，表示节点最后⼀次被修改时的事务ID。

mtime 就是 Modified Time，表示节点最后⼀次被修改的时间。

pZxid 表示该节点的子节点列表最后⼀次被修改时的事务 ID。只有子节点列表变更才会更新 pZxid，⼦节点内容变更不会更新。

cversion 表示子节点的版本号。

dataVersion 表示内容版本号。

aclVersion 标识 acl 版本

ephemeralOwner 表示创建该临时节点时的会话 sessionID，如果是持久性节点那么值为 0

dataLength 表示数据⻓长度。

numChildren 表示直系⼦子节点数。

3.3 Watcher 机制

实现分布式数据的发布/订阅功能

一对多的订阅关系，让多个订阅者同时监听某个主题对象，当主题对象状态变化时，会通知所有订阅者，并作出相应的处理

客户端在向 Zookeeper 服务器注册的同时，会将 Watcher 对象存储在客户端的 WatcherManager 当中
当 Zookeeper 服务器触发 Watcher 事件后，会向客户端发送通知
客户端线程从 WatcherManager 中取出对应的 Watcher 对象来执行回调逻辑

四、基础应用

4.1 命令行操作

客户端命令行

bin/zkCli.sh	连接本地的 zk 服务器
bin/zkCli.sh -server ip:port(2181)	连接指定的服务器

查看可用命令

[zk: localhost:2181(CONNECTED) 3] help
ZooKeeper -server host:port cmd args
	stat path [watch]
	set path data [version]
	ls path [watch]
	delquota [-n|-b] path
	ls2 path [watch]
	setAcl path acl
	setquota -n|-b val path
	history
	redo cmdno
	printwatches on|off
	delete path [version]
	sync path
	listquota path
	rmr path
	get path [watch]
	create [-s] [-e] path data acl
	addauth scheme auth
	quit
	getAcl path
	close
	connect host:port

创建节点

create [-s] [-e] path data acl
-s -e 分别指定节点特性(顺序 或 临时), 若不指定，则为 持久节点

创建顺序节点

create -s /zk-test 123

-s 顺序节点
/zk-test 节点名
123 节点内容

创建临时节点

create -e /zk-temp 123

创建持久节点

create /zk-permanent 123

读取节点

ls 查看节点
ls path

get 获取节点的数据内容和属性信息
get path

更新节点

set path data
如：
set /zk-permanent 456

删除节点

delete path
delete /zk-permanent

若删除节点存在子节点，则无法删除，需要先删除子节点

4.2 客户端操作

依赖:

<!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
<dependency>
	<groupId>org.apache.zookeeper</groupId>
	<artifactId>zookeeper</artifactId>
	<version>3.4.14</version>
	<type>pom</type>
</dependency>

<!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
<dependency>
	<groupId>com.101tec</groupId>
	<artifactId>zkclient</artifactId>
	<version>0.2</version>
</dependency>

创建会话

package com.lagou.zk.demo;

import org.I0Itec.zkclient.ZkClient;

public class ZkDemo {

    public static void main(String[] args) {

        // 1. 获取 zkClient 对象
        final ZkClient zkClient = new ZkClient("linux121:2181");
        System.out.println("zkclient is ready");
    }
}

创建节点

// 2. 创建节点
// 持久节点, 注意递归创建节点，需要 设置 true
zkClient.createPersistent("/lagou-client/lagou-c1", true);
System.out.println("node is created");

删除节点

// 3. 删除节点
// 递归删除
zkClient.deleteRecursive("/lagou-client");
System.out.println("node is deleted");

监听节点变化

注意监听器可用对 不存在的节点 进行监听

监听⽬录下子节点发生改变，可以接收到通知，携带数据有⼦节点列表

3 监听⽬录创建和删除本身也会被监听到

package com.lagou.zk.demo;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.List;

public class Get_ChildNode_Change {

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

        // 1. 获取 zkClient
        final ZkClient zkClient = new ZkClient("linux121:2181");

        // 2. 对指定节点进行监听，指定收到通知后的处理逻辑
        // 处理逻辑 通过重写 handleChildChange 方法实现
        zkClient.subscribeChildChanges("/lg-client", new IZkChildListener() {
            @Override
            public void handleChildChange(String path, List<String> childs) throws Exception {
                // 打印节点信息
                System.out.println(path + " child node changes - current children " + childs);
            }
        });

        // 3. 删除节点，验证监听是否有效
        zkClient.createPersistent("/lg-client");
        Thread.sleep(1000);
        zkClient.createPersistent("/lg-client/c1");
        Thread.sleep(1000);

        zkClient.delete("/lg-client/c1");
        Thread.sleep(1000);
        zkClient.delete("/lg-client");

        Thread.sleep(Integer.MAX_VALUE);
    }
}

监听、获取数据

监听数据变化的监听器，需要监听节点是否被删除，以及监听数据是否改变

zkClient.subscribeDataChanges("/lg-client", new IZkDataListener() {
            @Override
            public void handleDataChange(String path, Object data) throws Exception {
                // 定义接收通知之后的处理逻辑
                System.out.println(path + " data is changed, new data: " + data);

            }

            // 数据删除时
            @Override
            public void handleDataDeleted(String path) throws Exception {
                System.out.println(path + " path is deleted");
            }
        });

五、原理

5.1 Leader 选举

半数机制，集群中半数以上机器存活，集群可用
Zookeeper 工作时，有一个节点为 Leader, 其他为 Follower

选举机制:

假设 5 个节点

服务器 1 启动，发出去的保文没有任何响应，所以其选举状态一直是 Looking 状态
服务器 2 启动，与 1 通信，互相交换自己的选举结果，但是由于两者都没有历史数据，所以 id 值较大的 2 胜出，但是由于没有达到超过半数以上的服务器的投票，所以其不能当选。此时 1 2 均保持 looking 状态
服务器 3 启动，根据 id 值较大的胜出，且超过半数投票，此时 3 为 Leader, 其余节点为 Follower
服务器 4 启动，id 值大小胜出，但是不超过半数投票，此时 4 为 Folloer
服务器 5 同 4, Folloer

5.2 ZAB 一致性协议

分布式数据会有一致性问题

数据复制到分布式部署的多台机器中
负载均衡技术，让分布在不同机器中的数据副本全都可以对外提供服务

数据的多副本，以及多副本同时读写，会导致一致性问题

ZAB(ZooKeeper Atomic Broadcast, 原子消息广播协议)

主备模式，保证一致性

所有的数据都由主进程处理，然后复制给副本进程

ZAB 会将服务器数据的状态变更以事务 Proposal 的形式广播到所有的副本进程上，ZAB协议能够保证了事务操作的一个全局的变更更序号(ZXID)

广播消息

ZAB 协议广播消息过程，类似于二阶段提交过程

对于 client 发送的写请求，全部由 Leader 接收, Leader 将请求封装为事务 Proposal，分发给所有 Follower

Leader 会给这个事务分配一个全局递增的唯一 ID, 即事务 ID (ZXID)
ZAB 协议要求保证事务有序
Follower 反馈 ACK
如果超过半数 Follower 反馈 ACK，则执行 Commit 操作

先提交自己，再发送 Commit 给所有 Follower

不能正常反馈 Follower 恢复正常后会进入数据同步阶段最终与 Leader 保持一致

Leader 宕机奔溃

Leader 宕机后，选举新 Leader

ZAB 协议的选举算法，能够确保 Leader 提交的事务被集体接收，并且丢弃还没有提交的事务

因此，选举机制保证选举出的新 Leader 拥有集群中所有节点最大事务编号(ZXID) 的事务

六、应用实践

6.1 服务器状态监听

应用：服务器动态上下线监听

分布式系统中，主节点会有多台

主节点因为任何原因出现宕机或者下线，客户端都要能实时感知到主节点服务器的状态

案例:

client 向服务器发送请求，获取时间
服务器为 client 请求创建 zk 节点
在节点内启动 main 函数，接收 client 请求后，返回时间

6.2 分布式锁

锁

对变量或者堆代码块做同步，即为锁

目的是实现多个线程在一个时刻内，同一个代码块只能有一个线程可执行

单机程序中，多个线程可以同时改变某个变量时，为了保证线程安全，需要对变量或代码做同步，使其在改变变量时能够串行执行消除并发修改变量

分布式锁

对于分布式程序，多台机器上的多个线程，有了线程锁，但是由于操作的是同一个数据库, 分布式程序运行在不同机器的 JVM 里，线程锁无法作用与分布式系统

此时，需要分布式锁

在整个系统提供一个全局、唯一的锁

在分布式系统中的每个机器需要获取到该锁，才能执行后续的逻辑操作

zk 实现分布式锁

利用 zk 创建临时顺序节点，实现分布式锁

思路:

分布式锁，就是指定目录下序号最小的临时顺序节点
多个系统的多个线程都要在此目录下创建临时顺序节点
每个线程都先创建临时顺序节点，然后获取当前目录下最小序号的节点，判断最小节点是不是当前节点。如果是，则获取到锁，不是，则获取锁失败
获取锁失败的线程，获取当前节点上一个临时顺序节点，并对此节点进行监听，当该节点删除的时候，此线程会收到通知，代表获取到锁

七、Hadoop HA

7.1 High Available

高可用，消除单点故障

HDFS 的 HA
YARN 的 HA

搭建两个 NN 的集群，消除单点故障（Active NN / Standby NN）

7.2 HDFS - HA

元数据

两个 NN 各自保存一份元数据

Edits 日志只有 Active NN 可以写操作

两个 NN 都可以读取 Edits
需要一个状态管理功能模块

实现一个 zkfailover 常驻在两个 NN 节点上, 负责监控各自 NN 节点

利用 zk 进行状态标识，由 zkfailover 负责切换状态，切换过程需要防止 brain split 现象发生

集群中同时出现连个 Active NN
两个 NN 之间 ssh 免密登录
两个 NN 隔离，同一时刻只有一个 NN 对外提供服务

工作机制

HDFS - HA 的自动故障转移的实现是通过为 HDFS 部署两个新组件

zookeeper, 维护少量协调数据，通知客户端数据的改变，监控客户端故障
ZkFailoverController

自动故障转移

ZK 部分:

故障检测
现役 NN 选择

ZKC 部分:

是 ZooKeeper 的客户端

健康监测
ZK 会话管理
基于 ZK 选择

7.3 HDFS - HA 集群配置

集群规划

linux121	linux122	linux123
NN	NN
JournalNode	JournalNode	JournalNode
DN	DN	DN
ZK	ZK	ZK
	ResourceManager
NM	NM	NM

启动 zk 集群

停止原先 HDFS 集群

stop-dfs.sh

创建 HA 目录，拷贝愿 hadoop 目录到 HA 目录，删除里面的 data 目录

mkdir /opt/lagou/servers/ha
cp -r /opt/lagou/servers/hadoop-2.9.2 /opt/lagou/servers/ha
rm -rf /opt/lagou/servers/ha/hadoop-2.9.2/data

配置 hdfs-site.xml

<property> 
    <name>dfs.nameservices</name> 
    <value>lagoucluster</value>
</property>

<property>
    <name>dfs.ha.namenodes.lagoucluster</name>
    <value>nn1,nn2</value>
</property>

<property> 
    <name>dfs.namenode.rpc-address.lagoucluster.nn1</name> <value>linux121:9000</value>
</property>

<property>
    <name>dfs.namenode.rpc-address.lagoucluster.nn2</name>
    <value>linux122:9000</value>
</property>

<property> 
    <name>dfs.namenode.http-address.lagoucluster.nn1</name> <value>linux121:50070</value>
</property>

<property>
    <name>dfs.namenode.http-address.lagoucluster.nn2</name>
    <value>linux122:50070</value>
</property>

<property> 
    <!-- edits 贡献目录 -->
    <name>dfs.namenode.shared.edits.dir</name>
    <value>qjournal://linux121:8485;linux122:8485;linux123:8485/lagou</value>
</property>

<property> 
    <name>dfs.client.failover.proxy.provider.lagoucluster</name>
    <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyPr ovider</value>
</property>

<property>
    <name>dfs.ha.fencing.methods</name>
    <value>sshfence</value>
</property>

<property> 
    <name>dfs.ha.fencing.ssh.private-key-files</name> 
    <value>/root/.ssh/id_rsa</value>
</property>

<property>
    <name>dfs.journalnode.edits.dir</name>
    <value>/opt/journalnode</value>
</property>

<property>
    <name>dfs.ha.automatic-failover.enabled</name>
    <value>true</value>
</property>

配置 core-site.xml

<property> 
    <name>fs.defaultFS</name>
    <value>hdfs://lagoucluster</value>
</property>

<property>
    <name>hadoop.tmp.dir</name>
    <value>/opt/lagou/servers/ha/hadoop-2.9.2/data/tmp</value> 
</property>

<property>
    <name>ha.zookeeper.quorum</name> 
    <value>linux121:2181,linux122:2181,linux123:2181</value>
</property>

拷贝配置好的 hadoop ha 环境到其他节点

启动 HDFS - HA 集群

各个 JournalNode 节点上，启动 JN 服务

/opt/lagou/servers/ha/hadoop-2.9.2/sbin/hadoop-daemon.sh start journalnode

格式化 NN 1, 并启动

/opt/lagou/servers/ha/hadoop-2.9.2/bin/hdfs namenode -format 
/opt/lagou/servers/ha/hadoop-2.9.2/sbin/hadoop-daemon.sh start namenode

NN 2 同步 NN 1 的元数据信息

/opt/lagou/servers/ha/hadoop-2.9.2/bin/hdfs namenode -bootstrapStandby

NN 1 上初始化 ZKFC

/opt/lagou/servers/ha/hadoop-2.9.2/bin/hdfs zkfc -formatZK

NN 1 上启动集群

/opt/lagou/servers/ha/hadoop-2.9.2/sbin/start-dfs.sh

验证

kill Active NN 进程

7.4 Yarn - HA 配置

Yarn - HA 不需要像 HDFS 那样同步元数据

集群规划

linux121	linux122	linux123
NN	NN
JournalNode	JournalNode	JournalNode
DN	DN	DN
ZK	ZK	ZK
	ResourceManager	ResourceManager
NM	NM	NM

配置 yarn-site.xml

<configuration>
    <property> 
        <name>yarn.nodemanager.aux-services</name> 
        <value>mapreduce_shuffle</value>
    </property>

    <!--启⽤用resourcemanager ha--> 
    <property>
        <name>yarn.resourcemanager.ha.enabled</name>
        <value>true</value>
    </property>

    <!--声明两台resourcemanager的地址--> 
    <property>
        <name>yarn.resourcemanager.cluster-id</name>
        <value>cluster-yarn</value> 
    </property>

    <property> 
        <name>yarn.resourcemanager.ha.rm-ids</name> 
        <value>rm1,rm2</value>
    </property>

    <property> 
        <name>yarn.resourcemanager.hostname.rm1</name> 
        <value>linux122</value>
    </property>

    <property> 
        <name>yarn.resourcemanager.hostname.rm2</name> 
        <value>linux123</value>
    </property>

    <!--指定zookeeper集群的地址--> 
    <property>
        <name>yarn.resourcemanager.zk-address</name>
        <value>linux121:2181,linux122:2181,linux123:2181</value>
    </property>

    <!--启⽤用⾃自动恢复--> 
    <property>
        <name>yarn.resourcemanager.recovery.enabled</name>
        <value>true</value>
    </property>

    <!--指定resourcemanager的状态信息存储在zookeeper集群--> 
    <property>
        <name>yarn.resourcemanager.store.class</name> 
        <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</ value>
    </property>
</configuration>

同步到其他节点

启动 HDFS

sbin/start-yarn.sh

HBase

一个分布式海量列式非关系型数据库系统

提供超大规模数据集的实时随机读写

列式存储的优点：减少空值，减少存储空间占用

特点:

海量存储，底层是 HDFS
列式存储, 基于列族存储
易扩展
高并发
数据可以有多个版本值，根据版本号区分(版本号是插入数据的时间戳)
数据类型单一

应用:

交通、金融、电商、电信

适合海量明细数据的存储，并且需要很好的查询性能

一、HBase shell 操作

模糊查询用 FILTER 参数 => “函数名(参数)”，多列族用 COLUMNS 参数

进入 shell

hbase shell

帮助命令

help

查看数据库中国呢有哪些表

list

创建 lagou 表，包含 base_info, extra-info 两个列族

create 'lagou', 'base_info', 'extra_info'

-- 或者指定 列族信息
create 'lagou', {NAME => 'base_info', VERSIONS => '3'}, {NAME => 'extra_info', VERSIONS => '3'}

VERSIONS 表示此单元格内的数据可以保留最近的 3 个版本

添加数据

-- row key 为 rk1, 列族 base_info 中添加 name 列标识符，值为 wang
put 'lagou', 'rk1', 'base_info:name', 'wang'

-- row key 为 rk1, 列族 base_info 中添加 age 列标识符，值为 30
put 'lagou', 'rk1', 'base_info:age', 30

-- row key 为 rk1, 列族 extra_info 中添加 address 列标识符，值为 shanghai
put 'lagou', 'rk1', 'extra_info:address', 'shanghai'

查询数据

-- 查询表中 row key 为 rk1 的所有信息
get 'lagou', 'rk1'

-- 查看 row key 下面的某个列族的信息
get 'lagou', 'rk1', 'base_info'

-- 查看 row key 指定列族中字段的值
get 'lagou', 'rk1', 'base_info:name', 'extra_info:address'
-- 或者
get 'lagou', 'rk1', {COLUMN => ['base_info:name', 'extra_info:address']}

-- 指定 row key 与 列族中列值，进行查询 binary 二分查找
get 'lagou', 'rk1', {FILTER => "ValueFilter(=, 'binary:wang')"}

-- 对 列名 模糊查询 substring
get 'lagou', 'rk1', {FILTER => "QualifierFilter(=, 'substring:a')"}

-- 查询表中所有信息
scan 'lagou'

-- 列族查询
scan 'lagou', {COLUMNS => 'base_info'}

-- scan 也可以指定 VERSIONS 参数, RAW 参数等
scan 'lagou', {COLUMNS => 'base_info', RAW => true, VERSIONS => 3}

-- 查询多个列族里，对列名模糊查询
scan 'lagou', {COLUMNS => ['base_info', 'extra_info'], FILTER => "QualifierFilter(=, 'substring:a')"}

-- 查询 rowkey
-- 顺序查询
scan 'lagou', {COLUMNS => 'base_info', STARTROW => 'rk1', ENDROW => 'rk3'}
-- 模糊查询
scan 'lagou', {FILTER => "PrefixFilter('rk')"}

-- 更新数据
-- 同插入数据

-- 更新数据值
put 'lagou', 'rk1', 'base_info:name', 'liang'

-- 删除数据和表
-- 指定 rowkey 以及 列名，删除
delete 'lagou', 'rk1', 'base_info:name'
-- 指定 rowkey 以及 列名 和 时间戳信息 删除
delete 'lagou', 'rk1', 'base_info:name', 1634301287482

-- 删除列族
alter 'lagou', 'delete' => 'base_info'

-- 清空表数据
truncate 'lagou'

-- 删除表
-- 先 disable, 再 drop
disable 'lagou'
drop 'lagou'

二、HBase 原理

2.1 HBase 读数据

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9FWrsTxy-1634397879870)(HBase-readdata.png)]

HBase 读操作

首先从 zk 找到 meta 表的region位置，然后读取 meta 表中的数据，meta 表中存储了用户表的 region 信息
根据要查询的 namespace、表名和 rowkey 信息。找到写入数据对应的 region 信息
找到这个 region 对应的 regionServer，然后发送请求
查找对应的 region
先从 memstore 查找数据，如果没有，再从 BlockCache 上读取
HBase 上 Regionserver 的内存分为两个部分
- 一部分作为 Memstore 主要用来写
- 另一部分作为 BlockCache 主要用于读数据
如果 BlockCache 中也没有找到，再到 StoreFile 上进行读取

从 storeFile 中读取到数据之后，不是直接把结果数据返回给客户端，而是把数据先写入到 BlockCache 中，目的是为了加快后续的查询; 然后在返回结果给客户端。

2.2 HBase 写数据

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-OwzSbC0w-1634397879873)(HBase-writedata.png)]

首先从 zk 找到 meta 表的 region 位置，然后读取 meta 表中的数据，meta 表中存储了用户表的 region 信息
根据 namespace、表名和 rowkey 信息。找到写入数据对应的 region 信息
找到这个 region 对应的 regionServer，然后发送请求
把数据分别写到 HLog(write ahead log) 和 memstore 各一份
memstore 达到阈值后把数据刷到磁盘，生成 storeFile 文件
删除 HLog 中的历史数据

三、HBase API

package com.lagou.hbase.client;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class HbaseClientDemo {

    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() throws IOException {
        // 获取一个配置文件对象
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "linux121,linux122,linux123");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        // 通过 conf 获取到 hbase 的集群链接
        conn = ConnectionFactory.createConnection(conf);
    }

    // 创建一张 hbase 表
    @Test
    public void creatTable() throws IOException {
        // 获取 HbaseAdmin 对象
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        // 表描述器
        HTableDescriptor teacher = new HTableDescriptor(TableName.valueOf("teacher"));
        // 设置列族描述器
        teacher.addFamily(new HColumnDescriptor("info"));
        // 执行创建操作
        admin.createTable(teacher);
        System.out.println("table teacher is created ...");
    }

    // 插入数据
    @Test
    public void putData() throws IOException {
        // 获取 table 对象
        Table t1 = conn.getTable(TableName.valueOf("teacher"));
        // 创建 put 对象，设定 rowkey
        Put put = new Put(Bytes.toBytes("110"));
        // 列族，列，值
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("addr"), Bytes.toBytes("Beijing"));
        // 插入数据到 表 t1
        t1.put(put);
        // 插入多条时，用 list<puts> 遍历批量插入

        // 关闭对象
        t1.close();
        System.out.println("data is inserted to table ...");
    }

    // 删除数据
    @Test
    public void deleteData() throws IOException {
        // 获取 table 对象
        Table t1 = conn.getTable(TableName.valueOf("teacher"));
        // 创建 delete 对象, 需要指定 rowkey
        Delete delete = new Delete(Bytes.toBytes("110"));
        t1.delete(delete);

        // 关闭 table 对象
        t1.close();

        System.out.println("data is deleted ...");
    }

    // 查询某个列族数据
    @Test
    public void getDataByColumnFamily() throws IOException {

        // 获取 table 对象
        Table t1 = conn.getTable(TableName.valueOf("teacher"));
        // 创建 get 对象，指定 rowkey
        Get get = new Get(Bytes.toBytes("110"));
        // 指定 列族信息
        get.addFamily(Bytes.toBytes("info"));
        // 查询
        Result result = t1.get(get);
        // 获取 result 中的所有 cell 对象
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            String cf = Bytes.toString(CellUtil.cloneFamily(cell));
            String column = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            System.out.println(rowkey + "\t" + cf + "\t" + column + "\t" + value);
        }

        t1.close();
    }

    @Test
    public void scanAllData() throws IOException {

        // 获取表
        Table t1 = conn.getTable(TableName.valueOf("teacher"));
        // 创建 scan 对象
        Scan scan = new Scan();
        //
        ResultScanner resultScanner = t1.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "\t" + cf + "\t" + column + "\t" + value);
            }
        }

        t1.close();
    }

    // 通过 startRowKey 和 endRowKey 进行扫描
    @Test
    public void scanRowKey() throws IOException {

        // 获取表
        Table t1 = conn.getTable(TableName.valueOf("teacher"));
        // scan 对象
        Scan scan = new Scan();
        scan.setStartRow("001".getBytes());
        scan.setStopRow("2".getBytes());
        ResultScanner resultScanner = t1.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "\t" + cf + "\t" + column + "\t" + value);
            }
        }

        t1.close();
    }

    @After
    public void release() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}

Azkaban

1. 工作流调度系统

⼀个完整的数据分析系统通常都是由⼤量任务单元组成

shell 脚本程序
java 程序
mapreduce 程序
hive 脚本等

各任务单元有前后依赖关系，组成工作流

如：某个业务系统每天产生 20 G 原始数据，每天都要进行处理，逻辑如下：

通过 hadoop 将原始数据同步到 HDFS
MR 计算框架，对原始数据进行转换，生成的数据以分区表的形式存储到多张 hive 表中
对 hive 中多张表进行 join，得到一个明细数据大表
对明细数据进行各种统计分析，得到结果报表信息
将统计分析结果同步到业务系统中，供业务调用

对于这类复杂的工作流，crontab 定时任务无法满足需求，就需要一个调度系统来支持

2. Azkaban 使用

创建 job 描述文件

vi command.job

# 内容
type=command
command=echo 'hello'

打包 job 资源文件

zip command.job

在 azkaban 的 web 界面创建 project 并伤处 job 压缩包
执行 job

3. job 的类型

前后工作流依赖的 job

# 第二个 job 依赖 第一个 job

# 第一个 job, foo.job
type=command
command=echo 'foo'

# 第二个 job, bar.job
type=command
dependencies=foo
command=echo 'bar'

打包时，将这两个 job 文件打包到一起

HDFS 任务调度

# 将 hdfs 命令写入 job 文件, fs.job
type=command
command=/opt/lagou/servers/hadoop-2.9.2/bin/hdfs dfs -mkdir /azkaban

MR 任务调度

# 需要 mr 程序的 jar 包
# mrwc.job
type=command
command=/opt/lagou/servers/hadoop-2.9.2/bin/hadoop jar hadoop-mapreduce-examples-2.9.2.jar wordcount /wordcount/input /wordcount/azout

Hive 脚本任务调度

# hive 脚本 test.hql
use default;
drop table aztest;
create table aztest(id int, name string) row format delimited fields terminated by ',';

# hivef.job
type=command
command=/opt/lagou/servers/hive-2.3.7/bin/hive -f 'test.hql'

定时任务调度

在 Azkaban 的 web 界面配置 project 的定时调度信息

标签：info,--,ZooKeeper,Azkaban,lagou,path,HBase,数据,节点
来源： https://blog.csdn.net/badassname11/article/details/120805879

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9