设为首页 加入收藏

TOP

Hbase源码解析和开发实战笔记
2019-05-11 00:39:46 【大 中 小】 浏览:53次
Tags：Hbase 源码 解析 开发 实战 笔记
版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/china_demon/article/details/51760595
Hbase笔记
HBase基础
1）HBase简介
2）HBase特点
3）HBase数据模型
4）HBase体系结构
5）HBase存储模型
6）HBase应用




简介
1）HBase是一个分布式的、多版本的、面向列的开源数据库
2）HBase利用Hadoop HDFS作为其文件存储系统，提供高可靠性、高性能、列存储、可伸缩、实时读写、适用于非结构化数据存储的数据库系统
3）HBase利用Hadoop MapReduce来处理HBase中的海量数据
4）HBase利用Zookeeper作为分布式协同服务


特点：
1）数据量大：一个表可以有上亿行，上百万列（列多时，插入变慢）
2）面向列：面向列（族）的存储和权限控制，列（族）独立检索
3）稀疏：对于为空（null）的列，并不占用存储空间，因此，表可以设计的非常稀疏
4）多版本：每个cell中的数据可以有多个版本，默认情况下版本号自动分配，是单元格插入时的时间戳
5）无类型：HBase中的数据都是字符串，没有类型
6）强一致性：同一行数据的读写只在同一台Region Server上进行
7）有限查询方式：仅支持三种查询方式（单个rowkey查询，通过rowkey的range查询，全表扫描）
8）高性能随机读写


数据模型
1）行：同一个key对应的所有数据
2）列族：相似的列数据通常被划分成一个列族，建表时确定
3）列：列名在写入时确定
4）Cell及时间戳（版本）：
每个cell有任意多的版本
建表时设置每个列族可以保留多少个版本
5）三维有序
SortedMap(RowKey,List(SortedMap(Column，List(Value，TimeStamp))))


rowkey(ASC)+columnLabel(ASC)+Version(DESC)->value
|------------|--------------|--------------------|----------------------------|---------------|
|Row key |Time Stamp | Column "contents:" | Column "anchor:" | Column "mime:"|
|------------|--------------|--------------------|------------------|---------| |
|com.cnn.www |t9 | | anchor:cnnsi.com |CNN | |
|------------|--------------|--------------------|------------------|---------|---------------|
| |t8 | | achor:my.look.ca |CNN.com | |
|------------|--------------|--------------------|------------------|---------|---------------|
| |t6 | "<html>..." | | "text/html" |
|------------|--------------|--------------------|----------------------------|---------------|
| |t5 | "<html>..." | | |
|------------|--------------|--------------------|----------------------------|---------------|
| |t3 | "<html>..." | | |
|------------|--------------|--------------------|----------------------------|---------------|
--------------------------------------------------
1)面向列的存储
2)一张表可以被划分成若干个region
3)行按照rowkey进行字典序排序
4)支持随机读写
5）region是负载均衡调度的最小单位
------------------------------------------------------------------------------
体系结构
1）Client
包含访问HBase的接口并维护cache来加快对HBase的访问
2）Zookeeper
保证任何时候，集群中只有一个master
存储所有Region的寻址入口：
实时监控Region server的上线和下线信息，并实时通知给Master；
存储HBase的schema和table元数据
3）Master
为Region server分配region；
负责Region server的负载均衡
发现失效的Region server并重新分配其上的 region
管理用户对table的增删改查操作
4）Region Server
负责维护region,处理对这些region的IO请求
负责切分在运行过程中变得过大的region
5）ROOT表
记录META表中的每个region的位置，ROOT表最多只有一个region Zookeeper中记录了ROOT表的location
6）META表
记录各个表每个region所在的region server，META表可能包含多个region
------------------------------------------------------------------------------------------
HBase操作
1）flush
内存容量有限，需要定期将内存中的数据flush到磁盘
每次flush，每个region的每个column family都会产生一个HFile
读取操作，region server会把多个HFile数据归并到一起


2）compaction
flush操作产生的HFile会越来越低，需要归并来减少HFile的数量
旧数据会被清理


3）split
HFile大小增长到某个阀值就会split，同时把Region split成两个region，这两个region被分发到其他不同的region server上


4）scan
hbase原生提供的方法，顺序扫库；当然可以使用MapReduce并发扫库的方法


5）Bulk Load
快速导入大批量数据的方法
-------------------------------------------------------------------------------------------
存储模型
1）一个table中的一个region会被随机分配给一个region server
2）region是分布式存储和负载均衡的最小单位
------------------------------------------------------
最开始table只有一个region，但是随着数据的put，到达某个阀值的时候，一个大的region会split成几个小region，被分配到其他region server
------------------------------------------------
HFile结构
1）基于Block的存储结构
2）Block的索引驻留内存
---------------------------------
HBase应用
1）HBase环境搭建
配置hbase-env.sh
hbase-site.xml
regionservers
运行start-hbase.sh
启动后执行hbase shell进入交互式命令行


2）HBase API使用举例
Put/Get
Scan
.....


-----------------------------------------------------------------------------------------------------------
Hbase_test.java
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;




public class Hbase_test {
public static void main(String[] args){
Configuration conf = new Configuration();
HBaseConfiguration hbconf = new HBaseConfiguration(conf);

/*try {
HBaseAdmin admin = new HBaseAdmin(hbconf);
HTableDescriptor tableDesc = new HTableDescriptor("test");
tableDesc.addFamily(new HColumnDescriptor("info"));
admin.createTable(tableDesc);

} catch (MasterNotRunningException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ZooKeeperConnectionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}*/

try {
HTable table = new HTable(hbconf,"test");
/*Put put = new Put("row1".getBytes());
put.add("info".getBytes(),"A".getBytes(),"1".getBytes());
table.put(put);
table.close();*/

Scan scan = new Scan();
ResultScanner res = table.getScanner(scan);
for(Result r:res){
for(KeyValue kv: r.raw()){
System.out.println("rowkey=> " + new String(r.getRow())
+ " family=> " + new String(kv.getFamily())
+ " qualifier=> " + new String(kv.getQualifier())
+ " timestamp=> " + kv.getTimestamp()
+ " value=> " + new String(kv.getValue()));
}


}

} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
----------------------------------------------------------------------------------------------------------------------
HBase系统搭建与部署
1）开源社区版本下载
-http://hbase.apache.org/
2)Cloudera发行版下载
-http://archive-primary.cloudera.com/cdh5/cdh/5/


Hadoop2.0部署
1）配置hosts 确保涉及的主机名均可以解析
2）编辑hadoop-env.sh,mapred-env.sh,yarn-env.sh
3）编辑core-site.xml,hdfs-site.xml,yam-site.xml
5）编辑slaves文件
6)把Hadoop复制到其他节点
7)启动Hadoop
8）验证启动


Hadoop2.0 HA部署
采用QJM方案
1、启动JN
./sbin/hadoop-daemon.sh start jourmalnode
./sbin/hadoop-daemons.sh start journalnode
hdfs namenode -initialize SharedEdits


2、启动Active NN
hdfs namenode -format
bin/hadoop-daemon.sh start namenode


3、启动Standby NN
hdfs namenode -bootstrap Standby
./sbin/hadoop-daemon.sh start namenode


4、启动Automatic Failover
hdfs zkfc -formatZK
./sbin/hadoop-daemon.sh start zkfc


hdfs haadmin -transitionToStandby nn1


hdfs haadmin -failover nn1 nn2


hdfs haadmin -getServiceState nn1
-----------------------------------------------------------------------------------------------------------------------------
HBase Shell 操作
（1）建立一个表scores,有两个列族grad和courese
>>create 'scores','grade','course'
(2)查看Hbase中的所有表
>>list
(3)查看表结构
>>describe 'scores'


Shell操作
4、按设计的表结构插入值
put 'scores','Tom','grade:','5'
put 'scores','Tom','course:math','97'
put 'scores','Tom','course:art','87'
put 'scores','Jim','grade:','4'
put 'scores','Jim','course:','89'
put 'scores','Jim','course:','80'


put命令比较简单，只有这一种用法：
put 't1','r1','c1','value',ts1
t1指表名，r1指定键名，c1指列名，value指单元格值。ts1指时间戳，一般都省略掉了。


5、根据键值查询数据
>>get 'scores','Jim'
>>get 'scores','Jim','grade'
get 't1','r1'
get 't1','r1',{TIMERANGE=>[ts1,ts2]}
get 't1','r1',{COLUMN=>'c1'}
get 't1','r1',{COLUMN=>['c1','c2','c3']}
get 't1','r1',{COLUMN=>'c1',TIMESTAMP=>ts1}
get 't1','r1',{COLUMN=>'c1',TIMERANGE=>[ts1,ts2],VERSIONS=>4}
get 't1','r1',{COLUMN=>'c1',TIMESTAMP=>ts1,VERSION=>4}
get 't1','r1','c1'
get 't1','r1','c1','c2'
get 't1','r1',['c1','c2']


6、扫描所有数据
scan 'scores'
也可以指定一些修饰词：TIMERANGE,FILTER,LIMIT,STARTROW,STOPROW,TIMESTAMP,MAXLENGTH,or COLUMNS.没任何修饰词，就是上边例句，就会显示所有数据行


scan '.META.'
scan '.META.',{COLUMNS=>'info:regioninfo'}
scan 't1',{COLUMNS=>['c1','c2'],LIMIT=>10,STARTROW=>'xyz'}
scan 't1',{COLUMNS=>'c1',TIMERANGE=>[1303668804,1303668904]}
scan 't1',{FILTER=>"(PrefixFilter('row2')AND(QualifierFilter(>=,'binary:xyz')))AND(TimestampsFilter(123,456))"}
scan 't1',{FILTER=>org.apache,hadoop,hbase.filter,ColumnPaginationFilter.new(1,0)}


7、删除指定数据
delete 'scores','Jim','grade'
delete 'scores','JIm'
删除数据命令也没太多变化，只有一个：
delete 't1','r1','c1','ts1'
另外有一个deleteall命令，可以进行整行的范围的删除操作，慎用！
如果需要进行全表删除操作，就使用truncate命令，其实没有直接的全表删除命令，这个命令也是disable，drop，create三个命令组合出来的。


8、修改表结构
disable 'scores'
alter 'scores',NAME=>'info'
enable 'scores'
alter命令使用如下（如果无法成功的版本，需要先通过表disable）：
a、改变或添加一个列族
alter 't1',NAME=>'f1',VERSIONS=>5
b、删除一个列族：
alter 't1',NAME=>'f1',METHOD=>'delete'
alter 't1','delete'=>'f1'


9、统计行数
>>count 't1'
>>count 't1',INTERVAL=>100000
>>count 't1',CACHE=>1000
>>count 't1',INTERVAL=>10,CACHE=>1000


count一般会比较耗时，使用MapReduce进行统计，统计结果会缓存，默认是10行。统计间隔默认的是1000行（INTERVAL）


10、表的删除
先禁用表 distable 'table'
删除表 drop 'table'
------------------------
客户端API使用操作


|--------------------|--------------------------------|
|Java类 | HBase数据模型 |
|--------------------|--------------------------------|
|HBaseAdmin | 数据库（DataBase） |
|HBaseConfiguration | |
|--------------------|--------------------------------|
|HTable | 表（Table） |
|--------------------|--------------------------------|
|HTableDescriptor | 列族（Column Family） |
|--------------------|--------------------------------|
|Put | |
|Get | 列修饰符（Column Qualifier）|
|Scanner | |
|--------------------|--------------------------------|


HBaseConfiguration
包名：org.apache.hadoop.hbase.HBaseConfiguration
作用: 对HBase进行配置
用法示例：
HBaseConfiguration hconfig = new HBaseConfiguration();
hconfig.set("hbase.zookeeper.property.clientPort","2181");
|--------|-----------------------------------------|------------------------------------------------------------------------|
|返回值 | 函数 | 描述 |
|--------|-----------------------------------------|------------------------------------------------------------------------|
|void | addResource(Path file) |通过给定的路径所指的文件来添加资源 |
|--------|-----------------------------------------|------------------------------------------------------------------------|
|void | deal() |清空所有设置的属性 |
|--------|-----------------------------------------|------------------------------------------------------------------------|
|string | get(String name) |获取属性名对应的值 |
|--------|-----------------------------------------|------------------------------------------------------------------------|
|String | getBoolean |获取为boolean类型的属性值，如果其属性值类型部位boolean，则返回默认属性值|
|--------|-----------------------------------------|------------------------------------------------------------------------|
|void | set(String name,String value) |通过属性名来设置值 |
|--------|-----------------------------------------|------------------------------------------------------------------------|
|void | setBoolean(String name,boolean value) |设置boolean类型的属性值 |
|--------|-----------------------------------------|------------------------------------------------------------------------|


HBaseAdmin
包名：org.apache.hadoop.hbase.client.HBaseAdmin
作用：提供了一个接口来管理HBase数据库的表信息。它提供的方法包括：创建表，删除表，列出表项，使表有效或无效，以及添加或删除表列族成员等。
用法示例：
HbaseAdmin admin = new HBaseAdmin(config);
admin.disableTable("tablename")


|------------------------|--------------------------------------------------------|-----------------------------------------------------|
|返回值 | 函数 | 描述 |
|------------------------|--------------------------------------------------------|-----------------------------------------------------|
|void | addColumn(String tableName,HColumnDescriptor column) | 向一个已经存在的表添加列 |
| |--------------------------------------------------------|-----------------------------------------------------|
| | checkHBaseAvailable(HBaseConfiguration conf) | 静态函数，查看HBase是否处于运行状态| |
| |--------------------------------------------------------|-----------------------------------------------------|
| | create Table(HTableDescriptor desc) | 创建一个表，同步操作 |
| |--------------------------------------------------------|-----------------------------------------------------|
| | delete Table(byte[] tableName) | 删除一个已经存在的表 |
| |--------------------------------------------------------|-----------------------------------------------------|
| | enable Table(byte[] tableName) | 是表处于有效状态 |
| |--------------------------------------------------------|-----------------------------------------------------|
| | disable Table(byte[] tableName) | 使表处于无效状态 |
|------------------------|--------------------------------------------------------|-----------------------------------------------------|
|HTableDescription[] | list Tables() | 列出所有用户控件表项 |
|------------------------|--------------------------------------------------------|-----------------------------------------------------|
|void | modify Table(byte[] tableName,HTableDescriptor htd) | 修改表的模式，是异步的操作，可能需要花费一定的时间 |
|------------------------|--------------------------------------------------------|-----------------------------------------------------|
|boolean | tableExists(String tableName) | 检查表是否存在 |
|------------------------|--------------------------------------------------------|-----------------------------------------------------|


HTableDescriptor
包名：org.apache.hadoop.hbase.HTableDescriptor
作用：包含了表的名字极其对应表的列族
用法示例：
HTableDescriptor htd = new HTableDescrptor(table);
htd.addFamily(new HcolumnDescriptor("family"));
|------------------|-----------------------------------|--------------|
|返回值 | 函数 | 描述 |
|------------------|-----------------------------------|--------------|
|void | addFamily(HColumnDescriptor) | 添加一个列族 |
|------------------|-----------------------------------|--------------|
|HColumnDescriptor | removeFamily(byte[] column) | 移除一个列族 |
|------------------|-----------------------------------|--------------|
|byte[] | getName() | 获取表的名字 |
|------------------|-----------------------------------|--------------|
|byte[] | getValue(byte[] key) | 获取属性的值 |
|------------------|-----------------------------------|--------------|
|void | setValue(String key,String value) | 设置属性的值 |
|------------------|-----------------------------------|--------------|


HColumnDescriptor
包名：org.apache.hadoop.hbase.HColumnDescriptor
作用：维护着关于列族的信息，例如版本号，压缩设置等。它通常在创建表或者为表添加列族的时候使用。列族被创建后不能直接修改，只能通过删除然后重新创建的方式。列族被删除的时候，列族里面的数据也会同时被删除。
用法示例：
HTableDescriptor htd = new HTableDescriptor(tablename);
HColumnDescriptor col = new HColumnDescriptor("content");
htd.addFamily(col);


|-------|-----------------------------------|------------------------|
|返回值 | 函数 | 描述 |
|-------|-----------------------------------|------------------------|
|byte[] | getName() | 获取列族的名字 |
|-------|-----------------------------------|------------------------|
|byte[] | getValue(byte[] key) | 获取对应的属性的值 |
|-------|-----------------------------------|------------------------|
|void | setValue(String key,String value)| 设置对应属性的值 |
|-------|-----------------------------------|------------------------|


HTable
包名：org.apache.hadoop.hbase.client.HTable
作用：可以用来和HBase表直接通信。此方法对于更新操作来说是非线程安全的
用法示例：
HTable table = new HTable(conf,Bytes.toBytes(tablename));
ResultScanner scanner = table.getScanner(family);
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|返回值 | 函数 | 描述 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|void | checkAdnPut(byte[] row,byte[] family,byte[] qualifier,byte[] value,Put put)| 自动的检查row/family/qualifier是否与给定值匹配 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|void | close() | 释放所有的资源或挂起内部缓冲中的更新 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|Boolean | exists(Get get) | 检查Get实例所指定的值是否存在于HTable的列中 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|Result | get(Get get) | 获取指定行的某些单元格所对应的值 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|byte[][] | getEndKeys() | 获取当前一打开的表每个区域的结束键值 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|ResultScanner | getScanner(byte[] family) | 获取当前给定列族的scanner实例 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|HtableDescriptor | getTableDescriptor() | 获取当前表的HTableDescriptor实例 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|byte[] | getTableName() | 获取表名 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|static boolean | isTableEnabled(HBaseConfiguration conf,String tableName) | 检查表是否有效 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|
|void | put(Put put) | 向表中添加值 |
|------------------|----------------------------------------------------------------------------|--------------------------------------------------|


Put
包名：org.apache.hadoop.hbase.client.Put
作用：用来对单个行执行添加操作
用法示例：
Htable table = new HTable(conf.Bytes.toBytes(tablename));
Put p = new Put(brow);//为指定行创建一个Put操作
p.add(family.qualifier.value);
table.put(p);


|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|返回值 | 函数 | 描述 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|void | checkAndPut(byte[] row,byte[] family,byte[] qualifier,byte[] value,Put put) |自动的检查row/family/qualifier是否与给定的值匹配 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|void | close() |释放所有的资源或挂起内部缓冲区中的更新 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|Boolean | exists(Get get) exists(Get get) |检查Get实例所指定的值是否存在于HTable的列中 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|Result | get(Get get) |获取指定行的某些单元格所对应的值 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|byte[][] | getEndKeys() |获取当前一打开的表每个区域的结束键值 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|ResultScanner | getScanner(byte[] family) |获取当前给定列族的scanner实例 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|HTableDescriptor | getTableDescriptor() |获取当前表的HTableDescription实例 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|byte[] | getTableName() |获取表名 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|static boolean | is TableEnabled(HBaseConfiguration conf,String tableName) |检查表是否有效 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|
|void | put(Put put) |向表中添加值 |
|--------------------|-----------------------------------------------------------------------------------------|-----------------------------------------------------|


Get
包名：org.apache.hadoop.hbase.client.Get
作用:用来获取单个行的相关信息
用法示例：
HTable table = new HTable(conf,Bytes.toBytes(tablename));
Get g = new Get(Bytes.toBytes(row));
table.get(g);


|--------|-------------------------------------------|-----------------------------------|
|返回值 |函数 | 描述 |
|--------|-------------------------------------------|-----------------------------------|
|Get |addColumn(byte[] family,byte[] qualifier) |获取指定列族和列修饰符对应的列 |
|--------|-------------------------------------------|-----------------------------------|
|Get |addFamily(byte[] family) |通过指定的列族获取对应的所有列 |
|--------|-------------------------------------------|-----------------------------------|
|Get |setTimeRange(long minStamp,long maxStamp) |获取指定取件的列的版本号 |
|--------|-------------------------------------------|-----------------------------------|
|Get |setFilter(Filter filter) |当执行Get操作时设置服务器端的过滤器|
|--------|-------------------------------------------|-----------------------------------|


Result
包名:org.apache.hadoop.hbase.client.Result
作用：存储Get或者Scan操作后获取表的单行值。使用此类提供的方法可以直接获取值或者各种Map结构（key-value对）


|----------------------------|------------------------------------------------|-----------------------------------------------|
|返回值 | 函数 | 描述 |
|----------------------------|------------------------------------------------|-----------------------------------------------|
|boolean | containsColumn(byte[] family,byte[] qualifier) | 检查指定的列是否存在 |
|----------------------------|------------------------------------------------|-----------------------------------------------|
|NavigableMap<byte[],byte[]> | getFamilyMap(byte[] family) | 获取对应列族所包含的修饰符与值的键值对 |
|----------------------------|------------------------------------------------|-----------------------------------------------|
|byte[] | getValue(byte[] family,byte[] qualifier) | 获取对应列的最新值 |
|----------------------------|------------------------------------------------|-----------------------------------------------|


ResultScanner
包名：org.apache.hadoop.hbase.client.ResultScanner
作用:存储Get或者Scan操作后获取表的单行值。使用此类提供的方法可以直接获取值或者各种Map结构(key-value对)


|--------|---------|----------------------------------|
|返回值 |函数 | 描述 |
|--------|---------|----------------------------------|
|void |close() | 关闭scanner并释放分配给它的资源|
|--------|---------|----------------------------------|
|Result |next() | 获取下一行的值 |
|--------|---------|----------------------------------|


HTablePool
包名：org.apache.hadoop.hbase.client.HTablePool
作用：可以解决HTable存在的线程不安全问题，同时通过维护固定数量的HTable对象，能够在程序运行期间复用这些HTable资源对象
说明：
1、HTablePool可以自动创建HTable对象，而且对客户端来说使用上是完全透明的，可以避免多线程间数据并发修改问题。
2、HTablePool中的HTable对象之间是公用Configuration连接的，能够可以减少网络开销。
HTablePool的使用很简单：每次进行操作前，通过HTablePool的getTable方法HTabelPool的putTable方法将HTable对象放回到TablePool中。


------------------------------------------------------------------------------------------------------------------
create 'students_age','f1'
-------------------------创建表程序-----------------------------------------------------------------------------------------
package com.hbase;


import java.io.IOException;
import java.util.LinkedList;
import java.util.List;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.util.Bytes;


public class HbaseConnection{
private String rootDir;
private String zkServer;
private String port;


private Configuration conf;
private HConnection hConn = null;
public HbaseConnection(String rootDir,String zkServer,String port){
this.rootDir = rootDir;
this.zkServer = zkServer;
this.port = port;
conf = HBaseConfiguration.create();
conf.set("hbase.rootdir",rootDir);
conf.set("hbase.zookeeper.quorum",zkServer);
conf.set("hbase.zookeeper.property.clientPort",port);
try{
hConn = HConnectionManager.createConnection(conf);
}catch(IOException e){
e.printStackTrace();
}
}
public void createTable(String tableName,List<String> cols){
try{
HBaseAdmin admin = new HBaseAdmin(conf);
if(admin.tableExists(tableName))
throw new IOException("table exists");
else{
HTableDescriptor tableDesc = new HTableDescriptor(tableName);
for(String col : cols){
HColumnDescriptor colDesc = new HColumnDescriptor(col);
colDesc.setCompressionType(Algorithm.GZ);
colDesc.setDataBlockEncoding(DataBlockEncoding.DIFF);
tableDesc.addFamily(colDesc);
}
admin.createTable(tableDesc);
}


}catch(MasterNotRunningException e){
e.printStackTrace();
}catch(ZooKeeperConnectionException e){
e.printStackTrace();
}catch(IOException e){
e.printStackTrace();
}

}


public void saveData(String tableName,List<Put> puts){
try{
HTableInterface table = hConn.getTable(tableName);
table.put(puts);
table.setAutoFlush(false);
table.flushCommits();
}catch(IOException e){
e.printStackTrace();
}
}


public Result getData(String tableName,String rowkey){
try{
HTableInterface table = hConn.getTable(tableName);
Get get = new Get(Bytes.toBytes(rowkey));
return table.get(get);
}catch(IOException e){
e.printStackTrace();
}

return null;
}

public void format(Result result){
String rowkey = Bytes.toString(result.getRow());
KeyValue[] kvs = result.raw();
for(KeyValue kv: kvs){
String family = Bytes.toString(kv.getFamily());
String qualifier = Bytes.toString(kv.getQualifier());
System.out.println("rowkey->" + rowkey + "family->" + family + "qualifier->" + qualifier);
}
}

public void hbaseScan(String tableName){
Scan scan = new Scan();
scan.setCaching(1000);
try{
HTableInterface table = hConn.getTable(tableName);
ResultScanner scanner = table.getScanner(scan);
for(Result res: scanner){
format(res)
}
}catch(IOException e){
e.printStackTrace();
}
}


public void filterTest(String tableName){
Scan scan = new Scan():
scan.setCaching(1000);
RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,
///new BinaryComparator(Bytes.tobytes("Jack")));
new RegexStringComparator("J\\w+")));


PageFilter filter = new PageFilter(15);//分页显示
scan.setFilter(filter);
try{
HtableIntegerface table = hConn.getTable(tableName);
ResultScanner scanner = table.getScanner(scan);


for(Result res: scanner){
format(res);
}
}catch(IOException e){
e.printStackTrace();
}
}




public void pageFilterTest(String tableName){
PageFilter filter = new PageFilter(4);
byte[] lastRow = null;
int pageCount = 0;
try{
HtableInterface table = hConn.getTable(tableName);
while(++pageCount > 0){
System.out.println("pageCount = " + pageCount);
Scan scan = new Scan();
scan.setFilter(filter);
if(lastRow != null){
scan.setStartRow(lastRow);
}
ResultScanner scanner = table.getScanner(scan);
for(Result res: scanner){
format(res);
lastRow = res.getRow();
if(++count > 3)
break;
format(res);
}
if(count < 3)
break;

}
}catch(IOException e){
e.printStackTrace();
}



}


public static void main(String[] args){
String rootDir = "hdfs://hbase";
String zkServer = "192.168.1.198";
String port = "2181";
HbaseConnection conn = new HbaseConnection(rootDir,zkServer,port);
/* List<String> cols = new LinkedList<>();
cols.add("basicInfo");
cols.add("moreInfo");
conn.createTable("students",cols);*/


List<Put> puts = new LinkedList<Put>();
Put put1 = new Put(Bytes.toBytes("Tom"));
put1.add(Bytes.toBytes("basicInfo"),Bytes.toBytes("age"),Bytes.toBytes("27"));
put1.add(Bytes.toBytes("moreInfo"),Bytes.toBytes("tel"),Bytes.toBytes("11232"));

Put put2 = new Put(Bytes.toBytes("Jim"));
put2.add(Bytes.toBytes("basicInfo"),Bytes.toBytes("age"),Bytes.toBytes("28"));
put2.add(Bytes.toBytes("moreInfo"),Bytes.toBytes("tel"),Bytes.toBytes("11233"));
puts.add(put1);
puts.add(put2);
conn.saveData("students",puts);
Result result = conn.getData("students","Tom");
conn.format(result);


conn.hbaseScan("studnets");
conn.filterTest("students");
}
}
-------------------------------------------------------------------------------------------------------------------------




machine


1、公网ip
内网ip




2-4 内网ip


log4j.properties
log4j.rootLogger = info,stdout,hbase
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = %d{ABSOLUTE} %5p %c{1}:%L - %m%n


log4j.appender.hbase = org.apache.log4j.DailyRollingFileAppender
log4j.appender.hbase.File = ./log.log
log4j.appender.hbase.Append = true
log4j.appender.hbase.layout = org.apache.log4j.PatternLayout
log4j.appender.hbase.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss} [%t:%r] - [%p] %m%n


---------------------------------------------------------------------------------------------------------
HBase 扫描器
Scanner扫描器
Hbase在扫描数据的时候，使用scanner表扫描器。
HTable通过一个Scan实例，调用getScanner（scan）来获取扫描器。可以配置扫描起止位，以及其他的过滤条件。


通过迭代器返回查询结果，使用起来虽然不是很方便，不过并不复杂。但是这里有一点可能被忽略的地方，就是返回的scanner迭代器，每次调用next的获取下一条记录的时候，默认配置下会访问一次RegionServer。这在网络不是很好的情况下，对性能的影响是很大的。
建议配置扫描器缓存。






扫描器使用
扫描器缓存


hbase.client.scanner.caching配置项可以设置HBase scanner一次从服务端抓取的数据条数，默认情况下一次一条。通过将其设置成一个合理的值，可以减少scan过程中next()的而时间开销，代价是scanner需要通过客户端的内存来维持这些被cache的行记录。


有三个地方可以进行配置：
1）在HBase的conf配置文件中进行配置
2）通过调用HTable.setScannerCaching(int scannerCaching)进行配置
3）通过调用Scan.setCaching(int caching)进行配置。三者的优先级越来越高。


Hbase过滤器
过滤器简介
1、使用过滤器可以提高操作表的效率，HBase中两种数据读取函数get()和scan（）都支持过滤器，支持直接访问和通过指定起止行键来访问，但是缺少细粒度的筛选功能，如基于正则表达式对键或值进行筛选的功能。
2、可以使用预定义好的过滤器或者是实现自定义过滤器
3、过滤器在客户端创建，通过RPC传送到服务器端，在服务器端执行过滤操作，把数据返回给客户端


scan 'studnets',{STARTROW=>'J',ENDROW=>'K'}


过滤器：
1、Comparsion Filter(比较过滤器)
RowFilter
FamilyFilter
QualifierFilter
ValueFilter
DependentColumnFilter


2、Dedicated Filter(专用过滤器)
SingleColumnValueFilter
SingleColumnValueExcluderFilter
PrefixFilter
PageFilter
KeyOnlyFilter
FirstKeyOnlyFilter
TimestampsFilter
RandomRowFilter


3、Decorating Filter（附加过滤器）
SkipFilter
WhileMatchFilters




HBase协处理器


HBase作为列数据库最经常被人的特性包括:
无法轻易建立“二级索引”，难以执行求和、计数、排序等操作。


比如，在旧版本的(<0.92)HBase中，统计数据表的总行数，需要使用Counter方法，执行一次MapReduce Job才能得到。虽然HBase在数据存储层中集成了MapReduce，能够有效用于数据表的分布式计算。
然而在很多情况下，做一些简单的相加或者聚合计算的时候，如果直接将计算过程放置在server端，能够减少通讯开销，从而获得很好的性能提升。于是，HBase在0.92以后引入了处理器（coprocessors），
实现了一些激动人心的新特性：能够轻易建立二次索引、复杂过滤器以及访问控制等。


HBase协处理器的灵感来自于JEFFDean 09年的演讲。它根据该演讲实现了类似于bigtable的协处理器，包括以下特性：
1）每个表服务器的任意子表都可以运行代码
2）客户端的高层调用接口（客户端能够直接访问数据表的行地址，多行读写会自动分片成多个并行的RPC调用）
3）提供一个非常灵活的、可用于建立分布式服务的数据模型
4）能够自动化扩展、负载均衡、应用请求路由
HBase的协处理器灵感来自bigtable，但是实现细节不尽相同。HBase建立一个框架，它为用户提供类库和运行时环境，使得他们的代码能够在HBase region server和master上处理。

协处理器分两种类型，系统协处理器可以全局导入region servre上的所有数据表，表协处理器即是用户可以指定一张表使用协处理器。
协处理器框架为了更好支持其行为的灵活性，提供了两个不同方面的插件。一个观察者（observer），类似于关系数据库的触发器。另一个是终端（endpoint），动态的终端有点像存储过程。




Observer
观察者的设计意图是允许用户通过代码来重载处理器框架的upcall方法，而具体的事件触发的callback方法有HBase的核心代码来执行。处理器框架处理所有的callback调用细节，协处理器自身只需要插入添加或者改变的功能。
以Hbase0.92版本为例，它提供了三种观察者接口：


RegionObserver：提供客户端的数据操纵事件钩子：Get、Put、Delete、Scan等。
WALObserver：提供WAL相关操作钩子。
MasterObserver：提供DDL-类型的操作钩子。如创建、删除、修改数据表等。
这些接口可以同时使用在同一个地方，按照不同优先级顺序执行，用户可以任意基于协调处理器实现复杂的HBase功能层。HBase有很多种事件可以触发观察者方法，这些事件与方法从HBase0.92版本起，都会集成在HBase API中。不过这些API可能会由于各种原因有所改动，不同版本的接口改动比较大。
-----------------------------------------------------------------
EndPoint
终端是动态RPC插件的接口，它的实现代码被安装在服务器端，从而能够通过HBase RPC唤醒。客户端类库提供了非常方便的方法来调用这些动态接口，它们可以在任意时候调用一个终端，它们的实现代码会被目标region远程执行，结果会返回到终端。
用户可以结合使用这些强大的插件接口，为HBase添加全新的特性。终端的使用，如下面流程所示：
定义一个新的protocol接口，必须继承CoprocessorProtocol。
实现终端接口，该实现会被导入region环境执行。
继承抽象类BaseEndpointCoprocessor。
在客户端，终端可以被两个新的Hbase Client API调用。单个region：
HTableInterface.coprocessorProxy(Class<T> protocol,byte[] row).regions区域：HTableInterface.coprocessorExec(Class<T> protocol,byte[] startkey,byte[] endKey,Batch.Call<T,R> callable)


有三个方法对Endpoint进行设置：
A.启动全局aggregation，能过操纵所有的表上的数据。通过修改hbase-site。xml这个文件来实现，只需要添加如下代码：
<property>
<name>hbase.coprocessor.user.region.classes</name>
<value>org.apache.hadoop.hbase.coprocessor.RowCountEndpoint</value>
</property>


B.启动表aggregation，只对特定的表生效。通过HBase Shell来实现。
(1)disable指定表。
hbase>> disable 'students'
(2)添加aggregation hbase>> alter ‘students’,METHOD =>'table_att','coprocessor'=>'|org.apache.hadoop.hbase.coprocessor.RowCountEndpoint||'
alter ‘students’,'coprocessor'=>'|org.apache.hadoop.hbase.coprocessor.example.RowCountEndpoint||'


协处理器命令：
增加协处理器方法一：alter 'students','coprocessor'=>'|org.apache.hadoop.hbase.coprocessor.example.RowCountEndpoint||'
-----------------------------------------------------------------------------------------------------------------------------------------------------
增加协处理器方法二：


package coprocessor;
import java.io.IOException;
import java.util.List;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.util.Bytes;


public class RegionObserverTest extends BaseRegionObserver{
private static byte[] fixed_rowkey = Bytes.toBytes("John");


@Override
public void preGet(ObserverContext<RegionCoprocessorEnvironment> c,
Get get, List<KeyValue> result) throws IOException {
// TODO Auto-generated method stub
if(Bytes.equals(get.getRow(), fixed_rowkey)){
KeyValue kv = new KeyValue(get.getRow(),Bytes.toBytes("time"),Bytes.toBytes("time"),Bytes.toBytes(System.currentTimeMillis()));
result.add(kv);
}
}
}


alter 'students','coprocessor'=>'hdfs://192.168.1.198:8020/coprocessor.jar|coprocessor.RegionObserverTest||'
enable 'students'


java -cp .:/usr/lib/hbase/lib/* HbaseConnection


备注：程序打包，上传到hdfs根目录下，执行alter。


----------------------------------------------------------------------------------------------------------------------------------------------------------


enable 'students'
describe 'students'
删除协处理器
alter 'students',METHOD => 'table_att_unset',NAME =>'coprocessor$1'




sudo jar tf /usr/lib/hbase/lib/hbase-examples-1.0.0-cdh5.5.0.jar | grep RowCountEndpoint


org/apache/hadoop/hbase/coprocessor/example/RowCountEndpoint.class


hadoop fs -put coprocessor.jar /




javac -cp /home/cloudera/softback/hbase-0.96.2-hadoop2/lib/* HbaseConnection.java
java -cp /usr/lib/hbase/lib/* HbaseConnection




(3)重启指定表hbase > enable 'mytable'


C.API调用
HTableDescriptor htd = new HTableDescriptor("testTable");
htd.setValue("CORPROCESSOR$1",path.toString + "|" + RowCountEndpoint.class.getCanonicalName() + "|" + Coprocessor.Priority.USER);
---------------------------------------------------------
CoprocessorRowCounter.java


public class CoprocessorRowCounter{
public static void main(String[] args) throws IOException{
Configuration conf = HbaseConfiguration.create();
HTable table = new HTable(conf,"students");
final ExampleProtos.CountRequest request = ExampleProcess.CountRequest.getDefaultInstance();
Map<byte[],Long> results = table.coprocessorService(ExampleProtos.RowCountService.class,null,null,new Batch.Call<ExampleProtos.RowCountService,Long>(){
public Long call(ExampleProtos.RowCountService counter) throws IOException{
ServerRpcController controller = new ServerRpcController();
BlockingRpcCallback<ExamplerProtos.CountResponse>();
counter.getRowCount(controller,request,rpcCallback);
ExampleProtos.CountResponse response = rpcCallback.get();
if(controller.failedOnException()){
throw controller.getFailedOn();
}
return (response != null && response.hasCount())response.getCount():0;
}
});
}


}


几点说明：
1、协调处理器配置的加载顺序：先加载配置文件中定义的协处理器，后加载表描述符中的协处理器
2、COPROCESSOR$<number>中的number定义了加载的顺序
3、协处理器配置格式
Coprocess can also be configured to load on a per table basis，via a shell command “alter”
hbase > alter "t1".METHOD => 'table_alt'
'coprocessor1'=>'hdfs://foo.jar|com.foo.FooRegionObserver[100]arg1=1,arg2=2'


RegionObserverTest.java


import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
public class RegionObserverTest extends BaseRegionObserver{

}


>>describe 'students'


alter 'students',METHOD=>'table_att_unset',Name=>'coprocessor$1'


scan 'students'


---------------------------------------------------------------------------
MapReduce on HBase


MapReduce on Hbase
1、可以使用Mapreduce的方法操作hbase数据库
2、Hadoop Mapreduce 提供相关API，可以与hbase数据库无缝连接
3、API link: http://hbase.apache.org/devapidocs/index.html


--------------------HbaseMR.java---------------------------------------------------------------------------------
package com.hbase;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Mutation;
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.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;


public class HbaseMR {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
Configuration conf = HBaseConfiguration.create();
Job job = new Job(conf,"mapreduce on hbase");
job.setJarByClass(HbaseMR.class);
Scan scan = new Scan();
scan.setCaching(1000);


TableMapReduceUtil.initTableMapperJob("students", scan,MyMapper.class, Text.class, Text.class, job);
TableMapReduceUtil.initTableReducerJob("students_age",MyReducer.class, job);

job.waitForCompletion(true);

}
}




class MyMapper extends TableMapper<Text,Text>{
@Override
protected void map(
ImmutableBytesWritable key,
Result value,
Context context)
throws IOException, InterruptedException {
Text k = new Text(Bytes.toString(key.get()));
Text v = new Text(value.getValue(Bytes.toBytes("basicInfo"),Bytes.toBytes("age")));

context.write(v, k);
}
}


class MyReducer extends TableReducer<Text,Text,ImmutableBytesWritable>{


@Override
protected void reduce(
Text key,
Iterable<Text> values,
Context context)
throws IOException, InterruptedException {
Put put = new Put(Bytes.toBytes(key.toString()));

for(Text value: values){
put.add(Bytes.toBytes("f1"),Bytes.toBytes(value.toString()),Bytes.toBytes(value.toString()));
}
context.write(null, put);
}
}
------------------------------------------------------------------------------------------------------


数据存储格式
B+树
特点:
1、有n棵子数的结点中含有n个关键字，每个关键字不保存数据，只用来索引，所有数据都保存在叶子节点。
2、所有的叶子结点中包含了全部关键字的信息，及指向含这些关键字记录的指针，且叶子结点本身依关键字的大小自小而大顺序链接。
3、所有的非终端结点可以看成是索引部分，结点中仅含其子树（根结点）中的最大（或最小）关键字。


HFile格式
1、数据块-保存表中的数据，每一个数据块由块头和一些keyValue（record）组成，key的值是严格按照顺序存储的。块大小默认为64K（由建表时创建cf时指定或者HColumnDescriptor.setBlockSize(size)）,
这一部分可以压缩存储。在查询数据时，是以数据块为单位从硬盘load到内存。查找数据时，是顺序的遍历该块中的keyValue对。
2、元数据块（可选的）-保存用户自定义的kv对，可以被压缩。比如booleanfilter就是存在元数据块中的，该块只保留value值，key值保存在元数据索引块中。每一个元数据块由块头和value值组成。可以
快速判断key是都在这个HFile中。
3、File Info-Hfile的元信息，不被压缩，用户也可以在这一部分添加自己的元信息。
4、数据索引块-Data Block的索引，每条索引的key是被索引的block的第一条记录的key（格式为：头信息，数据块offset数据块大小块第一个记录的key,.....）


----------------------------------------------------------------------------------------------------
WAL机制
WAL预写日志
1）Client向RegionServer端提交数据的时候，会优先写WAL日志（HLog），只有当WAL日志写成功以后，Client才会被告诉提交数据成功，如果写WAL失败会告知客户端提交失败
2）一个RegionServer上所有的Region都共享一个HLog，一次数据的提交是先写WAL，再写memstore


HLog类
实现了WAL的类叫做HLog。当HRegion被实例化时，HLog实例会被当做一个餐所以和到HRegion的构造器中。当一个Region接收到一个更新操作时，它可以直接把数据保存到一个共享的WAL实例中去。




HLogKey类
1、当前的WAL使用的是Hadoop的SequenceFile格式，其key是HLogKey实例。HLogKey中记录了写入数据的归属信息，除了table和region名字外，同时还包括sequence number和timestamp，timestamp是“写入时间”，
sequence number的起始值为0，或者是最近一次存入文件系统中sequence number。
2、HLog Sequence File的Value是Hbase的KeyValue对象，即对应HFile中的KeyValue


WALEdit类
客户端发送的每个修改都会封装成WALEdit类，一个WALEdit类包含了多个更新操作，可以说一个WALEdit就是一个原子操作，包含若干个操作的集合。




LogSyncer类
Table在创建的时候，有一个参数可以设置，是否每次写Log日志都需要往集群里的其他机器同步一次，默认是每次都同步，同步的开销是比较大的，但不及时同步又可能因为机器宕而丢日志。
同步的操作现在是通过Pipeline的方式来实现的，Pipeline是指datanode接受数据后，再传给另外一台datanode，是一种串行的方式；n-Way Wirtess是指多datanode同时接受数据，最慢的一台结束就是整个结束。
差别在于一个延迟大，一个并发高，hdfs现在正在开发中，以便可以选择是按Pipeline还是n-Way Write来实现写操作。


Table如果设置每次不同步，则写操作会被RegionServer缓存，并启动一个LogSyncer线程来定时同步日志，定时时间默认是一秒也是由hbase.regionserver.optionallogflushinterval设置。


LogRoller类
1、日志写入的大小是有限制的。LogRoller类会作为一个后台线程运行，在特定的时间间隔内滚动日志。通过hbase.regionserver.logroll.period属性控制，默认1小时。
---------------------------------------------------------
Hbase在线数据备份


Hbase Replication
HBase复制是一种不同HBase部署中复制数据的方法。它可以作为一种故障恢复的方法，并提供HBase层次的高可用性。


HBase复制中最基本的架构模式是“主推送”(master-push),因为每个region server都有自己的WAL（或HLog），所以很容易保存现在正在复制的位置。正如众所周知的解决方案-MySql的主/从复制，只使用二进制
文件来跟踪修改。一个主集群可以将数据复制到任意数目的从集群，每个region server都会参与复制自己的修改。
来自每个region server的HLog是Hbase复制的基础，并且只要它们需要将数据复制到从集群，它们就必须被保存到HDFS上。每个region server从它需要的最老的日志开始复制，同时在zookeeper中保存当前恢复的位置
来简化错误恢复。每个从集群恢复的位置可能不同，但它们处理的HLog队列内容是相同的。参与复制的集群的规模可以不对等。主集群会通过随机分配尽量均衡从集群的负载。


解决问题：
数据管理人员的失误，不可逆的DDL操作。
底层HDFS文件BLOCK块corruption
短时间过度的读数据对集群造成的压力，增加服务器应对这种情况比较浪费资源
系统升级，维护，诊断问题会造成集群不可用时间增长。
双写的原子性难以保证
不可预计的一些原因。（如机房断电，大规模硬件损坏，断网等）
离线应用的MR计算对在线读写造成的较大的延迟影响


对于数据中心的数据冗余的备份方案，目前从一致性，事务性，延迟，吞吐量，数据损失，Failover几个角度来分析有一下几种方案。
简单备份模式通过定时不定时的Dump出集群数据保证数据的安全性，通常可以通过snapshot或设置时间戳来dump数据来实现这种方案，如果方案简介，设计优雅可以做到对在线数据中心低于干扰或无干扰的数据备份。但这
中方案缺点也是显而易见的，只是对时间点之前的数据安全性得到保障，如果发生突发事件会导致不可避免的整段时间的数据丢失，为很多人无法接受。
主从模式（Maser-Slave）这种模式比起简单的备份模式多了很多优点，可以通过最终一致性保证数据的一致，数据从主集群到备集群延时较低，异步写入不会对主集群带来性能压力，基本不会产生多少性能的影响，突发
事件来临时数据丢失很少，并且主集群的事务在备集群也可以得以保证。一般通过构造较好的Log系统加上check Point来实现，可以实现读写分离，主集群可以担当读写服务，但备集群一般只承担读服务。


主主模式（Master-Master）原理总体类似于主从模式，不同的是2个集群可以互相承担写的分离，都可承担读写服务。
2阶段提交这种方案保证了强一致性和事务，服务器返回给客户端成功则表明数据一定已经成功备份，不会造成任何数据丢失。每台服务器都可以承担读写服务。但缺点是造成集群延迟较高，总体吞吐下降。
Paxos算法基于Paxos算法的实现的强一致性方案，同一客户端连接的server能保证数据的一致性。缺点是实现复杂，集群延迟和吞吐随着集群服务器增加而边差。




部署步骤
1)编辑集群中所有机器的${HBASE_HOME}/conf/hbase-site.xml文件，增加如下配置：
<property>
<name>hbase.replication</name>
<value>true</value>
</property>
修改完成后，重启HBase集群，使配置生效




2）在HBase shell中运行如下命令：
>>add_peer 'ID''CLUSTER_KEY'
>>start_replication
第一条命令是为从集群设置zookeeper集群信息，这样可以使得修改被同步到从集群上。第二条命令真正将修改过的记录发布到从集群上。为了保证工作能按照预期进行，用户必须保证已经在从集群上建立了一个相同的表的副本，
表可以为空，但必须有相同的模式和表名。
注意：hbase-0.96和hbase-0.98已经没有start_replication命令和stop_replication命令。hbase-0.98相较hbase-0.96,新增了set_peer_tableCFs、show_peer_tablesCFs命令。在设定复制时，hbase-0.98需要使用set_peer_tableCFs设置。
具体的有对应的帮助命令可供参考。
ID必须是一个短整数，CLUSTER_KEY的内容请参考以下模板:
hbase.zookeeper.quorum:hbase.zookeeper.property.clientPort:zookeeper.znode.parent
比如，zk.server.com:2181:/hbase
注意：如果两个集群使用相同的zookeeper集群，你不得不使用不同的zookeeper.znode.parent,因为它们不能写入相同的文件夹中。




3）一旦你有一个对等（从）集群，你需要在你的列簇上使复制可用，要想达到这样的效果，可以在Hbase shell中执行如下命令：
disable ‘you——table’
alter ‘your-table’,{NAME =>'family_name',replication_scope=>'1'}
enable 'your_table'
scope值为0（默认值）意味着它不会被复制，而scope值为1意味着它将被复制.


4)运行如下命令可以列出所有配置的对等（从）集群：
》》list_peers


5)运行如下命令将使对等（从）集群不可用：
》》disable_peer 'ID'
运行完命令后，HBase将停止将向对等（从）集群发送修改，但是它将一直跟踪所有新的WALs文件，以便当从集群可用时继续复制。




6)可以运行如下命令将使之前设置为了不可用的对等（从）集群可用：
>>enable_peer'ID'
7)运行下面的命令，可以移除一个从集群：
stop_replication
remove_peer 'ID'
需要注意的是，停止复制仍会完成所有已在队列里的修改的复制，但是之后所有的处理都被停止了。为了确认你的配置都正常工作，你可以查看任何一个regionserver的日志文件，看是否由类似下面几行的内容：
Considering 1 rs,with ratio 0.1
Getting 1 rs from peer cluster#0
Choosing peer 10.10.1.49:62020


Hbase集群数据迁移
1）静态迁移方案
Hadoop distcp
在hbase停止的状态下进行数据的迁移
distcp + add_table.rb
------------------------------------------------------
2）动态迁移方案
-Replication备份方案
-CopyTable方案
-Export and Import方案


动态迁移方案
1、Replication备份方案
修改hbase-site.xml配置，增加hbase.replication属性
增加表属性REPLICATION_SCOPE属性
add_peer


2、CopyTable方案
命令：./hbase org.apache.hadoop.hbase.mapreduce.CopyTable--peer.adr=new cluster ip:2181:/hbase_table


说明：
（1）拷贝完成，不需要重启机器，在new cluster中就可以看到该表
（2）稳定性还需要考虑


3、Export and Import方案
步骤：（1）在old cluster上执行：./hbase
org.apache.hadoop.hbase.mapreduce.Export test hdfs://new clusterip:9000/xxx


(2)在new cluster上执行：./hbase
org.apache.hadoop.hbase.mapreduce.Import test hdfs://new clusterip:9000/xxx
说明:(1)一定要写全路径，不能写相对路径
(2)在Import前，需要将表事先在new cluster中创建好
-----------------------------------------------------------------------------------
3）手动方式
把hbase元数据表打包 tar zcvf ...
hadoop fs -put studnets /hbase/data/default


1、从源hbase集群中复制出HBase数据库表到本地目录
hadoop fs -get
2、目标HBase导入
hadoop fs -put
3、修复.META表
hbase hbck -fixMeta
4、重新分配数据到各RegionServer
hbase hbck -fixAssignments
---------------------------------------------------------------------------------
数据导入方案
1、利用ImportTsv将csv文件导入到HBase 备注：表要前创建
2、利用completebulkload将数据导入到HBase
3、利用Import将数据导入到HBase




数据导入方案一
利用ImportTsv将csv文件导入到HBase
命令：bin/hbase org.apache.hadoop.hbase.mapreduce.ImportTsv-Dimporttsv.separator="," -Dimporttsv.columns=HBASE_ROW_KEY,cf hbase-tbl-001 /simple.csv


simple.csv内容如下：
1、“tom”
2、“sam”
3、“jerry”
4、“marry”
5、“john”
-----------------------------------------------------------------------------
数据导入方案二
HBase支持bulkload的入库方式，它是利用hbase的数据信息按照特定格式存储在hdfs内这一原理，直接在HDFS中生成持久化的HFile数据格式文件，然后上传至合适位置，即完成巨量数据快速入库的办法。配合
mapreduce完成，高效便捷，而且不占用region资源，增添负载，在大数据量写入时能极大的提高写入效率，并降低对HBase节点的写入压力。


通过使用先生成HFile，然后再BulkLoad到HBase的方式来代替之前直接调用HTableOutputFormat的方法有如下好处：
（1）消除了对HBase集群的插入压力
（2）提高了Job的运行速度，降低了Job的执行时间


利用completebulkload将数据导入到HBase
1）先通过ImportTsv生成HFile
命令：bin/hbase org.apache.hadoop.hbase.mapreduce.ImportTsv -Dimporttsv.separator="," -Dimporttsv.bulk.output=/hfile_tmp -Dimporttsv.columns=HBASE_ROW_KEY,cf hbase-tbl-001 /simple.csv
2)通过completebulkload将数据导入表hbase-tbl-002
命令：hadoop jar lib/hbase-server-0.96.0.jar completebulkload /hfile_tmp hbase-tbl-002


利用Import将数据导入到HBase
1）HBase export工具导出的数据格式是sequence file。比如，在执行完命令“bin/hbase org.apache.hadoop.hbase.mapreduce.Export hbase-tbl-002/test-output”后，hbase会启动一个MapReduce作业，作业
完成后会在hdfs上面会生成sequence file格式的数据文件
2）对于这类Sequence file格式的数据文件，HBase是可以通过Import工具直接将它导入到HBase的表里面的。执行命令“bin/hbase org.apache.hadoop.hbase.mapreduce.Import hbase-tbl-003 /test-output”,
随后hbase会启动一个MapReduce作业，然后表test会成功导入数据。
---------------------------------------------------------------------------
Hbase二级索引


MapReduce方案
IndexBuilder：利用MR的方式构建Index


优点：并发批量构建Index
缺点：不能实时构建Index


举例：
原表：row 1 f1:name zhangsan
row 2 f1:name lisi
row 3 f1:name wangwu


索引表：row zhangsan f1:id 1
row lisi f2:id 2
row wangwu f3:id 3




create 'student','f1'
put 'student','1','f1:name','lisi'
put 'student','2','f1:name','zhangsan'
put 'student','3','f1:name','wangwu'


create 'student-name','f1'


程序放到/usr/lib/hbase/lib/目录下
>>hbase com.hbase.IndexBuilder student f1 name
给表student 添加二级索引 studnet-name
示例如下：
-----------------------------------------IndexBuilder.java----------------------------------------------------------------------------
package com.hbase;
import java.io.IOException;
import java.util.HashMap;
import java.util.Set;
import javax.ws.rs.PUT;
import org.apache.hadoop.conf.Configuration;
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.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.MultiTableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.GenericOptionsParser;


public class IndexBuilder {
static class MyMapper extends TableMapper<ImmutableBytesWritable,Put>{
private HashMap<byte[], ImmutableBytesWritable> indexes = new HashMap<byte[],ImmutableBytesWritable>();
private String familyName;

@Override
protected void map(
ImmutableBytesWritable key,
Result value,
Context context)
throws IOException, InterruptedException {
Set<byte[]> keys = indexes.keySet();
for(byte[] k:keys){
ImmutableBytesWritable indexTableName = indexes.get(k);
byte[] val = value.getValue(Bytes.toBytes(familyName),k);

if(val !=null){
Put put = new Put(val);
put.add(Bytes.toBytes("f1"),Bytes.toBytes("id"),key.get());
context.write(indexTableName, put);
}
}
}


@Override
protected void setup(Context context)
throws IOException, InterruptedException {
Configuration conf = context.getConfiguration();
String tableName = conf.get("tableName");
String familyName = conf.get("familyName");

String[] qualifiers = conf.getStrings("qualifiers");
for(String q: qualifiers){
indexes.put(Bytes.toBytes(q),new ImmutableBytesWritable(Bytes.toBytes(tableName + "-" + q)));
}
}

}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
Configuration conf = HBaseConfiguration.create();
String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();

if(otherArgs.length < 3){
System.exit(-1);
}
String tableName = args[0];
String columnFamily = args[1];
conf.set("tableName", tableName);
conf.set("columnFamily", columnFamily);

String[] qualifiers = new String[otherArgs.length -2];
for(int i =0; i < qualifiers.length;i++){
qualifiers[i] = otherArgs[i+2];
}
conf.setStrings("qualifiers", qualifiers);

Job job = new Job(conf,tableName);

job.setJarByClass(IndexBuilder.class);
job.setMapperClass(MyMapper.class);
job.setNumReduceTasks(0);

job.setInputFormatClass(TableInputFormat.class);
job.setOutputFormatClass(MultiTableOutputFormat.class);

Scan scan = new Scan();
scan.setCaching(1000);

TableMapReduceUtil.initTableMapperJob(tableName, scan, MyMapper.class, ImmutableBytesWritable.class, PUT.class, job);
job.waitForCompletion(true);
}
}
------------------------------Hbase二级索引------------------------------------------------------------------------------------------------
1、Mapreduce方案
2、ITHBASE方案
3、IHBASE方案
4、Coprocessor方案
5、Solr + hbase方案


IHBASE方案
优点：IHBase(Indexed HBase)是HBASE的一个扩展，用于支持更快的扫描。
缺点：需要重构hbase
原理：在Memstore满了以后刷磁盘时，IHBase会进行拦截请求并为这个memstore的数据构建索引，索引另一个CF的方式存储在表内。scan的时候，IHBase会结合索引列中的标记，来加速scan


ITHBASE方案
优点：ITHBASE(Indexed Transactional HBase)是HBase的一个事务型的带索引的扩展
缺点：需要重构hbase，几年没有更新


Coprocessor方案
HIndex-来自华为的HBase二级索引


Solr方案
Solr是一个独立的企业级搜索应用服务器，它对外提供类似于Web-service的API接口。用户可以通过http请求，向搜索引擎服务器提交一定格式的XML文件，生成索引；也可以通过Http Get操作
提出查找请求，并得到XML格式的返回结果。


Solr是一个高性能，采用Java5开发，基于Lucene的全文搜索服务器。同时对其进行扩展，提供了比Lucene更为丰富的查询语言，同时实现了可配置、可扩展并对查询性能进行了优化，并且提供了一个
完善的功能管理界面，是一款非常优秀的全文搜索引擎。


HBase无可置疑拥有其优势，但其本身只对rowkey支持毫秒级的快速检索，对于多字段的组合查询却无能为力


基于Solr的HBase多条件查询原理很简单，将HBase表中涉及条件过滤的字段和rowkey在Solr中建立索引，通过Solr的多条件查询快速获得符合过滤条件的rowkey值，拿到这些rowkey之后在HBase中通过
指定rowkey进行查询。


--------------------------------------------------------------------------------------------------------
Hbase快照技术
Snapshot（快照）
什么是快照
快照就是一份元信息的集合，允许管理员恢复到表的先前状态。快照不是表的复制而是一个文件名称列表，因而不会复制数据。


完全快照恢复是指恢复到之前的“表结构”以及当时的数据，快照之后发生的数据不会恢复。


快照的作用
hbase中存在的备份或克隆表的方法就是使用复制、导出表或者在关闭表之后拷贝HDFS中所有的hfile。复制/导出是通过一系列工具调用MapReduce来扫描并复制表，这样会对RegionServer有直接的影响。
关闭表会停止所有的读写操作，实际环境中往往无法接受。


相比之下Hbase快照允许管理员不拷贝数据直接克隆一张表，这对域服务器产生的影响最小。将快照导出至其他集群不会直接影响到任何域服务器；导出只是带有一些额外逻辑的群间数据同步。


快照使用场景
从用户、应用异常中还原
从一个已知的安全状态恢复、还原
查看之前的快照并有选择性的合并不同写入产品环境
当主应用程序升级或改版时保存快照
在指定时间审查和、或报告数据
按照规定捕获月度数据
生成日终、月末、季末报告
应用测试
通过快照模拟生产环境下结构或应用发生的变化，测试完成即可丢弃
例如：生成快照，利用快照中内容构建新表（原有结构+数据）并且修改新表的结构，添加或删除列之类。（原始表、快照和新表保持互相独立）
减少工作压力
生成快照，导入到其他集群，然后运行MapReduce jobs。因为导出的快照是HDFS级别，所以不会像复制表那样降低HBase主集群的效率。


快照操作
生成快照：本操作尝试对指定表生成快照。如果集群在执行数据均衡、分隔或合并等操作是，可能会引起操作失败。
克隆快照：本操作使用与指定快照相同的结构数据构建一张新表。操作结果会生成一张有完整功能的表，对该表的任意修改不会对原表或快照产生影响。
还原快照：本操作将表结构和数据恢复到生成快照时的状态。（注意：本操作会舍弃快照生成后任何改变）。
删除快照：本操作将系统中的快照删除，释放未共享的磁盘空间，而且不会影响其他克隆或快照。
导出快照：本操作将快照数据和元数据复制到其他集群。操作只会涉及HDFS，不会与Master或域服务器产生任何联系，因此HBase集群可以关闭。




快照优势
导出快照与复制、导出表除了更好的保持一致性外，主要的不同在于导出快照是在HDFS的层面操作的。
这意味着Master和域服务器与操作无关。因此不需要为不必要的数据创建缓存空间，也不会有扫描过程因为大量对象创建引起的GC暂停。对于HBase来说主要性能影响就是DataNode额外的网络和磁盘负载。


snapshot 'students','students_snapshot'


list_snapshots
clone_snapshot 'students_snapshot','students_new'


数据导入方案三
通过检查hbase-site.xml的hbase.snapshot.enable是否设置为true确认打开了快照许可。
1、获取指定表的快照使用snapshot命令(不产生文件复制)
snapshot 'tableName','snapshotName'


2、列出所有的快照，使用list_snapshot命令。会展示出快照名称，源表，以及创建日期和时间
list_snapshots


3、删除快照使用deleted_snapshot命令。删除快照不会影响到克隆表或者之后生成的快照。
delete_snapshot 'snapshotName'


4、使用clone_snapshot命令从指定的快照生成新表（克隆）。由于不会产生数据复制，所以最终用到的数据不会是之前的两倍。
clone_snapshot 'snapshotName','newTableName'


5、使用restore_snapshct命令指定快照内容替换当前表结构/数据。
disable 'students'
restore_snapshot 'snapshotName'
enable 'students'
6、使用ExportSnapshot工具将现有快照导出至其他集群。导出工具不会影响到域服务器负载，只是工作在HDFS层面所以需要指定HDFS路径（其他集群的hbase根目录）
hbase org.apache.hadoop.hbase.snapshot ExportSnapshot -snapshot
SnapshotName -copy -to hdfs://srv2:8020/hbase




delete 'students','Tom','moreInfo:tel'
-----------------------------------------------------------------------
Hbase BloomFilter(布隆过滤器)
集合表示和元素查询
下面我们具体来看BloomFilter是如何用位数组表示集合的。初始状态时，BloomFilter是一个包含m位的位数组，每一位都置为0
为了表达S=（x1，x2，x3，。。。xn）这样一个n个元素的集合，BloomFilter使用k个相互独立的哈希函数（HashFunction），它们分别将集合中的每个元素映射到（1，。。。m）的范围中。对任意一个元素x，
第i个哈希函数映射的位置h1（x）就会被置为1（1<=i<=k）.注意，如果一个位置多次被置为1，那么只有第一次会起作用，后面几次将没有任何效果。在下图中，k=3，且有两个和希函数选中同一个位置（从左边数第五位）


在判断y是否属于这个集合时，我们对y应用k次哈希函数，如果所有hi（y）的位置都是1（1<=i<=k）,那么我们就认为y是集合中的元素，否则就认为y不是集合中的元素，下图中y1就不是集合中的元素。y2或者属于这个集合，
或者刚好是一个false positive。


Bloomfilter在HBase中的作用？
HBase利用Bloomfilter来提高随机读（Get）的性能，对于顺序读（Scan）而言，设置Bloomfilter是没有作用的（0.92以后，如果设置了bloomfilter为ROWCOL，对于指定了qualifier的Scan有一定的优化）


Bloomfilter在Hbase中的开销？
Bloomfilter是一个列族cf级别的配置属性，如果在表中设置了Bloomfilter，那么HBase会在生成StoreFile时包含一份bloomfilter结构的数据，称其为MeataBlock；MetaBlock与DataBlock（真实的KeyValue数据）
一起由LRUBlockCache维护。所以开启bloomfilter会有一定的存储及内存cache开销。


HBase中的Bloomfilter的类型及使用？
a)ROW,根据KeyValue中的row来过滤storefile
举例：假设有2个storefile文件sf1和sf2
sf1包含kv1（r1 cf：q1 v）、kv2（r2 cf：q1 v）
sf2包含kv3（r3 cf：q1 v）、kv4（r4 cf：q1 v）
如果设置了CF属性中的bloomfilter为ROW，那么get（r1）时就会过滤sf2，get（r3）就会过滤sf1


b)ROWCOL，根据KeyValue中的row+qualifier来过滤storefile
举例：假设有2个storefile文件sf1和sf2，
sf1包含kv1（r1 cf：q1 v）、kv2（r2 cf：q1 v）
sf2包含kv3（r1 cf：q2 v）、kv4（r2 cf：q2 v）
如果设置了CF属性中的bloomfilter为ROW，无论get（r1，q1）还是get（r1，q2），都会读取sf1+sf2；而如果设置了CF属性中的bloomfilter为ROWCOL，那么get（r1，q1）就会过滤sf2，get（r1，q2）就会过滤sf1


ROWCOL和ROW对比
a)ROWCOL只对指定列（Qualifier）的随机读（Get）有效，如果应用中的随机读get，只含row，而没有指定读哪个qualifier，那么设置ROWCOL是没有效果的，这种场景就应该使用ROW
b)如果随机读中指定的列（Qualifier）的数目大于等于2，在0.90版本中ROWCOL是无效的，而在0.92版本以后，HBase-2794对这一情景作了优化，是有效的（通过KeyValueScanner#seekExactly）
c)如果同一row多个列的数据在应用上是同一时间put的，那么ROW与ROWCOL的效果近似相同，而ROWCOL只对指定了列的随机读才会有效，所以设置为ROW更佳
d)ROWCOL与ROW只在名称上有联系，ROWCOL并不是ROW的扩展，不能取代ROW
--------------------------------------------------------------------------------------------
基于Solr的实时查询
实时查询方案
hbase+solr+hbase indexer
1）hbase提供海量数据存储
2）solr提供索引构建与查询
3）hbase indexer提供自动化索引构建（从hbase到solr）
----------------------------------------------------------------------------
部署与安装
CDH5.x发行版中提供了hbase、solr以及hbase indexer的下载
实验选取CDH5.0.2版本，选择需下载以下三个：
hbase-0.96.1.1-cdh5.0.2.tar.gz
hbase-solr-1.3-cdh5.0.2.tar.gz
solr-4.4.0-cdh5.0.2.tar.gz


其中hbase-solr-xxx，来源NGDATA的开源项目，使用及说明参考


hbase->hbase inexder -> solr
---------------------------------------------------------------------------------
Solr查询
利用Solr操作solr API，使用SolrJ操作Solr会比利用httpClient来操作Solr要简单
查询使用的类：
1、HttpSolrServer
2、SolrQuery


192.168.1.198:8983/solr/selectq=1astname_s:Louis


Main.java
public class Main{
public static void main(String[] args){
String url = "http://192.168.1.198:8983/solr";
HttpSolrServer serve = new HttpSolrServer(url);
server.setConnectionTimeout(1000);
server.setSoTimeout(3000);
server.setDefaultMaxConnectionPerHost(1000);
server.setFollowRedirects(false);
server.setAllowCompression(true);
server.setMaxRetries(1);


SolrQuery query = new SolrQuery();
query.setQuery("firstname s:Jim");
try{
QueryResponse resp = server.query(query);
List<Item> item = resp.getBeans(Item.class);


for(Item item: items){
System.out.println(item.id);
conn.format(conn.getData("indexdemo-user",item.id));
}
}catch(SolrServerException e){
e.printStackTrace();
}





}


}


Item.java
public class Item{
@Field
String id;


@Field
String firstname_s;


@Field
String lastname_s;
}


---------------------------------------------------------------------------------------------------
HBaseConnection.java
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.util.Bytes;


public class HbaseConnection{
private String rootDir;
private String zkServer;
private String port;


private Configuration conf;
private HConnection hConn = null;
public HbaseConnection(String rootDir,String zkServer,String port){
this.rootDir = rootDir;
this.zkServer = zkServer;
this.port = port;
conf = HBaseConfiguration.create();
conf.set("hbase.rootdir",rootDir);
conf.set("hbase.zookeeper.quorum",zkServer);
conf.set("hbase.zookeeper.property.clientPort",port);
try{
hConn = HConnectionManager.createConnection(conf);
}catch(IOException e){
e.printStackTrace();
}
}
public void createTable(String tableName,List<String> cols){
try{
HBaseAdmin admin = new HBaseAdmin(conf);
if(admin.tableExists(tableName))
throw new IOException("table exists");
else{
HTableDescriptor tableDesc = new HTableDescriptor(tableName);
for(String col : cols){
HColumnDescriptor colDesc = new HColumnDescriptor(col);
colDesc.setCompressionType(Algorithm.GZ);
colDesc.setDataBlockEncoding(DataBlockEncoding.DIFF);
tableDesc.addFamily(colDesc);
}
admin.createTable(tableDesc);
}


}catch(MasterNotRunningException e){
e.printStackTrace();
}catch(ZooKeeperConnectionException e){
e.printStackTrace();
}catch(IOException e){
e.printStackTrace();
}

}


public void saveData(String tableName,List<Put> puts){
try{
HTableInterface table = hConn.getTable(tableName);
table.put(puts);
table.setAutoFlush(false);
table.flushCommits();
}catch(IOException e){
e.printStackTrace();
}
}


public Result getData(String tableName,String rowkey){
try{
HTableInterface table = hConn.getTable(tableName);
Get get = new Get(Bytes.toBytes(rowkey));
return table.get(get);
}catch(IOException e){
e.printStackTrace();
}

return null;
}

public void format(Result result){
String rowkey = Bytes.toString(result.getRow());
KeyValue[] kvs = result.raw();
for(KeyValue kv: kvs){
String family = Bytes.toString(kv.getFamily());
String qualifier = Bytes.toString(kv.getQualifier());
System.out.println("rowkey->" + rowkey + "family->" + family + "qualifier->" + qualifier);
}
}

public void hbaseScan(String tableName){
Scan scan = new Scan();
scan.setCaching(1000);
try{
HTableInterface table = hConn.getTable(tableName);
ResultScanner scanner = table.getScanner(scan);
for(Result res: scanner){
format(res)
}
}catch(IOException e){
e.printStackTrace();
}
}


public void filterTest(String tableName){
Scan scan = new Scan():
scan.setCaching(1000);
RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,
///new BinaryComparator(Bytes.tobytes("Jack")));
new RegexStringComparator("J\\w+")));


PageFilter filter = new PageFilter(15);//分页显示
scan.setFilter(filter);
try{
HtableIntegerface table = hConn.getTable(tableName);
ResultScanner scanner = table.getScanner(scan);


for(Result res: scanner){
format(res);
}
}catch(IOException e){
e.printStackTrace();
}
}




public void pageFilterTest(String tableName){
PageFilter filter = new PageFilter(4);
byte[] lastRow = null;
int pageCount = 0;
try{
HtableInterface table = hConn.getTable(tableName);
while(++pageCount > 0){
System.out.println("pageCount = " + pageCount);
Scan scan = new Scan();
scan.setFilter(filter);
if(lastRow != null){
scan.setStartRow(lastRow);
}
ResultScanner scanner = table.getScanner(scan);
for(Result res: scanner){
format(res);
lastRow = res.getRow();
if(++count > 3)
break;
format(res);
}
if(count < 3)
break;

}
}catch(IOException e){
e.printStackTrace();
}



}


public static void main(String[] args){
String rootDir = "hdfs://hbase";
String zkServer = "192.168.1.198";
String port = "2181";
HbaseConnection conn = new HbaseConnection(rootDir,zkServer,port);
/* List<String> cols = new LinkedList<>();
cols.add("basicInfo");
cols.add("moreInfo");
conn.createTable("students",cols);*/


List<Put> puts = new LinkedList<Put>();
Put put1 = new Put(Bytes.toBytes("Tom"));
put1.add(Bytes.toBytes("basicInfo"),Bytes.toBytes("age"),Bytes.toBytes("27"));
put1.add(Bytes.toBytes("moreInfo"),Bytes.toBytes("tel"),Bytes.toBytes("11232"));

Put put2 = new Put(Bytes.toBytes("Jim"));
put2.add(Bytes.toBytes("basicInfo"),Bytes.toBytes("age"),Bytes.toBytes("28"));
put2.add(Bytes.toBytes("moreInfo"),Bytes.toBytes("tel"),Bytes.toBytes("11233"));
puts.add(put1);
puts.add(put2);
conn.saveData("students",puts);
Result result = conn.getData("students","Tom");
conn.format(result);


conn.hbaseScan("studnets");
conn.filterTest("students");
}
}


备注：因为solr和zookpeer有问题，实例没有通过，有时间在调整。
--------------------------------------------------------------------------------------------
【大 中 小】【打印】 【繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部】
上一篇：VMWare虚拟机下为Ubuntu 12.04.1.. 下一篇：CSDN-markdown编辑器

最新文章

2018-3-7       ..
azkaban-web-start.sh启动时出现Tab..
Hadoop提交作业------>hadoop工..
结合案例讲解MapReduce重要知识点 ..
Hadoop pipes初学笔记

热门文章

Hadoop 中web服务的REST API介绍
ThingsBoard源码分析 —— 调试环境..
使用Eclipse插件连接配置Mapreduce..
解决Exception in thread "mai..
hadoop和spark比较

Hot 文章

Hadoop 中web服务的REST API介绍
ThingsBoard源码分析 —— 调试环境..
使用Eclipse插件连接配置Mapreduce..
解决Exception in thread "mai..
hadoop和spark比较

Python

C 语言

C++基础

大数据基础

linux编程基础

C/C++面试题目