总结：HBase实践篇

一般我们都是通过DFS操作Hbase，基本不会通过shell的方式，以下主要是了解下底层操作原理吧。

1、进入Hbase环境：./bin/hbase shell

2、创建表与插入数据：

# create语法
create '表名', {NAME => '列族名1'}, {NAME => '列族名2'}, {NAME => '列族名3'}
# 此种方式是上上面的简写方式，使用上面方式可以为列族指定更多的属性，如VERSIONS、TTL、BLOCKCACHE、CONFIGURATION等属性
create '表名', '列族名1', '列族名2', '列族名3'

hbase(main):003:0> create 'test', 'cf'
0 row(s) in 1.2200 seconds
hbase(main):003:0> list 'table'
test
1 row(s) in 0.0550 seconds
hbase(main):004:0> put 'test', 'row1', 'cf:a', 'value1'
0 row(s) in 0.0560 seconds
hbase(main):005:0> put 'test', 'row2', 'cf:b', 'value2'
0 row(s) in 0.0370 seconds
hbase(main):006:0> put 'test', 'row3', 'cf:c', 'value3'
0 row(s) in 0.0450 seconds

3、put：添加或修改的表的值。

类似mysql的insert语句，案例如下：

# 语法
# 当列族中只有一个列时'列族名:列名'使用'列族名'
put '表名', '行键', '列族名', '列值'
put '表名', '行键', '列族名:列名', '列值'

# 示例

# 创建表
create 'tbl_user', 'info', 'detail', 'address'

# 第一行数据
put 'tbl_user', 'mengday', 'info:id', '1'
put 'tbl_user', 'mengday', 'info:name', '张三'
put 'tbl_user', 'mengday', 'info:age', '28'

put 'tbl_user', 'mengday', 'detail:birthday', '1990-06-26'
put 'tbl_user', 'mengday', 'detail:email', 'abc@163.com'
put 'tbl_user', 'mengday', 'detail:create_time', '2019-03-04 14:26:10'

put 'tbl_user', 'mengday', 'address', '上海市'

# 第二行数据
put 'tbl_user', 'vbirdbest', 'info:id', '2'
put 'tbl_user', 'vbirdbest', 'info:name', '李四'
put 'tbl_user', 'vbirdbest', 'info:age', '27'

put 'tbl_user', 'vbirdbest', 'detail:birthday', '1990-06-27'
put 'tbl_user', 'vbirdbest', 'detail:email', 'xxx@gmail.com'
put 'tbl_user', 'vbirdbest', 'detail:create_time', '2019-03-05 14:26:10'

put 'tbl_user', 'vbirdbest', 'address', '北京市'


# 第一行数据
put 'tbl_user', 'xiaoming', 'info:id', '3'
put 'tbl_user', 'xiaoming', 'info:name', '王五'
put 'tbl_user', 'xiaoming', 'info:age', '26'

put 'tbl_user', 'xiaoming', 'detail:birthday', '1990-06-28'
put 'tbl_user', 'xiaoming', 'detail:email', 'xyz@qq.com'
put 'tbl_user', 'xiaoming', 'detail:create_time', '2019-03-06 14:26:10'

put 'tbl_user', 'xiaoming', 'address', '杭州市'

4、scan：扫描表信息

通过对表的扫描来获取对用的值

scan ‘表名’
扫描某个列族： scan ‘表名’, {COLUMN=>‘列族名’}
扫描某个列族的某个列： scan ‘表名’, {COLUMN=>‘列族名:列名’}
查询同一个列族的多个列： scan ‘表名’, {COLUMNS => [ ‘列族名1:列名1’, ‘列族名1:列名2’, …]}

# 扫描命名空间hbase下的meta表，显示出meta表的所有数据
  hbase> scan 'hbase:meta'

# 扫描命名空间hbase下的meta表的列族info的列regioninfo，显示出meta表的列族info下的regioninfo列的所有数据
  hbase> scan 'hbase:meta', {COLUMNS => 'info:regioninfo'}

# 扫描命名空间ns1下表t1的列族'c1'和'c2'。显示出命名空间ns1下表t1的列族'c1'和'c2'的所有数据
     hbase> scan 'ns1:t1', {COLUMNS => ['c1', 'c2']}

# 扫描命名空间ns1下表t1的列族'c1'和'c2'。显示出命名空间ns1下表t1的列族'c1'和'c2'，且只显示前10个rowkey的数据。
  hbase> scan 'ns1:t1', {COLUMNS => ['c1', 'c2'], LIMIT => 10}

# 扫描命名空间ns1下表t1的列族'c1'和'c2'。显示出命名空间ns1下表t1的列族'c1'和'c2'，且只显示从rowkey=“xyz”开始的前10个rowkey的数据。
  hbase> scan 'ns1:t1', {COLUMNS => ['c1', 'c2'], LIMIT => 10, STARTROW => 'xyz'} 

# 扫描默认命名空间下表t1的列族c1时间戳从'1303668804'到'1303668904'的数据
  hbase> scan 't1', {COLUMNS => 'c1', TIMERANGE => [1303668804, 1303668904]}

# 反向显示表t1的数据
  hbase> scan 't1', {REVERSED => true}

# 过滤显示表t1的数据
  hbase> scan 't1', {FILTER => "(PrefixFilter ('row2') AND
    (QualifierFilter (>=, 'binary:xyz'))) AND (TimestampsFilter ( 123, 456))"}

# RAW为true，显示出表t1的所有数据，包括已经删除的
  hbase> scan 't1', {RAW => true, VERSIONS => 10}

# 表t1的引用的扫描
  hbase> t11 = get_table 't1'
  hbase> t11.scan

5、get

#得到命名空间ns1下表t1的rowkey为r1的数据
  hbase> get 'ns1:t1', 'r1'

#得到默认命名空间下表t1的rowkey为r1的数据
  hbase> get 't1', 'r1'

#得到默认命名空间下表t1的rowkey为r1，时间戳范围在ts1和ts2之间的数据
  hbase> get 't1', 'r1', {TIMERANGE => [ts1, ts2]}

#得到默认命名空间下表t1的rowkey为r1的c1列的数据
  hbase> get 't1', 'r1', {COLUMN => 'c1'}

#得到默认命名空间下表t1的rowkey为r1的c1,c2,c3列的数据
  hbase> get 't1', 'r1', {COLUMN => ['c1', 'c2', 'c3']}

#得到默认命名空间下表t1的rowkey为r1的c1列，时间戳为ts1的数据
  hbase> get 't1', 'r1', {COLUMN => 'c1', TIMESTAMP => ts1}

#得到默认命名空间下表t1的rowkey为r1的c1列，时间戳范围为ts1到ts2，版本数为4的数据
  hbase> get 't1', 'r1', {COLUMN => 'c1', TIMERANGE => [ts1, ts2], VERSIONS => 4}

#应用对象的用法
  hbase> t.get 'r1'
  hbase> t.get 'r1', {TIMERANGE => [ts1, ts2]}
  hbase> t.get 'r1', {COLUMN => 'c1'}
  hbase> t.get 'r1', {COLUMN => ['c1', 'c2', 'c3']}
  hbase> t.get 'r1', {COLUMN => 'c1', TIMESTAMP => ts1}
  hbase> t.get 'r1', {COLUMN => 'c1', TIMERANGE => [ts1, ts2], VERSIONS => 4}
  hbase> t.get 'r1', {COLUMN => 'c1', TIMESTAMP => ts1, VERSIONS => 4}

6、count 'hubble_uid'

@Before
public void init() throws Exception {
		config = HbaseConfiguration.create();// 配置
		config.set("hbase.zookeeper.quorum", "192.168.33.61");// zookeeper地址
		config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口
		connection = ConnectionFactory.createConnection(config);
		table = connection.getTable(TableName.valueOf("dept"));
}

也可以将zookeeper地址放到hdfs-site.xml中，如：

  hbase.zookeeper.property.clientPort
  2181
  Property from ZooKeeper's config zoo.cfg. The port at which the clients will connect



  hbase.zookeeper.quorum
  hbase-zookeeper01-zjy.xx.hadoop,hbase-zookeeper02-zjy.xx.hadoop
  Comma separated full list of servers in the ZooKeeper Quorum 

并不是谁都可以访问Hbase服务进行操作的，所以要有安全认证规范。

可采用Kerberos 身份认证（比如我们使用的keytab就是Kerberos的身份认证方式）。

Kerberos 是一种基于对称密钥技术的身份认证协议，它作为一个独立的第三方的身份认证服务，可以为其它服务提供身份认证功能，且支持 SSO。

Kerberos认证支持两种方式：密码认证及keytab认证。认证有效时间默认为24小时。

密码认证：通过输入用户正确的密码完成身份认证。主要在运维管理场景中使用“人机”用户进行认证，命令为kinit 用户名。
keytab认证：keytab文件包含了用户的安全信息。使用keytab文件认证时，系统自动使用加密的凭据信息进行认证无需输入用户密码。主要在组件应用开发场景中使用，且使用“机机”用户。keytab文件也支持在kinit命令中使用。
 

如何指定使用什么身份认证？

在hdfs-site.xml中配置：

  hbase.security.authentication
  kerberos

大致有四步操作

3.1、第一步：构建数据列表

List puts = new ArrayList<>();

3.2、第二步：构建PUT元组，返回类型为Put

	private static Put build(Map nodeMap) {
		Map itemMap = new TreeMap() {
			private static final long serialVersionUID = -2149400882864567718L;
		};
		// Map itemset= new TreeMap(){};
		String value = null;
		// 后期这里会变成1分钟1次的心跳和mysql获取数据的
		String ignoreWord = ",timestamp,value,counterType,valueStr,"; // ,group,idc
		Iterator nodeIterator = nodeMap.keySet().iterator();
		while (nodeIterator.hasNext()) {
			String key = nodeIterator.next();
			if (ignoreWord.indexOf("," + key + ",") == -1) {
				if (key.contains(".")) {
					Object val = nodeMap.get(key);
					key = key.replace(".", "|_|_|");
					itemMap.put(key, val);
				} else {
					itemMap.put(key, nodeMap.get(key));
				}

			}
		}

		String mD5Str = DigestUtils.md5Hex(itemMap.toString());

		String timestamp = String.valueOf(nodeMap.get("timestamp"));

		if (nodeMap.containsKey("valueStr")) {
			value = String.valueOf(nodeMap.get("valueStr"));
			logger.debug("valueStr :{}", value);
			if (value.length() > 100) {
				value = value.substring(0, 100);
			}
		} else {
			value = String.valueOf(nodeMap.get("value"));
		}

		String timehour = String.valueOf(Long.parseLong(timestamp) / 3600 * 3600);
		String timesecond = String.valueOf(Long.parseLong(timestamp) % 3600);

		StringBuffer rowkeyBuffer = new StringBuffer();
		rowkeyBuffer.append(mD5Str);
		rowkeyBuffer.append("|");
		rowkeyBuffer.append(timehour);

		logger.debug("MD5 Str is {}", itemMap.toString());
		logger.debug("build rowkeyBuffer:{},timesecond:{},value:{}", rowkeyBuffer.toString(), timesecond, value);
		byte[] rowKeyArray = Bytes.toBytes(rowkeyBuffer.toString());
		Put put = null;

		put = new Put(rowKeyArray);

		put.addColumn(Bytes.toBytes(CF), Bytes.toBytes(timesecond), Bytes.toBytes(value));
		return put;
	}

3.3、第三步：填充puts数据

第一步生成的Put元组add到list中

3.4、第四步：数据批量写入

    public static void putByHTableSuccessful(String tablename, List puts) throws Exception {
        long currentTime = System.currentTimeMillis();
        Connection conn = getConnection();
        HTable htable = (HTable) conn.getTable(TableName.valueOf(tablename));
        htable.setAutoFlushTo(false);//是否自动刷新提交
        htable.setWriteBufferSize(5 * 1024 * 1024);
        Integer successNum = 0;
        Integer allNum = 0;
        try {
            htable.put((List) puts);
            htable.flushCommits();
            successNum = puts.size();
        } finally {
            htable.close();
            allNum = puts.size();
        }
    }

通过调用HTable.setAutoFlushTo(false)方法可以将HTable写客户端自动flush关闭，这样可以批量写入数据到Hbase，而不是有一条put就执行一次更新，只有当put填满客户端写缓存的时候，才会向Hbase服务端发起写请求。默认情况下auto flush是开启的。
 

//第一步：由于是scan查询，需要先组织好开始rowkey和结束的rowkey
String prefixRow = md5Id.toString();
String startRow = prefixRow + "|"  +  starthourString;
String stopRow = prefixRow + "|" + endhourString;
String tablename = "XXX";

//第二步：
public static TreeMap scanPrefixgetMap(String tablename, String prefixRow, String startRow,
			String StopRow, String filterOperation, String filterValue) throws IOException {
        //定义扫描的表
		Table table = null;
        //扫描得到的数据
		ResultScanner resultScanners = null;
        //封装数据给client
		TreeMap timestampValue = new TreeMap();
		try {
            //获取表
			table = conn.getTable(TableName.valueOf(tablename));

            //扫描条件封装
			Scan scan = new Scan();
			byte[] startRowBytes = Bytes.toBytes(startRow);
			byte[] stopRowBytes = Bytes.toBytes(StopRow);
			byte[] prefixRowBytes = Bytes.toBytes(prefixRow);
			if (filterOperation != null) {
				Filter filter = null;
				byte[] filterValueByte = Bytes.toBytes(filterValue);
				switch(filterOperation){
				case "!=":
					filter = new ValueFilter(CompareOp.NOT_EQUAL,
							new BinaryComparator(filterValueByte));
					scan.setFilter(filter);
					break;
				case "=":
					filter = new ValueFilter(CompareOp.EQUAL,
							new BinaryComparator(filterValueByte));
					scan.setFilter(filter);
					break;
				case "<":
					filter = new ValueFilter(CompareOp.LESS,
							new BinaryComparator(filterValueByte));
					scan.setFilter(filter);
					break;
				case ">":
					filter = new ValueFilter(CompareOp.GREATER,
							new BinaryComparator(filterValueByte));
					scan.setFilter(filter);
					break;
				}
			}
			
			scan.setRowPrefixFilter(prefixRowBytes);
			scan.setStartRow(startRowBytes);
			scan.setStopRow(stopRowBytes);
			scan.setCaching(1024 * 5);

            //扫描核心方法 - 核心方法
			resultScanners = table.getScanner(scan);

            //数据转换
			timestampValue = transferResults(resultScanners);
			resultScanners.close();

			startRowBytes = null;
			stopRowBytes = null;
			prefixRowBytes = null;
		} catch (IOException e) {
			LOGGER.error(String.format("HbaseUtil.scanPrefixRow has error, tablename = {}, prefixRow = {}", tablename,
					prefixRow), e);
		} finally {
			if (table != null) {
				table.close();
			}
		}
		return timestampValue;
	}

参考：Hbase配置文件详解（一） - 简书

Hbase使用与Hadoop相同的配置系统，所有配置文件都位于 conf/ 目录中，需要保持群集中每个节点的同步。

backup-masters - 这是一个纯文本文件，其中列出了主服务器应在其上启动备份主进程的主机列表，每行一台主机名（默认情况下不存在）hadoop-metric2-hbase.properties - 用于连接Hbase和Hadoop的Metric2框架，默认情况下只包含注释出的示例/hbase-env.cmd 和 /hbase-env.sh - 用于Windows和Linux/Unix环境的脚本，用于设置Hbase的工作环境，包括 JAVA和其他环境变量的配置（该文件包含许多注释示例来提供指导，该文件的改动需要重启Hbase才能生效）hbase-policy.xml - RPC服务器使用默认策略配置文件对客户端请求进行授权决策（仅在启用Hbase安全性的情况下使用）hbase-site.xml - 主要的Hbase配置文件，该文件覆盖了Hbase的默认配置的配置选项（可以在Hbase Web UI的Hbase配置选项中查看整个集群的有效配置，包括默认和覆盖的）log4j.properties - Hbase日志记录的配置文件regionservers - 纯文本文件，包含在Hbase集群中运行RegionSever的主机名或IP列表，每行一个（默认情况下，这个文件包含单条条目 localhost） 2、重要属性

用于启动Hbase：

hbase.tmp.dir：本地文件系统 的临时文件夹（重新启动计算机将清空 /tmp目录）

类型：string默认值：${java.io.tmpdir}/hbase-${user.name}hbase.rootdir：这个目录是region服务器共享的目录，用来持久存储Hbase的数据，URL必须完全正确，其中包含了文件系统的schema。例如，要表示HDFS中的/hbase目录，HDFS实例的namenode需要运行在服务器 namenode.example.org 的9000端口，且需要将这个属性设置为 hdfs://namenode.example.org:9000/hbase。默认情况下，Hbase是写到/tmp下的，如果不修改这个配置，数据将会在集群重启时丢失。

类型：string默认值：${hbase.tmp.dir}/hbasehbase.cluster.distributed：Hbase集群的运行模式，该值为 false 时，集群为单机模式；该值为 true 时，集群是分布式模式。如果将该值设置为false，则Hbase与Zookeeper的守护进程将运行在一个JVM中。

类型：boolean默认值：falsehbase.zookeeper.quorum：Zookeeper Quorum中的服务器列表，使用逗号分隔。默认设置为“host1.mydomain.com，host2.mydomain.com，host3.mydomain.com”，协助伪分布式模式使用。在完全分布式模式下，用户需要把所有Zookeeper Quorum节点添加进去，如果在 hbase-env.sh 文件中设置了 Hbase_MANAGES_ZK变量，则此列表中的节点就是我们启动或停止Zookeeper服务的节点。

类型：string默认值：localhost