HBase浅度学习

摘要：是中分布式存储和负载均衡的最小单元，即不同的可以分别在不同的上，但同一个是不会拆分到多个上。目前会有一个线程来负责的操作。当文件的数量增长到一定阈值后，系统会进行合并，在合并过程中会进行版本合并和删除工作，形成更大的。

存储数据

支持海量数据的存储
hbase是构建在hdfs上的分布式数据库

检索数据

hbase支持对存储在hbase表中的海量数据进行随机的实时的查询服务 
hbase对其大表中的海量数据构建了层层索引 
    
    

要存储的数据为海量的数据

RDBMS 
    集群性能比较弱，不容易集群节点扩展
    一旦存储的表的数据量较大，导致表的索引文件也变大，影响到后续的读写效率            
    
hbase 
    构建在hdfs上分布式数据库，支持节点无限扩展
    hbase的出现就是RDBMS在面对海量数据存储及检索时的一个可替代工具/框架

要存储的数据为非结构化的数据

结构化数据 
    mysql或hive表中的数据结构化的数据 

非结构化的数据 
    每条数据的字段数量不相同 
    图片、视频、文本数据都是非结构化的数据
                
        

“Not Only SQL”的缩写，不仅仅是sql ,以nosql数据库在记录数据时所使用的数据结构的不同将nosql数据库分为四大家族

列存储数据库家族 -- 代表 hbase  
    表中每列的数据是一个连续的存储单元 
    hive表或者mysql中默认每行的数据是一个连续的存储单元 

文档型数据库家族 -- 代表MongoDB  
    以文档形式记录数据库中的数据
    爬虫项目中
    
键值对数据库家族 --代表redis
    以key-value形式记录数据的数据库 
    redis是基于内存的key-value数据库 
        sparkStreaming/strom/Flink进行实时分析计算-》redis-》接入前端进行实时更新展示

图形结构数据库家族--代表Neo4J 
    以图形形式记录数据 
    

https://baike.baidu.com/item/...

接入在线业务提供内容查询服务 （借助hbase分布式数据库进行海量用户数据的存储，并依靠其完善的检索机制为海量数据提供随机的实时的查询服务）

微博、论坛、朋友圈等社交数据的存储
海量数据
图片、视频、文本数据都是非结构化的数据

各大电商平台的订单数据
未完成的订单（热数据）-- oracle
已完成的订单（冷数据）-- hbase

物流信息存储查询

银行客户交易数据

支付公司的交易数据

移动电信公司的话单记录存储

交通数据存储

医疗数据

大数据分析平台中的数据存储库
可以用hbase作为大数据分析平台中的数据源
MapReduce、hive、spark等计算框架可以直接从hbase表中读写数据

HBase在滴滴出行的应用场景和最佳实践
https://blog.51cto.com/xiaogo...

hbase源自于谷歌三大论文之一的 BigTable     
    GFS  -- hdfs  
    MapReduce  -- MapReduce 
    BigTable -- hbase 
hbase在hadoop生态圈中的地位 
    构建在hdfs上分布式数据库，支持海量数据的存储 
    可以MapReduce、hive、spark框架进行集成使用  
基于【列存储】的数据库 
    列存储与行存储  
        RDBMS数据库都是默认行存储  
            每行的数据在底层是一个连续的存储单元 
            在select查询时如果只涉及到表中的其中几列数据，则无关列也会被加载读取，大大增加系统的io流 
        Hbase数据库默认是列存储 
            每列的数据在底层是一个连续的存储单元 
            在select查询时如果只涉及到表中的其中几列数据，则只有涉及到的列会被加载读取
            因为每列的数据保存在一起，并且每列的数据的数据类型相同，则更容易实现压缩存储  
适合存储非结构化和结构化的数据  
基于key-value形式存储的数据库 
    根据key来获取对应的value值
    rowkey+列簇+列+时间戳 =》value
高可靠、高性能、可伸缩的分布式数据库 
    高可靠、可伸缩：构建在hdfs上 
    高性能：对比RDBMS，针对海量数据的读写是高性能的 
    

行存储与列存储
https://blog.csdn.net/dc_726/...

hive
    面向分析的数据仓库工具（非数据库），使用的是Hql语句分析
    查询高延迟
    存储结构化数据
    不能直接接入web前端业务使用
    默认行存储，是纯逻辑表
    Hive本身不存储和计算数据，它完全依赖于HDFS和MapReduce，本质就是将hql转化为mr。
hbase
    面向数据存储和检索的数据库
        数据存储 -- 海量数据存储（底层是hdfs）
        数据检索 -- 支持海量数据随机的、实时的数据检索（依靠层层索引）
    查询低延迟
        可以实现随机实时检索大表中的数据
    适合存储结构化和非结构化数据        
    可以接入web前端业务使用
    hbase是列存储，是物理表，不是逻辑表，通过索引方便快速查询更新等操作
    hbase是一个在hdfs上开发的面向列的分布式数据库，本身不支持sql，但是可以借助其他插件实现sql功能。
        通过hive创建与hbase表的关联映射 -- 使用hql实现离线分析
        通过Phoenix插件实现实时分析

    Phoenix介绍：
        针对hbase开发的第三方插件，目前已贡献给Apache               
        Phoenix是构建在HBase上的一个SQL层
        Phoenix完全使用Java编写，作为HBase内嵌的JDBC驱动能让我们用标准的JDBC API而不是HBase客户端API来创建表、插入数据和查询数据。
        Phoenix查询引擎会将SQL查询转换为一个或多个HBase扫描，编排执行以生成标准的JDBC结果集。
                JDBC是一种用于执行SQL语句的Java API               
        版本    Phoenix 2.x/3.x - HBase 0.94.x
                Phoenix 4.x - HBase 0.98.1+    
                

** HBase是分布式架构,支持服务器在线添加和移除，可以很好的扩展（基于hdfs）
** RDBMS可以使用sql语句,HBase通常使用API来访问[第三方插件支持phoenix]
** RDBMS是基于行存储,HBase是基于列存储,可以支持更好的存储和压缩
** RDBMS适合存储结构化的数据,HBase适合结构化和非结构化的数据存储
** RDBMS支持比较好的事务,HBase不支持事务
** RDBMS支持多表Join,HBase不支持Join
** 通常HBase表的应用场景比较简单,不适合业务逻辑很复杂的查询环境
** RDBMS相比较hbase有更完善的索引机制
** HBase通常应用都是单表数据量巨大,用关系型数据库无法满足的情况        
    
    
    

hbase分布式数据库是主从架构 
    

hbase的集群的主节点 
管理用户对hbase表的增删改查（表整体非表内的数据）
管理并分配表的region给regionserver从节点,分配时遵循一个【散列原则：同一张表多个分片尽量分配个不同的机器】及【负载均衡原则：不同的机器尽量处理相同数量的分片】  
监控regionserver的运行状态及regionserver节点宕机后的容灾处理  
    master借助zookeeper间接监控regionserver 
    zookeeper会感知regionserver节点的上线和下线 
        regionserver启动后会在zookeeper上注册信息
    但某个regionserver节点宕机后zookeeper会第一时间感知并及时通知master，master会进行容灾处理将此regionserver节点上管理的region重构在其他regionserver节点上
    

hbase的从节点  
管理master所分配的region 
真正响应客户端对表的读写请求的节点

table在行的方向上分隔为多个Region。Region是HBase中分布式存储和负载均衡的最小单元，即不同的region可以分别在不同的Region Server上，但同一个Region是不会拆分到多个server上。

每个region至少一个store，每个列簇对应一个store
一个Store由一个memStore和0或者 多个StoreFile组成。 HBase以store的大小来判断是否需要切分region

HLog

HLog(WAL log)：WAL意为write ahead log，用来做灾难恢复使用，HLog记录数据的所有变更，一旦region server 宕机，就可以从log中进行恢复。
每个 Region Server 维护一个 Hlog,而不是每个 Region 一个。
HLog文件就是一个普通的Hadoop Sequence File， Sequence File的value是key时HLogKey对象，其中记录了写入数据的归属信息，除了table和region名字外，还同时包括sequence number和timestamp，timestamp是写入时间，sequence number的起始值为0，或者是最近一次存入文件系统中的sequence number。 Sequence File的value是HBase的KeyValue对象，即对应HFile中的KeyValue。

memStore

memStore 是放在内存里的。保存修改的数据即keyValues。当memStore的大小达到一个阀值（默认128MB）时，memStore会被flush到文 件，即生成一个快照。目前hbase 会有一个线程来负责memStore的flush操作。

StoreFile

memStore内存中的数据写到文件后就是StoreFile，StoreFile底层是以HFile的格式保存。当storefile文件的数量增长到一定阈值后，系统会进行合并（minor、major compaction），在合并过程中会进行版本合并和删除工作（majar），形成更大的storefile。
    

hbase表数据最终落地在hdfs上 
为hbase的表数据提供了一个存储平台 

hbase强依赖zookeeper  
    kafka、storm也是强依赖zookeeper
zookeeper基于第三方观察者模式监控regionserver、master的运行状态及节点宕机后的容灾处理 
    保证hbase集群的高可用性
        hbase可以为master节点配置多个备份 
        当active master宕机后zookeeper会感知并从备份master中选举出一个作为后续的active master  
zookeeper持有hbase集群的所有的服务器节点信息及所有用户表的元数据信息的位置信息 
    持有hbase集群的所有的服务器节点信息 
        hbase服务启动后master及regionserver节点会向zookeeper上注册自己的节点信息 
            
    持有所有用户表的元数据所在的meta表的region位置信息！！！ 
        hive表的元数据 -- 存在RDBMS数据库 
        hbase表的元数据 -- 存在在hbase上的一张名称为meta的系统表中  
        
        hbase上所有用户表的元数据信息都存在hbase上的一个meta表中  
        meta表的数据量通常比较少，所有meta表通常只有一个region
        此meta表的region会被master分配给某个regionserver节点管理  
        zookeeper持有该meta表的region所在的regionserver节点信息 （知道meta表的region在哪个regionserver节点上） 
        
        
        meta表存有用户表的哪些元数据信息呢？      
            用户表分成了几个region 
            每个region的key的范围信息  
            每个region所在regionserver节点信息 
            …… 


                

1、安装hadoop并启动hdfs服务

2、安装并启动zookeeper服务

3、上传解压并修改hbase配置文件

$ tar zxf /opt/softwares/hbase-1.2.0-cdh5.14.2.tar.gz -C /opt/cdh-5.14.2/
    
1）修改hbase-env.sh    
    # The java implementation to use.  Java 1.7+ required.
    export JAVA_HOME=/opt/cdh-5.14.2/jdk1.8.0_112
    # Tell HBase whether it should manage it"s own instance of Zookeeper or not.
    export HBASE_MANAGES_ZK=false
    
2）修改hbase-site.xml 

声明hbase表相关数据在hdfs上的根目录

     
        hbase.rootdir
        hdfs://192.168.134.101:8020/hbase
      
      
      
        hbase.cluster.distributed
        true
      
      
      
      
        hbase.zookeeper.quorum
        192.168.134.101
      

3）修改 regionservers 
    声明集群中的哪些服务器作为hbase的regionserver节点 
    类似于hadoop中配置slaves
    
    192.168.134.101    
    

4、启动hbase服务

在主节点服务器上启动master进程 
    $ bin/hbase-daemon.sh start master
在所有的hbase的从节点服务器上启动regionserver进程 
    $ bin/hbase-daemon.sh start regionserver 
    
或者在master节点服务器上执行 
    $ bin/start-hbase.sh 
    

5、hbase的web管理平台

http://192.168.134.101:60010/master-status 

    user table -- 用户表  
    system table --系统表 
        hbase:meta    
            hbase的元数据表 
            存储了所有用户表的region的引用 
        hbase:namespace 
            命名空间表  
            

6、启动hbase后的zookeeper及hdfs上的变化

    hdfs上 
        /hbase/data/
            hbase的表数据的命名空间库目录 
            自定义的命名空间库目录将出现在该data目录下
        /hbase/data/default 
            默认命名空间库目录 
            在hbase上建表时如果未指定表的命名空间库则默认创建在default下 
        /hbase/data/hbase 
            系统命名空间库目录 
            meta --- meta表数据目录  
            namespace --命名空间库元数据表     
            
    zookeeper 
        $ bin/zkCli.sh 
        ls /hbase                    
            meta-region-server  -- meta表的位置信息 
            rs -- regionserver节点的信息 
            backup-masters -- master节点的备份信息 
            master -- active master节点信息 
        

rowkey：行健  
    用来标识hbase表中唯一一行数据  
    类似RDBMS中的主键  
column family ：列簇/列族    
    可以将某些列组织到一起形成一个家族 
    列簇为一张表的列增加一层索引
    在hbase表创建时可以声明多个列簇，一般应用中列簇的数量不超过3个，最好只设置1个列簇？？？？ 
    在创建表至少要声明一个列簇 ,但是不需要声明列名 
column ： 列 
    值所属的列  
    hbase表中不同的行可以有不同的列 
    在hbase表中插入具体数据时再指定列名 
    某个列必须要属于某个列簇  
cell ： 单元格  
    单元格是hbase表最小最基本的存储单元  
    单元格是实际值的存储地 
    每个单元格的组成 : rowkey+列簇+列+时间戳=》value值  
        时间戳： 
            value值在插入到单元格那一刻的时间戳 
            时间戳可以用来区分一个单元格中多个历史版本的值 
        版本： 
            hbase表的单元格中可以存储多个历史版本值  
            一个单元格中默认显示的最新版本的值 
        
在hbase表中如何指定唯一的一个单元格/唯一的值 
    rowkey+列簇+列=》单元格 
    rowkey+列簇+列+时间戳=》唯一的值 
            
    
    

与hbase进行交互的几种方式

hbase shell -- 测试 
web ui（hue）-- 测试  
java api / Phoenix -- 生产 

    

$ bin/hbase shell 进入到hbase的shell交互命令行

退格键出现乱码：

xshell 
    文件-属性-终端-键盘-两个都选择ASCII 127  
secureCRT 
    选项-会话选项-仿真-终端-选择linux 
    选项-会话选项-映射键-两个勾选 
            

help 查看hbase shell支持的命令

COMMAND GROUPS:
  Group name: general
  Commands: status, table_help, version, whoami

  Group name: ddl
  Commands: alter, alter_async, alter_status, create, describe, disable, disable_all, drop, drop_all, enable, enable_all, exists, get_table, is_disabled, is_enabled, list, locate_region, show_filters

  Group name: namespace
  Commands: alter_namespace, create_namespace, describe_namespace, drop_namespace, list_namespace, list_namespace_tables

  Group name: dml
  Commands: append, count, delete, deleteall, get, get_counter, get_splits, incr, put, scan, truncate, truncate_preserve

  Group name: tools
  Commands: assign, balance_switch, balancer, balancer_enabled, catalogjanitor_enabled, catalogjanitor_run, catalogjanitor_switch, close_region, compact, compact_mob, compact_rs, flush, major_compact, major_compact_mob, merge_region, move, normalize, normalizer_enabled, normalizer_switch, split, trace, unassign, wal_roll, zk_dump

  Group name: replication
  Commands: add_peer, append_peer_tableCFs, disable_peer, disable_table_replication, enable_peer, enable_table_replication, get_peer_config, list_peer_configs, list_peers, list_replicated_tables, remove_peer, remove_peer_tableCFs, set_peer_tableCFs, show_peer_tableCFs, update_peer_config

  Group name: snapshots
  Commands: clone_snapshot, delete_all_snapshot, delete_snapshot, list_snapshots, restore_snapshot, snapshot

  Group name: configuration
  Commands: update_all_config, update_config

  Group name: quotas
  Commands: list_quotas, set_quota

  Group name: security
  Commands: grant, list_security_capabilities, revoke, user_permission

  Group name: procedures
  Commands: abort_procedure, list_procedures

  Group name: visibility labels
  Commands: add_labels, clear_auths, get_auths, list_labels, set_auths, set_visibility

  Group name: rsgroup
  Commands: add_rsgroup, balance_rsgroup, get_rsgroup, get_server_rsgroup, get_table_rsgroup, list_rsgroups, move_servers_rsgroup, move_tables_rsgroup, remove_rsgroup

创建表命令

create "ns1:t1", {NAME => "f1", VERSIONS => 5}

    在ns1命名空间库下创建一个t1表
    t1表有一个名称为f1的列簇，并且该列簇下的单元格可最大支持5个版本值

create_namespace "ns1" 创建一个命名空间库

list_namespace

list 查看用户表

create "t1", {NAME => "f1"}, {NAME => "f2"}, {NAME => "f3"}

    在默认命名空间库下创建一个t1表  
    该表有三个名称为 f1 f2 f3的列族   
        

describe "t1" 描述一张表

create "t2", {NAME => "f1", VERSIONS => 2}, {NAME => "f2", VERSIONS => 4}, {NAME => "f3"}

在hbase表同一张表中不同的列簇可以有不同的属性值 
    因为同一张表下的同一个列簇中的cell存在同一个文件中，不同列簇中的cell存放在不同文件中

create "t1", "f1", "f2", "f3"

创建一个t1表，并且该表的三个列簇都使用默认属性    
如果希望对列簇进行特殊属性设定，需要使用 {NAME => "f1", VERSIONS => 2}
    

创建一个员工表并插入数据

create "emp" , "info1","info2"   

# put插入/更新数据 
# 一次只能插入一个cell，注意和sql的区别，不能insert插入一整行数据

hbase(main):037:0> put "emp", "10003", "info1:name", "tom"
0 row(s) in 0.2320 seconds

hbase(main):038:0> put "emp", "10003", "info1:age", "25"
0 row(s) in 0.0090 seconds

hbase(main):039:0> put "emp", "10003", "info2:school", "jiaoda"
0 row(s) in 0.0200 seconds

hbase(main):040:0> put "emp", "10003", "info2:qq", "1234554321"
0 row(s) in 0.0100 seconds

hbase(main):041:0> put "emp", "10004", "info1:name", "lio"
0 row(s) in 0.0060 seconds

hbase(main):042:0> put "emp", "10004", "info1:sex", "boy"
0 row(s) in 0.0090 seconds

hbase(main):043:0> put "emp", "10004", "info1:tall", "175cm"
0 row(s) in 0.0300 seconds

hbase(main):044:0> put "emp", "10004", "info2:job1", "java"
0 row(s) in 0.0140 seconds

hbase(main):045:0> put "emp", "10004", "info2:job2", "bigdata"
0 row(s) in 0.0330 seconds

hbase(main):046:0> put "emp", "10005", "info1:name", "lili"
0 row(s) in 0.0190 seconds

hbase(main):047:0> put "emp", "10005", "info2:school", "ligong"
0 row(s) in 0.0190 seconds

hbase(main):048:0> put "emp", "10006", "info1:name", "mary"
0 row(s) in 0.0510 seconds

hbase(main):049:0> put "emp", "10006", "info1:age", "18"
0 row(s) in 0.0440 seconds

hbase(main):050:0> put "emp", "10006", "info2:job1", "UI"

scan查询数据

scan "emp"    
scan "emp" , {COLUMNS => "info1"}     查看某张表中某个列簇下的数据  
scan "emp" , {COLUMNS => "info1:name"}    查看某列数据  
scan "emp" , {COLUMNS => ["info1:name","info1:age"]}     查看多列数据  
scan "emp" , {COLUMNS => ["info1:name","info1:age"]}     查看多列数据  
scan "emp" , {COLUMNS => "info1:name",LIMIT => 3}     最前面的3条数据 
scan "emp" , {COLUMNS => "info1",STARTROW => "10004",STOPROW=>"10006"}    某个rowkey范围内的数据  

显示出来的每行是一个cell

get查询数据

scan 是一个 范围扫描查询 
get 只能查询某条范围内的数据   

get "emp","10004"  查询某行数据
get "emp","10004" ,{COLUMN => "info1"} 查询某行数据下的某个列簇下的数据
get "emp","10004" ,{COLUMN => "info1:name"}  查询某个cell中的数据最新版本的值  
get "t1", "r1", {COLUMN => "c1", TIMESTAMP => ts1}  查询指定历史版本的值 
get "t1", "r1", {COLUMN => "c1",  VERSIONS => 4} 查询指定的cell中最新4个历史版本值 

delete/deleteall删除数据

delete--删除某个value值

deleteall--删除的某行或某个cell

delete "t1", "r1", "c1", ts1  删除指定历史版本的值  
delete "emp","10004","info2:job1"  不指定时间戳删除的是最新版本的值 
deleteall "ns1:t1", "r1"  删除某行数据  
deleteall "t1", "r1"
deleteall "t1", "r1", "c1"  删除某个cell单元格  
deleteall "t1", "r1", "c1", ts1  测试 ！！！     

演示多版本值：

create "emp2" ,{NAME => "info", VERSIONS => 5}
put "emp2","10005","info:age" ,"18"
put "emp2","10005","info:age" ,"19"
put "emp2","10005","info:age" ,"20" 
scan "emp2",{  VERSIONS => 2}   查询所有cell最新2个版本的值  
delete "emp2","10005","info:age" 不指定时间戳删除的是最新版本的值  
delete "emp2","10005","info:age" ,1543999422333  删除指定历史版本的值     

修改表的hbase shell 命令

alter "t1", NAME => "f1", VERSIONS => 5  将t1表中的f1列簇的VERSIONS属性值改为 5  
alter "ns1:t1", NAME => "f1", METHOD => "delete"  将t1表中的f1列簇删除 
alter "t1", NAME => "f2", METHOD => "delete" 

truncate "emp2" 清空表

disable "emp1" 禁用

drop "emp1" 删除表

count "emp" 统计多少行

hbase表的读写流程

十五章-46页 架构图     

HBase读数据流程： --根据rowkey查询emp表数据

1、client先去访问zookeeper,从zookeeper里面获取meta表所在位置信息
        0.96之前的版本除了meta表还有一个root表，root表存储meta表位置信息，先通过zookeeper获取root表位置
        在从root表中读取meta表位置
        现在直接将meta表位置存入zookeeper中，减少会话次数
2、client向meta所在regionserver发起访问,读取meta表数据，获取hbase集群上所有user表的元数据
3、根据meta表中emp表的region的位置信息，client找到了当前需要访问的emp表对应的region及所在的regionserver服务器
4、client向对应regionserver服务器发起读请求
5、regionserver收到client访问请求，先扫描memstore，在扫描blockcache，最后再读取storefile[HDFS]文件
6、regionserver把数据响应给client

HBase写数据流程： --根据rowkey写入emp表

1、client先去访问zookeeper,从zookeeper里面获取meta表所在位置信息
2、client向meta所在regionserver发起访问,读取meta表数据，获取hbase集群上所有user表的元数据
3、根据meta表中emp表的region的位置信息，client找到了当前需要访问的emp表对应的region及所在的regionserver服务器
4、client向对应regionserver服务器发起写请求
5、regionserver收到client请求,并响应，client先把数据写入HLog,防止数据丢失
6、再把数据写入memstore内存缓存区,默认128M
7、当Hlog和memstore都写入成功,则这条数据写入成功    
8、当memstore达到阀值[128M],会把memstore里面的数据Flush成storefile
9、当[128M]storefile越来越多,会触发compact合并操作,把多storefile合并成一个大的storefile
    合并的期间会删除过期的版本或数据，例如更新或delete删除的数据并未直接删除，而是打了删除标签直到此时才会真正删除    
10、当单个storefiles[region]越来越大,达到一定阀值（10G或其他动态阈值）时会触发split操作,region被一分为二被管理    

写流程中的三大机制

flush 机制
compaction机制 
split机制  

hbase java api使用

增删改查操作  

hbase中将表分成了1个或多个region进行分布式管理
region是hbase表管理的基本单元   

hbase为什么将表分成多个region进行管理呢？ 
    如果表比较大不利于表数据的并行读写操作 
    分而治之
    
一张表如何分成多个region的？ 
    创建表时进行预分区 
        默认建表时不进行预分区则表只有1个region   
        建表时可以指定一个表的region的个数及region的key的范围
            这个指定的key是rowkey的前缀
    region被动split分割
        当向某张表的某个region中持续写入数据，region所承载的数据量达到一定的阈值【10G或小于10G的一个动态值】，则会进行被动的split分割，1个region分为2个region 
        
表的region如何被master分配管理的？ 
    一个表的多个region被master【散列】分配个regionserver集群进行管理
    每个regionserver节点可以管理不同表的region，但是默认情况下每个regionserver节点最终管理的region的总数量是相同的【负载均衡】

    
region的内部结构 
    每个region是由1个Hlog及1到多个store组成(每个region中store的数量=该表的列簇的数量，每个store下存储了某一个列簇下的所有的数据)      
        每个store是由1个memstore及0到多个storefile组成 
            memstore： 
                写缓存，用来加快hbase表的写速度 
                memstore的阈值是128M，当memstore中缓存的数据量达到128M则会flush成storeFile文件 
            
            storefile
                由memstore中flush出    
                storefile文件就是hbase表数据的存储文件 
                storefile文件最终落地到hdfs上最终形成HFile 
                
        Hlog文件 
            每个region中还包含一个Hlog文件  
            Hlog文件是一个预写制日志文件【WAL】  
            写数据时默认情况下数据会先写入到对应region的Hlog中进行备份
            再将数据写入到memstore中 
            当数据成功写入Hlog+memstore后则后台判断此次写入数据成功 
            当因为regionserver节点宕机导致memstore中的数据丢失时可以从Hlog进行恢复 
            
            某些场景用可以配置关闭WAL预写机制  
            在1.x版本中可以配置一个regionserver节点对应一个Hlog文件（一个regionserver节点上所有的region共用一个Hlog文件）=》可以减少HLog文件的寻址次数  


    hbase表数据在hdfs上的存储目录结构 
        
        /hbase/data/default  
            在各个命名空间库下包含有多个以表明命名的目录 
        /hbase/data/default/emp 
            在表目录下包含有1个或多个以region编号命名的目录（该目录下存储了该表对应region下的数据）                    
            region的name  
                可以在60010web平台上查看到  
                region name的命名规则： 表名+Start Key+时间戳+region编号   
        /hbase/data/default/emp/8b1ae0ef2208947a238544b4b94ffeaa 
            在region编码命名得目录下会出现1个或多个以列簇名命名的目录 
        /hbase/data/default/emp/8b1ae0ef2208947a238544b4b94ffeaa/info1 
            在列簇名命名的目录下出现0到多个storefile文件 
            
            
            

十五章-46页 架构图 

1）client  客户端 
    hbase shell --交互命令行  
    Hue - web  
    MapReduce  
    hive 
    spark 
    …… 
2）zookeeper  
    监控master及regionserver的状态，保证hbase集群的高可用 
    持有hbase集群的节点信息及meta表的region的位置信息  
3）master  
    负责分配表的region给regionserver  
    负责集群的负载均衡  
    读写过程没有经过master节点，所以master节点负载率通常比较低 
4）regionserver  
    regionserver节点通常与hdfs的datanode节点部署在同一台物理机上 
    regionserver负责管理表的region的节点 
    是真正相应客户端读写请求的节点（响应客户端IO请求的节点）    
5）hadoop 
    hadoop的hdfs为hbase提供了一个数据存储平台 
    hdfs上主要存储了hbase的两种文件  
        HFile  
            就是hbase的StoreFile
            表数据的存储文件  
            Hfile是hadoop的二进制格式文件   
        Hlog  
            预写制日志文件  
            Hlog是hadoop的sequence格式文件 
            

                
1、client先去访问zookeeper,从zookeeper里面获取meta表的region所在的regionserver节点信息
                0.96之前的版本除了meta表还有一个root表，root表存储meta表位置信息，先通过zookeeper获取root表位置
    在从root表中读取meta表位置
    现在直接将meta表位置存入zookeeper中，减少会话次数
    
2、client向meta表的region所在regionserver发起读请求,读取meta表数据，获取hbase集群上所有用户表的元数据
3、根据meta表中emp表的region的位置信息，client找到了当前需要访问的emp表对应的region及所在的regionserver服务器
4、client向对应regionserver服务器发起读请求
5、regionserver收到client访问请求，在当前节点上找到emp表的region，并确定目标store，先扫描memstore（因为数据此时可能存在memstore并未flush成storeFile），在扫描blockcache（读缓存，近期已经读过的数据会加载到该缓冲区中），最后再读取storefile[HDFS]文件，
6、regionserver把数据响应给client


            

                    
1、client先去访问zookeeper,从zookeeper里面获取meta表的region所在的regionserver节点信息
2、client向meta表的region所在regionserver发起读请求,读取meta表数据，获取hbase集群上所有用户表的元数据
3、根据meta表中emp表的region的位置信息，client找到了当前需要访问的emp表对应的region及所在的regionserver服务器
4、client向对应regionserver服务器发起写请求   
5、regionserver收到client写请求并响应，默认情况下regionserver先把数据写入目前region的HLog中,防止数据丢失
6、再把数据写入到目标store下的memstore内存缓存区,memstore内存缓存区的默认大小128M 
    $ vi hbase-common/src/main/resources/hbase-default.xml      
    hbase.hregion.memstore.flush.size => 128M  
    
7、当Hlog和memstore都写入成功,则此次数据写入成功    
8、当memstore内的数据量达到阀值[128M]，会把memstore里面的数据Flush成storefile
9、当持续写入数据，导致store下的storefile越来越多,当store文件数量或每间隔一段时间则会触发compact合并操作,hbase会把每个store下的所有的storefile合并成一个大的单一的storefile
10、当合并后的单个storefiles文件越来越大,达到一定阀值（10G或其他小于10G的动态阈值）时会触发split操作,region被一分为二，并由master将新的region分配给regionserver节点管理 

写数据过程中发送的三大机制flush、compaction、split

当memstore内的数据量达到阀值[128M]，会把memstore里面的数据Flush成storefile

hbase.hregion.memstore.flush.size
134217728

Memstore will be flushed to disk if size of the memstore
exceeds this number of bytes.  Value is checked by a thread that runs
every hbase.server.thread.wakefrequency.

    思考： 
        例如一个regionserver节点上同时管理了100个region，平均每个region内有2个memstore，则该regionserver需要同时维护100*2=200 个  
        
        假如单个memstore内的数据量都没有达到128M，但是都在100M以上，则此时所有的memstore内的缓存的数据量将超过  200*100M = 20G  
        
        占用的内容空间是 regionserver -jvm -heap堆内存空间 

hbase.regionserver.global.memstore.size

Maximum size of all memstores in a region server before new
    updates are blocked and flushes are forced. Defaults to 40% of heap (0.4).
    Updates are blocked and flushes are forced until size of all memstores
    in a region server hits hbase.regionserver.global.memstore.size.lower.limit.
    The default value in this configuration has been intentionally left emtpy in order to
    honor the old hbase.regionserver.global.memstore.upperLimit property if present.

        默认值为 regionserver -jvm -heap堆内存空间的40%   
        当一个regionserver节点上的所有的memstore内缓存的数据量超过regionserver-jvm -heap堆内存空间的40%时，
        将会阻塞此regionserver节点上所有的region的更新操作
        
        如何配置 regionserver-jvm -heap堆内存空间 大小 
            $ vi conf/hbase-env.sh 
            export HBASE_HEAPSIZE=1G 
            正常大小都在  4G-20G 

hbase.regionserver.global.memstore.size.lower.limit

Maximum size of all memstores in a region server before flushes are forced.
    Defaults to 95% of hbase.regionserver.global.memstore.size (0.95).
    A 100% value for this value causes the minimum possible flushing to occur when updates are
    blocked due to memstore limiting.
    The default value in this configuration has been intentionally left emtpy in order to
    honor the old hbase.regionserver.global.memstore.lowerLimit property if present.

        默认值为 regionserver -jvm -heap堆内存空间的40%*0.95=38%  
        当一个regionserver节点上的所有的memstore内缓存的数据量超过regionserver -jvm -heap堆内存空间的38%时，
        会进行强制的flush操作
    

    

    当flush到store下的storefile文件过多过小将影响到hbase后续的读效率
    hbase后启动compaction合并机制，最终将每个store下的所有的storeFile文件合并为一个大的单一的storeFile文件 
        
    compaction机制 分割两个合并过程
        minor compaction --小合并  
            系统后台会有一个专用的线程监控每个store下的storeFile文件的数量
            当数量超过3个时会进行小合并 
            小合并只是一个归类合并没有用扫描读取文件内的内容也没有进行排序操作 
            该过程较快并且不会消耗集群资源 
        major compaction --大合并 
            所有的region每隔3.5天~10.5天之间会进行一次大合并 
            大合并过程中会加载读取一个store下所有的storefile文件并进行以下操作
                会依据cell中的rowkey值的字典顺序进行排序 
                会对打上’删除‘标签的cell或者过期的cell进行彻底的删除  
                    执行的删除cell的操作并没有将cell立即从storeFile文件中删除
                    只是打上一个’删除‘标签，在大合并期间cell才会被彻底的删除

    
        hbase.hregion.majorcompaction
        604800000
        The time (in miliseconds) between "major" compactions of all
        HStoreFiles in a region.  Default: Set to 7 days.  Major compactions tend to
        happen exactly when you need them least so enable them such that they run at
        off-peak for your deploy; or, since this setting is on a periodicity that is
        unlikely to match your loading, run the compactions via an external
        invocation out of a cron job or some such.
    
    
    
        hbase.hregion.majorcompaction.jitter
        0.50
        Jitter outer bound for major compactions.
        On each regionserver, we multiply the hbase.region.majorcompaction
        interval by some random fraction that is inside the bounds of this
        maximum.  We then add this + or - product to when the next
        major compaction is to run.  The idea is that major compaction
        does happen on every regionserver at exactly the same time.  The
        smaller this number, the closer the compactions come together.
    
    
    
        hbase.hstore.compactionThreshold
        3
        
        If more than this number of HStoreFiles in any one HStore
        (one HStoreFile is written per flush of memstore) then a compaction
        is run to rewrite all HStoreFiles files as one.  Larger numbers
        put off compaction but when it runs, it takes longer to complete.
    
    
        hbase.hstore.blockingStoreFiles
        10
        
        If more than this number of StoreFiles in any one Store
        (one StoreFile is written per flush of MemStore) then updates are
        blocked for this HRegion until a compaction is completed, or
        until hbase.hstore.blockingWaitTime has been exceeded.
    
    
        hbase.hstore.blockingWaitTime
        90000
        
        The time an HRegion will block updates for after hitting the StoreFile
        limit defined by hbase.hstore.blockingStoreFiles.
        After this time has elapsed, the HRegion will stop blocking updates even
        if a compaction has not been completed.
    
    
        hbase.hstore.blockingWaitTime
        90000
        
        The time an HRegion will block updates for after hitting the StoreFile
        limit defined by hbase.hstore.blockingStoreFiles.
        After this time has elapsed, the HRegion will stop blocking updates even
        if a compaction has not been completed.
    
    
        hbase.hstore.compaction.kv.max
        10
        How many KeyValues to read and then write in a batch when flushing
                or compacting.  Do less if big KeyValues and problems with OOME.
                Do more if wide, small rows.
    

            大合并过程会消耗大量的hbase集群资源（IO，cpu，内存 ）  
            大合并可以看做是以短期的集群资源消耗换取以后长期的读效率的提升 

            

当大合并后的某个region下的某个store下的storefile文件大小达到一定的阈值，则会引起该region的split分割 

0.94版本之前
    是一个定值
    单纯使用hbase.hregion.max.filesize控制的是某个store下storeFile的大小，默认是10G时自动split分割
    默认配置文件hbase-common/src/main/resources/hbase-default.xml 
        $ echo $((10737418240/1024/1024/1024))    
    结论：
        对小表不友好
        数据量较小的表就有可能永远不会触发分裂，容易产生数据热点
    
0.94版本-1.2版本
    由hbase.hregion.max.filesize和hbase.regionserver.region.split.policy共同控制
    split.policy的默认值 IncreasingToUpperBoundRegionSplitPolicy 底层是根据一个公式来计算是否要split
    公式：Min (R*R* hbase.hregion.memstore.flush.size , “hbase.hregion.max.filesize”) 
        R为同台regionserver上同一个表的region的个数
        hbase.hregion.memstore.flush.size默认值是128M
        假如有一张表tableA，其多个region分布在regionserver集群上：
        如果在某一个regionserver上持有了tableA的1个region，则该regionserver上的tableA的region触发分割时的大小 => min(10G，1*1*128M)=128M
        如果在某一个regionserver上持有了tableA的2个region，则该regionserver上的tableA的region触发分割时的大小 => min(10G，2*2*128M)=512M
        … 3… =>：min(10G，3*3*128M)=1152M
        ……
        ……
        … 9… =>min( 10G ，9*9*128M= 10.125G )=>10G              
        结论:
            同一个regionserver节点上持有某张表的region数量达到9个时，则该表的region分割触发值才开始按照hbase.hregion.max.filesize最大值10G进行分割
            这种切分策略很好的弥补了ConstantSizeRegionSplitPolicy的短板，能够自适应大表和小表
            但是很多小表会在大集群中产生大量小region，最终表数据在集群中过于分散
                例如节点数量为10个，一个表最终达到分割阈值10G前将会有产生90个region，前期分割过频繁
            
1.2版本 
    由hbase.hregion.max.filesize和hbase.regionserver.region.split.policy共同控制
    split.policy默认值改为SteppingSplitPolicy
    控制策略： 
        某台regionserver上持有某个表的region个数为1个时=》split切分阈值为flush size * 2
        其他情况为hbase.hregion.max.filesize=》10G 
    结论： 
        弥补IncreasingToUpperBoundRegionSplitPolicy带来的问题
        小表前期随数据量增加不会过度过多分割
            例如节点数量为10个，一个表最终达到分割阈值10G前将会有产生10个region
            每台上存在1个region后，以后分割阈值即按照10G 执行 

        
region在分割时： 
    消耗集群资源 
    region在split时会短暂的offline下线导致客户端访问不稳定 
    
    
为什么在建表时表的列簇的数量不宜过多 ？？？ 
    因为一个region中store的数量等于该表的列簇的数量 
    region的split分割的依据是region下的某个store下的storeFile文件的大小 
    如果一个region下有多个store，其中一个store下的storeFile文件大小逐渐变大
    而其他的store下的数据量比较小
    如果数据量较大的store下的文件大小达到阈值则会触发整个region的split分割 
    数据量少的store会被迫分裂在多个region中，导致数据过于分散增加hbase对该store下数据的检索成本 
    
        
        
        

使用java 编程对hbase进行增删改查操作 

1、配置maven工程依赖

1）关闭eclipse ，并合并repository目录  
    
2）新建maven工程或使用之前的maven工程并再pom.xml文件中添加以下依赖 

        
        
            org.apache.hadoop
            hadoop-client
            2.6.0
        
        
        
        
            org.apache.hbase
            hbase-server
            1.2.0
        
        
            org.apache.hbase
            hbase-client
            1.2.0
        
        
        
        
            org.apache.hive
            hive-exec
            1.1.0
        

3）下载hadoop及hbase的配置文件并加入到maven工程中 

    在maven工程中创建source funder 
        src/main/resources  
    将hadoop及hbase的配置文件加入到src/main/resources下 

            

2、hbase java api演示

如何通过hbase java api创建表、删除表、修改表

http://hbase.apache.org/book.... =》95. Examples

package com.example.hbase.admin;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;

public class Example {

  private static final String TABLE_NAME = "MY_TABLE_NAME_TOO";
  private static final String CF_DEFAULT = "DEFAULT_COLUMN_FAMILY";

  public static void createOrOverwrite(Admin admin, HTableDescriptor table) throws IOException {
    if (admin.tableExists(table.getTableName())) {
      admin.disableTable(table.getTableName());
      admin.deleteTable(table.getTableName());
    }
    admin.createTable(table);
  }

  public static void createSchemaTables(Configuration config) throws IOException {
    try (Connection connection = ConnectionFactory.createConnection(config);
         Admin admin = connection.getAdmin()) {

      HTableDescriptor table = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
      table.addFamily(new HColumnDescriptor(CF_DEFAULT).setCompressionType(Algorithm.NONE));

      System.out.print("Creating table. ");
      createOrOverwrite(admin, table);
      System.out.println(" Done.");
    }
  }

  public static void modifySchema (Configuration config) throws IOException {
    try (Connection connection = ConnectionFactory.createConnection(config);
         Admin admin = connection.getAdmin()) {

      TableName tableName = TableName.valueOf(TABLE_NAME);
      if (!admin.tableExists(tableName)) {
        System.out.println("Table does not exist.");
        System.exit(-1);
      }

      HTableDescriptor table = admin.getTableDescriptor(tableName);

      // Update existing table
      HColumnDescriptor newColumn = new HColumnDescriptor("NEWCF");
      newColumn.setCompactionCompressionType(Algorithm.GZ);
      newColumn.setMaxVersions(HConstants.ALL_VERSIONS);
      admin.addColumn(tableName, newColumn);

      // Update existing column family
      HColumnDescriptor existingColumn = new HColumnDescriptor(CF_DEFAULT);
      existingColumn.setCompactionCompressionType(Algorithm.GZ);
      existingColumn.setMaxVersions(HConstants.ALL_VERSIONS);
      table.modifyFamily(existingColumn);
      admin.modifyTable(tableName, table);

      // Disable an existing table
      admin.disableTable(tableName);

      // Delete an existing column family
      admin.deleteColumn(tableName, CF_DEFAULT.getBytes("UTF-8"));

      // Delete a table (Need to be disabled first)
      admin.deleteTable(tableName);
    }
  }

  public static void main(String... args) throws IOException {
    Configuration config = HBaseConfiguration.create();

    //Add any necessary configuration files (hbase-site.xml, core-site.xml)
    config.addResource(new Path(System.getenv("HBASE_CONF_DIR"), "hbase-site.xml"));
    config.addResource(new Path(System.getenv("HADOOP_CONF_DIR"), "core-site.xml"));
    createSchemaTables(config);
    modifySchema(config);
  }
}    

作业：

region的结构及region相关概念  --简述
hbase表数据读写流程及写过程中的三种机制  --简述 
hbase java api的使用  -- 提交  
    

    
    

day3：

hbase与MapReduce集成使用  

hbase与hive集成使用  

hbase与sqoop集成使用    

hbase与Hue集成使用  

利用MapReduce并行计算框架对hbase表数据进行读写操作         
MapReduce程序 
    使用hbase自带mr-jar功能包 lib/hbase-server-1.2.0-cdh5.14.2.jar，实现向hbase中批量的导入或统计数据  
    自定义MapReduce程序向hbase表实现读写操作 
        

        

一、使用使用hbase自带MapReduce功能 jar包

lib/hbase-server-1.2.0-cdh5.14.2.jar              
    借助此jar包中的MapReduce程序可以实现利用MapReduce并行计算框架

                                    
        

1、使用自带的MapReduce功能jar包需要进行环境配置

配置jar包中的MapReduce程序在读写hbase表时作为一个客户端所需要的hbase的jar包依赖 
使这些MapReduce程序在执行时可以在当前环境中加载读取到hbasejar包 

找一个固定的会话窗口执行命令

$ export HBASE_HOME=/opt/cdh-5.14.2/hbase-1.2.0-cdh5.14.2    
$ export HADOOP_HOME=/opt/cdh-5.14.2/hadoop-2.6.0-cdh5.14.2
$ bin/hbase  mapredcp  -- 返回MapReduce程序在读写hbase表数据时所需要的hbase的jar包的本地路径 


$ export HADOOP_CLASSPATH=`bin/hbase  mapredcp`
将MapReduce程序读写hbase表数据时所需要的jar包本地路径加入到hadoop的环境变量中。之后MapReduce在执行时会自动获取HADOOP_CLASSPATH的值并从中获取所需的hbase jar包的路径 

$ echo ${HADOOP_CLASSPATH}   验证 
            

2、测试执行hbase自带的MapReduce-jar包程序

$ /opt/cdh-5.14.2/hadoop-2.6.0-cdh5.14.2/bin/yarn jar /opt/cdh-5.14.2/hbase-1.2.0-cdh5.14.2/lib/hbase-server-1.2.0-cdh5.14.2.jar
An example program must be given as the first argument.
Valid program names are:
CellCounter: Count cells in HBase table. 统计hbase表cell详情
WALPlayer: Replay WAL files.
completebulkload: Complete a bulk data load.
copytable: Export a table from local cluster to peer cluster. 利用MapReduce程序实现拷贝hbase表
export: Write table data to HDFS. 利用MapReduce程序从hbase表中导出数据
exportsnapshot: Export the specific snapshot to a given FileSystem.
import: Import data written by Export. 利用MapReduce程序向hbase导入数据
importtsv: Import data in TSV format. 利用MapReduce程序向hbase导入tsv格式数据
rowcounter: Count rows in HBase table. 利用MapReduce程序统计行

1）利用hbase自带的MapReduce程序统计hbase表的行

2）利用hbase自带的MapReduce程序实现向hbase表中批量导入数据 --- 直接导入方式

在hbase 上创建目标表 
    > create "student" ,"info"  
    
创建tsv测试文件并上传到hdfs上 
    $ vi student.tsv    
        10003   tom     boy     20
        10004   lili    girl    19
        10005   lio     boy     21
        10007   litao   boy     18
        10008   mary    girl    20
        
    $ bin/hdfs dfs -put student.tsv /user  
        
执行导入MapReduce任务 

    $ /opt/cdh-5.14.2/hadoop-2.6.0-cdh5.14.2/bin/yarn jar /opt/cdh-5.14.2/hbase-1.2.0-cdh5.14.2/lib/hbase-server-1.2.0-cdh5.14.2.jar  
    importtsv 
    -Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:sex,info:age   
    student 
    hdfs://192.168.134.101:8020/user/student.tsv               

3）使用bulk load方式将一个tsv格式的文件导入到hbase表中

    
直接导入方式： 
    regionserver响应客户端的写请求-》regionserver将数据写入Hlog-》写入memstore中=》从memstore从flush成storeFile文件-》compaction合并
    
    
bulk load方式导入： 
    先使用MapReduce程序将到导入的tsv文件直接转化为最终的storeFile文件并存储到hdfs上的某个目录下=》将最终的storeFile文件写入（移动）到目标表的store下
    
    
使用bulk load方式导入的好处： 
    bulk load方式导入可以避开直接导入数据时的hbase集群的资源消耗（内存、io、cpu）（避开了数据先写入hlog-》memstore-》flush等），将tsv文件转化为storeFile文件的工作压力转移给MapReduce分布式实现，最终hbase只需将转化后的storeFile文件移动写入到目标表中 


            
            
创建新的目标表 
        > create "student1","info"


        
过程1：文件格式转化（将tsv文件转化为storeFile文件）

        
    执行转化文件格式任务     
        -Dimporttsv.bulk.output=/path/for/output  在直接导入方式基础上次参数则当前MapReduce任务将会把要导入到hbase表中的tsv文件转换为Hfile文件，而不是直接写入到hbase表 
        /path/for/output为由MapReduce转化后的Hfile文件在hdfs上的存储路径 
        
        yarn jar /opt/modules/hbase-1.2.0-cdh5.14.2/lib/hbase-server-1.2.0-cdh5.14.2.jar importtsv -Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:sex,info:age -Dimporttsv.bulk.output=/path/for/output student1 hdfs://centos01:8020/user/student.tsv 
        
    现象： 
        此任务是MapReduce任务 
        > scan "student1"  此时数据还未写入到hbase表中  
        /path/for/output/info  --- Hfile文件  

        

过程2：将由MapReduce转化好的Hfile文件写入（移动）到目标region的store下 

    用completebulkload完成bulkload上传

    yarn jar /opt/modules/hbase-1.2.0-cdh5.14.2/lib/hbase-server-1.2.0-cdh5.14.2.jar completebulkload /path/for/output student3


    现象： 
        此任务不是MapReduce任务,此过程由hbase自行完成  
        > scan "student1"   数据已经写入 
        /path/for/output/info  --- Hfile文件已经消失
        /hbase/data/default/student1/1b9a3a4e47cc3859a0909dc096685260/info -出现新的文件  
        
        

二、自定义MapReduce程序完成读写hbase表数据

需求：     
    自定义MapReduce程序将hbase中的student表中的10004行到10007行之间的info:name和info:age 写入到 hbase的user表中 basic:XM ,basic:NL，rowkey不变  

    
    
 ReadStudentMapper extends  TableMapper 
    如果map从hbase表中读数据则自定义的Mapper类需要继承TableMapper  
    
        声明的是map端输出的key和value类型  
        ImmutableBytesWritable是对字节数组的封装类并继承了WritableComparable类型

        
        
map(ImmutableBytesWritable key, Result value, Context context)        
    ImmutableBytesWritable key  - 是从目标表中读取到的某行数据的rowkey值    
    Result value  -- 是从目标表中读取到的对应行所有cell的封装实例 

package com.hadoop.hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Student2UserMapReduce extends Configured implements Tool {

    // Step 1 : Mapper
    public static class ReadStudentMapper extends  TableMapper {

        @SuppressWarnings("deprecation")
        @Override
        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {

            // create put
            Put put = new Put(key.get()); //直接引用原表中的rowkey作为目标表的rowkey
            // 将student表中该行中的info:name info:age 两列数据过滤数并添加put实例中 
            for (Cell cell : value.rawCells()) {
                // get CF:info 过滤出特定列簇因为可能有多个列簇
                if ("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))) {
                    // add name
                    if ("name".equals(Bytes.toString(CellUtil
                            .cloneQualifier(cell)))) {
                        put.add(Bytes.toBytes("basic"), Bytes.toBytes("XM"),
                                CellUtil.cloneValue(cell));
                    }
                    // add age
                    else if ("age".equals(Bytes.toString(CellUtil
                            .cloneQualifier(cell)))) {
                        put.add(Bytes.toBytes("basic"), Bytes.toBytes("NL"),
                                CellUtil.cloneValue(cell));
                    }
                }
            }

            // context output
            context.write(key, put);
        }
    }

    // Step 2 : Reducer

    public static class WriteUserReducer extends
            TableReducer {

        @Override
        protected void reduce(ImmutableBytesWritable key, Iterable values,
                Context context) throws IOException, InterruptedException {

            for (Put put : values) {
                context.write(NullWritable.get(), put);
            }

        }

    }

    // Step 3 : Driver
    public int run(String[] args) throws Exception {
        // 1) Configuration
        Configuration conf = this.getConf();
        // 2) create job
        Job job = Job.getInstance(conf, this.getClass().getSimpleName());
        job.setJarByClass(Student2UserMapReduce.class);

        // 3) set job
        // set scan 设置map端的查询条件 ，
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes("10004"));
        scan.setStopRow(Bytes.toBytes("10007"));
        scan.addFamily(Bytes.toBytes("info"));  //只扫描特定的列簇
        // setMapper
        TableMapReduceUtil.initTableMapperJob(
                "student", // input table
                scan, // Scan instance
                ReadStudentMapper.class, // mapper class
                ImmutableBytesWritable.class, // mapper output key
                Put.class, // mapper output value
                job
                );

        // setReducer
        TableMapReduceUtil.initTableReducerJob(
                "user", // output table
                WriteUserReducer.class, // reducer class
                job);

        // set reduce nums
        job.setNumReduceTasks(1); // at least one, adjust as required

        boolean isSuccess = job.waitForCompletion(true);
        if (!isSuccess) {
            throw new IOException("error with job!");
        }
        return isSuccess ? 0 : 1;

    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        int status = ToolRunner.run(//
                conf, //
                new Student2UserMapReduce(), //
                args //
                );
        System.exit(status);

    }

}

测试执行 
    在hbase上提前创建目标表 
         create "user","basic"
    导出jar包并上传到linux集群并提交执行
        $ bin/yarn jar jars/s2u.jar  
    检查执行结果
        > scan "user"   

如何为hbase的master节点配置备份节点（tar包安装方式搭建的集群）  HMaster HA

防止master单节点故障
    虽然hbase表数据的读写不经过master，master宕机一段时间内集群还可以正常读写，当时还是有不可或缺的作用
如何实现
    Master HA的实现是借助于zookeeper基于观察者模式监控master状态
    一旦active master节点宕机后zookeeper会第一时间感知并从其他的多个备份master节点（backup-master）中选举出一个master作为后续的active master节点 
    

1、搭建Apache Hadoop集群并启动

$ sbin/start-dfs.sh              --启动HDFS

2、搭建zookeeper集群并启动

$ bin/zkServer.sh start

3、部署HBase集群

先在某台节点上操作，完成后再拷贝给其他节点
    $ vi conf/regionservers  //添加regionserver服务器主机名或IP
    $ vi conf/backup-masters     // 在HABASE_HOME/conf目录下添加backup-masters文件，里面定义哪些服务器是备用master
    $ vi conf/hbase-site.xml   //向hbase-site.xml中添加配置信息

              
                hbase.rootdir  
                hdfs://hadoop-senior01.bf.com:8020/hbase
            
            
                hbase.cluster.distributed
                true
                        
            
                hbase.zookeeper.quorum
                blue01.mydomain,blue02.mydomain,blue03.mydomain
            
            
                hbase.master
                hdfs://192.168.134.101:60000
            
            
            hbase.master  => 声明集群中的哪台服务器作为最初的active master节点

拷贝此台hbase安装目录到其他两个节点：
    $ scp -r hbase-0.98.6-hadoop2/ blue02.mydomain:/opt/modules/
    $ scp -r hbase-0.98.6-hadoop2/ blue03.mydomain:/opt/modules/

4、启动hbase服务进程

$ bin/start-hbase.sh  //在hbase.master 定义的服务器上执行该命令

5、观察每个服务器的角色

启动hbase服务后，会发现除hbase.master 定义的服务器上有Hmaster进程外
在conf/backup-masters内定义的服务器上也有master进程
active  master默认在hbase.master 定义的服务器上

6、测