神奇cpp

这个屌丝很懒，什么也没留下！

热门标签

Springboot整合HBase——大数据技术之HBase2.x_springboot hbase

作者：神奇cpp | 2024-07-31 08:33:45

踩

springboot hbase

一、简介

1.1、基本介绍

Apache HBase 是以hdfs为数据存储的，一种分布式、可扩展的noSql数据库。是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统，利用HBase技术可在廉价PC Server上搭建起大规模结构化存储集群。

HBase使用与BigTable（BigTable是一个稀疏的、分布式的、持久化的多维排序map）非常相似的数据模型。用户将数据行存储在带标签的表中。数据行具有可排序的键和任意数量的列。该表存储稀疏，因此如果用户喜欢，同一表中的行可以具有疯狂变化的列。

最终理解HBase数据模型的关键在于稀疏、分布式、多维、排序的映射。其中映射map指非关系型数据库的key-value结构

1.2、HBase逻辑结构

HBase可以用于存储多种结构的数据，以JSON为例，存储的数据原貌为：

{
	"row_key1":{
		"personal_info":{
			"name":"zhangsan",
			"city":"北京",
			"phone":"131********"
		},
		"office_info":{
			"tel":"010-1111111",
			"address":"atguigu"
		}
	},
	"row_key11":{
		"personal_info":{
			"city":"上海",
			"phone":"132********"
		},
	"office_info":{
		"tel":"010-1111111"
		} 
	},
"row_key2":{
......
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

在这里插入图片描述

1.3、HBase物理存储结构

物理存储结构即为数据映射关系，而在概念视图的空单元格，底层实际根本不存储。就是上面的空单元格没有存储。

在这里插入图片描述

1.4、数据模型

1、Name Space

命名空间，类似于关系型数据库的 database 概念，每个命名空间下有多个表。HBase 两
个自带的命名空间，分别是 hbase 和 default，hbase 中存放的是 HBase 内置的表，default表是用户默认使用的命名空间。

2、Table

类似于关系型数据库的表概念。不同的是，HBase 定义表时只需要声明列族即可，不需
要声明具体的列。因为数据存储时稀疏的，所有往 HBase 写入数据时，字段可以动态、按需
指定。因此，和关系型数据库相比，HBase 能够轻松应对字段变更的场景。

3、Row

HBase 表中的每行数据都由一个 RowKey 和多个 Column（列）组成，数据是按照 RowKey
的字典顺序存储的，并且查询数据时只能根据 RowKey 进行检索，所以 RowKey 的设计十分重
要。

4、Column

HBase 中的每个列都由 Column Family(列族)和 Column Qualifier（列限定符）进行限
定，例如 info：name，info：age。建表时，只需指明列族，而列限定符无需预先定义。

5、Time Stamp

用于标识数据的不同版本（version），每条数据写入时，系统会自动为其加上该字段，
其值为写入 HBase 的时间

6、Cell

由{rowkey, column Family：column Qualifier, timestamp} 唯一确定的单元。cell 中的数
据全部是字节码形式存贮。

1.5、HBase 基础架构

在这里插入图片描述

架构角色：

1、Master
实现类为 HMaster，负责监控集群中所有的 RegionServer 实例。主要作用如下：

（1）管理元数据表格 hbase:meta，接收用户对表格创建修改删除的命令并执行
（2）监控 region 是否需要进行负载均衡，故障转移和 region 的拆分。

通过启动多个后台线程监控实现上述功能:

①LoadBalancer 负载均衡器

周期性监控 region 分布在 regionServer 上面是否均衡，由参数 hbase.balancer.period 控
制周期时间，默认 5 分钟。

②CatalogJanitor 元数据管理器

定期检查和清理 hbase:meta 中的数据。meta 表内容在进阶中介绍。

③MasterProcWAL master 预写日志处理器

把 master 需要执行的任务记录到预写日志 WAL 中，如果 master 宕机，让 backupMaster读取日志继续干

2、Region Server

Region Server 实现类为 HRegionServer，主要作用如下:

（1）负责数据 cell 的处理，例如写入数据 put，查询数据 get 等
（2）拆分合并 region 的实际执行者，有 master 监控，有 regionServer 执行。

3、Zookeeper

HBase 通过 Zookeeper 来做 master 的高可用、记录 RegionServer 的部署信息、并且存储有 meta 表的位置信息。
HBase 对于数据的读写操作时直接访问 Zookeeper 的，在 2.3 版本推出 Master Registry模式，客户端可以直接访问 master。使用此功能，会加大对 master 的压力，减轻对 Zookeeper的压力

4、HDFS

HDFS 为 Hbase 提供最终的底层数据存储服务，同时为 HBase 提供高容错的支持。

二、安装部署

docker 安装HBase

$ docker pull harisekhon/hbase:latest
1

docker run -d -h docker-hbase \
        -p 2181:2181 \
        -p 8080:8080 \
        -p 8085:8085 \
        -p 9090:9090 \
        -p 9000:9000 \
        -p 9095:9095 \
        -p 16000:16000 \
        -p 16010:16010 \
        -p 16201:16201 \
        -p 16301:16301 \
        -p 16020:16020\
         -v /home/hbase-data/data:/hbase-data/data \
        --name hbase \
        harisekhon/hbase
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

访问HBase WebUI界面 : http://localhost:16010

在这里插入图片描述

三、HBase Shell 操作

3.1、进入 HBase 客户端命令行

先进入dockers容器内
进入shell命令模式

$ hbase shell
1

在这里插入图片描述

$ help
1

能够展示 HBase 中所有能使用的命令，主要使用的命令有 namespace 命令空间相关，DDL 创建修改表格，DML 写入读取数据。

3.2、Name Space 相关命令

1)、使用特定的 help 语法能够查看命令如何使用。

$ help 'create_namespace'
1

2)、创建命名空间 bigdata

$ create_namespace 'bigdata'
1

3)、查看所有的命名空间

$ list_namespace
1

3.3、DDL 相关命令

1)、创建表

在 bigdata 命名空间中创建表格 student，两个列族。info 列族数据维护的版本数为 5 个，
msg不写默认版本数为 1

$ create 'bigdata:student', {NAME => 'info', VERSIONS => 5}, {NAME => 'msg'}
1

如果创建表格只有一个列族，没有列族属性，可以简写。
如果不写命名空间，使用默认的命名空间 default。

$ create 'student1','info'
1

2)、查看表

查看所有的表名

$ list
1

查看一个表的详情

$ describe 'student1'
1

3)、修改表

表名创建时写的所有和列族相关的信息，都可以后续通过 alter 修改，包括增加删除列
族。

（1）增加列族和修改信息都使用覆盖的方法

$ alter 'student1', {NAME => 'f1', VERSIONS => 3}
1

（2）删除信息使用特殊的语法

$ alter 'student1', NAME => 'f1', METHOD => 'delete'
$ alter 'student1', 'delete' => 'f1'
1
2

4)、删除表

shell 中删除表格,需要先将表格状态设置为不可用。

$ disable 'student1'
$ drop 'student1'
1
2

3.4、DML 相关命令

1)、写入数据

在 HBase 中如果想要写入数据，只能添加结构中最底层的 cell。可以手动写入时间戳指
定 cell 的版本，推荐不写默认使用当前的系统时间。

// 参数1：表面  参数2:rowKey  参数3：列名  参数4：值
$ put 'bigdata:student','1001','info:name','zhangsan'
$ put 'bigdata:student','1001','info:name','lisi'
$ put 'bigdata:student','1001','info:age','18'
1
2
3
4

如果重复写入相同 rowKey，相同列的数据，会写入多个版本进行覆盖。

2)、读取数据

get 最大范围是一行数据，也可以进行列的过滤，读取数据的结果为多行 cell。

$ get 'bigdata:student','1001'  // 根据rowKey查询整条数据
$ get 'bigdata:student','1001' , {COLUMN => 'info:name'} // 只查这个列
1
2

也可以修改读取 cell 的版本数，默认读取一个。最多能够读取当前列族设置的维护版本
数。

$ get 'bigdata:student','1001' , {COLUMN => 'info:name', VERSIONS => 6}
1

scan 是扫描数据，能够读取多行数据，不建议扫描过多的数据，推荐使用 startRow 和
stopRow 来控制读取的数据，默认范围左闭右开。

scan 'bigdata:student',{STARTROW => '1001',STOPROW => '1002'}
1

3)、删除数据

delete 表示删除一个版本的数据，即为 1 个 cell，不填写版本默认删除最新的一个版本。

$ delete 'bigdata:student','1001','info:name'
1

deleteall 表示删除所有版本的数据，即为当前行当前列的多个 cell。（执行命令会标记数据为要删除，不会直接将数据彻底删除，删除数据只在特定时期清理磁盘时进行）

$ deleteall 'bigdata:student','1001','info:name'
1

四、HBase API

依赖

<dependency>
	<groupId>org.apache.hbase</groupId>
	<artifactId>hbase-client</artifactId>
	<version>2.4.13</version>
	<exclusions>
		<exclusion>
			<artifactId>slf4j-log4j12</artifactId>
			<groupId>org.slf4j</groupId>
		</exclusion>
	</exclusions>
</dependency>
1
2
3
4
5
6
7
8
9
10
11

yml添加hbase信息

hbase:
  zookeeper:
    quorum: localhost
    property:
      clientPort: 2181

zookeeper:
  znode:
    parent: /hbase
1
2
3
4
5
6
7
8
9

添加HBase配置信息类

package org.jeecg.common.util.hbase;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * HBase相关配置
 */
@Configuration
public class HBaseConfig {

	@Value("${hbase.zookeeper.quorum}")
	private String zookeeperQuorum;

	@Value("${hbase.zookeeper.property.clientPort}")
	private String clientPort;

	@Value("${zookeeper.znode.parent}")
	private String znodeParent;

	@Bean
	public HBaseService getHbaseService(){
		org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", zookeeperQuorum);
		conf.set("hbase.zookeeper.property.clientPort", clientPort);
		conf.set("zookeeper.znode.parent", znodeParent);
		return new HBaseService(conf);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

封装HBase工具类

package org.jeecg.common.util.hbase;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.math.BigInteger;
import java.text.MessageFormat;
import java.util.*;

@Slf4j
public class HBaseService {

	/**
	 * 声明静态配置
	 */
	private Configuration conf = null;
	private Connection connection = null;

	public HBaseService(Configuration conf) {

		this.conf = conf;
		try {
			connection = ConnectionFactory.createConnection(conf);
		} catch (IOException e) {
			log.error("获取HBase连接失败");
		}
	}

	/**
	 * 创建表
	 * @author 
	 * @date 2018/7/3 17:50
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param columnFamily 列族名
	 * @return void
	 */
	public boolean creatTable(String tableName, List<String> columnFamily) {
		Admin admin = null;
		try {
			admin = connection.getAdmin();

			List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>(columnFamily.size());

			columnFamily.forEach(cf -> {
				familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).build());
			});

			TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))
					.setColumnFamilies(familyDescriptors)
					.build();

			if (admin.tableExists(TableName.valueOf(tableName))) {
				log.debug("table Exists!");
			} else {
				admin.createTable(tableDescriptor);
				log.debug("create table Success!");
			}
		} catch (IOException e) {
			log.error(MessageFormat.format("创建表{0}失败",tableName),e);
			return false;
		}finally {
			close(admin,null,null);
		}
		return true;
	}

	/**
	 * 预分区创建表
	 * @param tableName 表名
	 * @param columnFamily 列族名的集合
	 * @param splitKeys 预分期region
	 * @return 是否创建成功
	 */
	public boolean createTableBySplitKeys(String tableName, List<String> columnFamily, byte[][] splitKeys) {
		Admin admin = null;
		try {
			if (StringUtils.isBlank(tableName) || columnFamily == null
					|| columnFamily.size() == 0) {
				log.error("===Parameters tableName|columnFamily should not be null,Please check!===");
				return false;
			}
			admin = connection.getAdmin();
			if (admin.tableExists(TableName.valueOf(tableName))) {
				return true;
			} else {
				List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>(columnFamily.size());

				columnFamily.forEach(cf -> {
					familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).build());
				});

				TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))
						.setColumnFamilies(familyDescriptors)
						.build();

				//指定splitkeys
				admin.createTable(tableDescriptor,splitKeys);
				log.info("===Create Table " + tableName
						+ " Success!columnFamily:" + columnFamily.toString()
						+ "===");
			}
		} catch (IOException e) {
			log.error("",e);
			return false;
		}finally {
			close(admin,null,null);
		}

		return true;
	}

	/**
	 * 自定义获取分区splitKeys
	 */
	public byte[][] getSplitKeys(String[] keys){
		if(keys==null){
			//默认为10个分区
			keys = new String[] {  "1|", "2|", "3|", "4|",
					"5|", "6|", "7|", "8|", "9|" };
		}
		byte[][] splitKeys = new byte[keys.length][];
		//升序排序
		TreeSet<byte[]> rows = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
		for(String key : keys){
			rows.add(Bytes.toBytes(key));
		}

		Iterator<byte[]> rowKeyIter = rows.iterator();
		int i=0;
		while (rowKeyIter.hasNext()) {
			byte[] tempRow = rowKeyIter.next();
			rowKeyIter.remove();
			splitKeys[i] = tempRow;
			i++;
		}
		return splitKeys;
	}

	/**
	 * 按startKey和endKey，分区数获取分区
	 */
	public static byte[][] getHexSplits(String startKey, String endKey, int numRegions) {
		byte[][] splits = new byte[numRegions-1][];
		BigInteger lowestKey = new BigInteger(startKey, 16);
		BigInteger highestKey = new BigInteger(endKey, 16);
		BigInteger range = highestKey.subtract(lowestKey);
		BigInteger regionIncrement = range.divide(BigInteger.valueOf(numRegions));
		lowestKey = lowestKey.add(regionIncrement);
		for(int i=0; i < numRegions-1;i++) {
			BigInteger key = lowestKey.add(regionIncrement.multiply(BigInteger.valueOf(i)));
			byte[] b = String.format("%016x", key).getBytes();
			splits[i] = b;
		}
		return splits;
	}

	/**
	 * 获取table
	 * @param tableName 表名
	 * @return Table
	 * @throws IOException IOException
	 */
	private Table getTable(String tableName) throws IOException {
		return connection.getTable(TableName.valueOf(tableName));
	}

	/**
	 * 查询库中所有表的表名
	 */
	public List<String> getAllTableNames(){
		List<String> result = new ArrayList<>();

		Admin admin = null;
		try {
			admin = connection.getAdmin();
			TableName[] tableNames = admin.listTableNames();

			for(TableName tableName : tableNames){
				result.add(tableName.getNameAsString());
			}
		}catch (IOException e) {
			log.error("获取所有表的表名失败",e);
		}finally {
			close(admin,null,null);
		}

		return result;
	}

	/**
	 * 遍历查询指定表中的所有数据
	 * @author 
	 * @date 2018/7/3 18:21
	 * @since 1.0.0
	 * @param tableName 表名
	 * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
	 */
	public Map<String, Map<String,String>> getResultScanner(String tableName){
		Scan scan = new Scan();
		return this.queryData(tableName,scan);
	}

	/**
	 * 根据startRowKey和stopRowKey遍历查询指定表中的所有数据
	 * @author 
	 * @date 2018/7/4 18:21
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param startRowKey 起始rowKey
	 * @param stopRowKey 结束rowKey
	 * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
	 */
	public Map<String,Map<String,String>> getResultScanner(String tableName, String startRowKey, String stopRowKey){
		Scan scan = new Scan();

		if(StringUtils.isNotBlank(startRowKey) && StringUtils.isNotBlank(stopRowKey)){
			scan.withStartRow(Bytes.toBytes(startRowKey));
			scan.withStopRow(Bytes.toBytes(stopRowKey));
		}

		return this.queryData(tableName,scan);
	}

	/**
	 * 通过行前缀过滤器查询数据
	 * @author 
	 * @date 2018/7/4 18:21
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param prefix 以prefix开始的行键
	 * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
	 */
	public Map<String,Map<String,String>> getResultScannerPrefixFilter(String tableName, String prefix){
		Scan scan = new Scan();

		if(StringUtils.isNotBlank(prefix)){
			Filter filter = new PrefixFilter(Bytes.toBytes(prefix));
			scan.setFilter(filter);
		}

		return this.queryData(tableName,scan);
	}

	/**
	 * 通过列前缀过滤器查询数据
	 * @author 
	 * @date 2018/7/4 18:21
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param prefix 以prefix开始的列名
	 * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
	 */
	public Map<String,Map<String,String>> getResultScannerColumnPrefixFilter(String tableName, String prefix){
		Scan scan = new Scan();

		if(StringUtils.isNotBlank(prefix)){
			Filter filter = new ColumnPrefixFilter(Bytes.toBytes(prefix));
			scan.setFilter(filter);
		}

		return this.queryData(tableName,scan);
	}

	/**
	 * 查询行键中包含特定字符的数据
	 * @author 
	 * @date 2018/7/4 18:21
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param keyword 包含指定关键词的行键
	 * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
	 */
	public Map<String,Map<String,String>> getResultScannerRowFilter(String tableName, String keyword){
		Scan scan = new Scan();

		if(StringUtils.isNotBlank(keyword)){
			Filter filter = new RowFilter(CompareOperator.GREATER_OR_EQUAL,new SubstringComparator(keyword));
			scan.setFilter(filter);
		}

		return this.queryData(tableName,scan);
	}

	/**
	 * 查询列名中包含特定字符的数据
	 * @author 
	 * @date 2018/7/4 18:21
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param keyword 包含指定关键词的列名
	 * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
	 */
	public Map<String,Map<String,String>> getResultScannerQualifierFilter(String tableName, String keyword){
		Scan scan = new Scan();

		if(StringUtils.isNotBlank(keyword)){
			Filter filter = new QualifierFilter(CompareOperator.GREATER_OR_EQUAL,new SubstringComparator(keyword));
			scan.setFilter(filter);
		}

		return this.queryData(tableName,scan);
	}



	/**
	 * 通过表名以及过滤条件查询数据
	 * @author 
	 * @date 2018/7/4 16:13
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param scan 过滤条件
	 * @return java.util.Map<java.lang.String,java.util.Map<java.lang.String,java.lang.String>>
	 */
	private Map<String,Map<String,String>> queryData(String tableName,Scan scan){
		//<rowKey,对应的行数据>
		Map<String,Map<String,String>> result = new HashMap<>();

		ResultScanner rs = null;
		// 获取表
		Table table= null;
		try {
			table = getTable(tableName);
			rs = table.getScanner(scan);
			for (Result r : rs) {
				//每一行数据
				Map<String,String> columnMap = new HashMap<>();
				String rowKey = null;
				for (Cell cell : r.listCells()) {
					if(rowKey == null){
						rowKey = Bytes.toString(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength());
					}
					columnMap.put(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()), Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
				}

				if(rowKey != null){
					result.put(rowKey,columnMap);
				}
			}
		}catch (IOException e) {
			log.error(MessageFormat.format("遍历查询指定表中的所有数据失败,tableName:{0}"
					,tableName),e);
		}finally {
			close(null,rs,table);
		}

		return result;
	}

	/**
	 * 根据tableName和rowKey精确查询一行的数据
	 * @author 
	 * @date 2018/7/3 16:07
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param rowKey 行键
	 * @return java.util.Map<java.lang.String,java.lang.String> 返回一行的数据
	 */
	public Map<String,String> getRowData(String tableName, String rowKey){
		//返回的键值对
		Map<String,String> result = new HashMap<>();

		Get get = new Get(Bytes.toBytes(rowKey));
		// 获取表
		Table table= null;
		try {
			table = getTable(tableName);
			Result hTableResult = table.get(get);
			if (hTableResult != null && !hTableResult.isEmpty()) {
				for (Cell cell : hTableResult.listCells()) {
//                System.out.println("family:" + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
//                System.out.println("qualifier:" + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
//                System.out.println("value:" + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
//                System.out.println("Timestamp:" + cell.getTimestamp());
//                System.out.println("-------------------------------------------");
					result.put(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()), Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
				}
			}
		}catch (IOException e) {
			log.error(MessageFormat.format("查询一行的数据失败,tableName:{0},rowKey:{1}"
					,tableName,rowKey),e);
		}finally {
			close(null,null,table);
		}

		return result;
	}

	/**
	 * 根据tableName、rowKey、familyName、column查询指定单元格的数据
	 * @author 
	 * @date 2018/7/4 10:58
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param rowKey rowKey
	 * @param familyName 列族名
	 * @param columnName 列名
	 * @return java.lang.String
	 */
	public String getColumnValue(String tableName, String rowKey, String familyName, String columnName){
		String str = null;
		Get get = new Get(Bytes.toBytes(rowKey));
		// 获取表
		Table table= null;
		try {
			table = getTable(tableName);
			Result result = table.get(get);
			if (result != null && !result.isEmpty()) {
				Cell cell = result.getColumnLatestCell(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
				if(cell != null){
					str = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
				}
			}
		} catch (IOException e) {
			log.error(MessageFormat.format("查询指定单元格的数据失败,tableName:{0},rowKey:{1},familyName:{2},columnName:{3}"
					,tableName,rowKey,familyName,columnName),e);
		}finally {
			close(null,null,table);
		}

		return str;
	}

	/**
	 * 根据tableName、rowKey、familyName、column查询指定单元格多个版本的数据
	 * @author 
	 * @date 2018/7/4 11:16
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param rowKey rowKey
	 * @param familyName 列族名
	 * @param columnName 列名
	 * @param versions 需要查询的版本数
	 * @return java.util.List<java.lang.String>
	 */
	public List<String> getColumnValuesByVersion(String tableName, String rowKey, String familyName, String columnName,int versions) {
		//返回数据
		List<String> result = new ArrayList<>(versions);

		// 获取表
		Table table= null;
		try {
			table = getTable(tableName);
			Get get = new Get(Bytes.toBytes(rowKey));
			get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
			//读取多少个版本
			get.readVersions(versions);
			Result hTableResult = table.get(get);
			if (hTableResult != null && !hTableResult.isEmpty()) {
				for (Cell cell : hTableResult.listCells()) {
					result.add(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
				}
			}
		} catch (IOException e) {
			log.error(MessageFormat.format("查询指定单元格多个版本的数据失败,tableName:{0},rowKey:{1},familyName:{2},columnName:{3}"
					,tableName,rowKey,familyName,columnName),e);
		}finally {
			close(null,null,table);
		}

		return result;
	}

	/**
	 * 为表添加 or 更新数据
	 * @author 
	 * @date 2018/7/3 17:26
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param rowKey rowKey
	 * @param familyName 列族名
	 * @param columns 列名数组
	 * @param values 列值得数组
	 */
	public void putData(String tableName,String rowKey, String familyName, String[] columns, String[] values) {
		// 获取表
		Table table= null;
		try {
			table=getTable(tableName);

			putData(table,rowKey,tableName,familyName,columns,values);
		} catch (Exception e) {
			log.error(MessageFormat.format("为表添加 or 更新数据失败,tableName:{0},rowKey:{1},familyName:{2}"
					,tableName,rowKey,familyName),e);
		}finally {
			close(null,null,table);
		}
	}

	/**
	 * 为表添加 or 更新数据
	 * @author 
	 * @date 2018/7/3 17:26
	 * @since 1.0.0
	 * @param table Table
	 * @param rowKey rowKey
	 * @param tableName 表名
	 * @param familyName 列族名
	 * @param columns 列名数组
	 * @param values 列值得数组
	 */
	private void putData(Table table, String rowKey, String tableName, String familyName, String[] columns, String[] values) {
		try {
			//设置rowkey
			Put put = new Put(Bytes.toBytes(rowKey));

			if(columns != null && values != null && columns.length == values.length){
				for(int i=0;i<columns.length;i++){
					if(columns[i] != null && values[i] != null){
						put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columns[i]), Bytes.toBytes(values[i]));
					}else{
						throw new NullPointerException(MessageFormat.format("列名和列数据都不能为空,column:{0},value:{1}"
								,columns[i],values[i]));
					}
				}
			}

			table.put(put);
			log.debug("putData add or update data Success,rowKey:" + rowKey);
			table.close();
		} catch (Exception e) {
			log.error(MessageFormat.format("为表添加 or 更新数据失败,tableName:{0},rowKey:{1},familyName:{2}"
					,tableName,rowKey,familyName),e);
		}
	}

	/**
	 * 为表的某个单元格赋值
	 * @author 
	 * @date 2018/7/4 10:20
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param rowKey rowKey
	 * @param familyName 列族名
	 * @param column1 列名
	 * @param value1 列值
	 */
	public void setColumnValue(String tableName, String rowKey, String familyName, String column1, String value1){
		Table table=null;
		try {
			// 获取表
			table=getTable(tableName);
			// 设置rowKey
			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column1), Bytes.toBytes(value1));

			table.put(put);
			log.debug("add data Success!");
		}catch (IOException e) {
			log.error(MessageFormat.format("为表的某个单元格赋值失败,tableName:{0},rowKey:{1},familyName:{2},column:{3}"
					,tableName,rowKey,familyName,column1),e);
		}finally {
			close(null,null,table);
		}
	}

	/**
	 * 删除指定的单元格
	 * @author 
	 * @date 2018/7/4 11:41
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param rowKey rowKey
	 * @param familyName 列族名
	 * @param columnName 列名
	 * @return boolean
	 */
	public boolean deleteColumn(String tableName, String rowKey, String familyName, String columnName) {
		Table table=null;
		Admin admin = null;
		try {
			admin = connection.getAdmin();

			if(admin.tableExists(TableName.valueOf(tableName))){
				// 获取表
				table=getTable(tableName);
				Delete delete = new Delete(Bytes.toBytes(rowKey));
				// 设置待删除的列
				delete.addColumns(Bytes.toBytes(familyName), Bytes.toBytes(columnName));

				table.delete(delete);
				log.debug(MessageFormat.format("familyName({0}):columnName({1})is deleted!",familyName,columnName));
			}

		}catch (IOException e) {
			log.error(MessageFormat.format("删除指定的列失败,tableName:{0},rowKey:{1},familyName:{2},column:{3}"
					,tableName,rowKey,familyName,columnName),e);
			return false;
		}finally {
			close(admin,null,table);
		}
		return true;
	}

	/**
	 * 根据rowKey删除指定的行
	 * @author 
	 * @date 2018/7/4 13:26
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param rowKey rowKey
	 * @return boolean
	 */
	public boolean deleteRow(String tableName, String rowKey) {
		Table table=null;
		Admin admin = null;
		try {
			admin = connection.getAdmin();

			if(admin.tableExists(TableName.valueOf(tableName))){
				// 获取表
				table=getTable(tableName);
				Delete delete = new Delete(Bytes.toBytes(rowKey));

				table.delete(delete);
				log.debug(MessageFormat.format("row({0}) is deleted!",rowKey));
			}
		}catch (IOException e) {
			log.error(MessageFormat.format("删除指定的行失败,tableName:{0},rowKey:{1}"
					,tableName,rowKey),e);
			return false;
		}finally {
			close(admin,null,table);
		}
		return true;
	}

	/**
	 * 根据columnFamily删除指定的列族
	 * @author 
	 * @date 2018/7/4 13:26
	 * @since 1.0.0
	 * @param tableName 表名
	 * @param columnFamily 列族
	 * @return boolean
	 */
	public boolean deleteColumnFamily(String tableName, String columnFamily) {
		Admin admin = null;
		try {
			admin = connection.getAdmin();

			if(admin.tableExists(TableName.valueOf(tableName))){
				admin.deleteColumnFamily(TableName.valueOf(tableName), Bytes.toBytes(columnFamily));
				log.debug(MessageFormat.format("familyName({0}) is deleted!",columnFamily));
			}
		}catch (IOException e) {
			log.error(MessageFormat.format("删除指定的列族失败,tableName:{0},columnFamily:{1}"
					,tableName,columnFamily),e);
			return false;
		}finally {
			close(admin,null,null);
		}
		return true;
	}

	/**
	 * 删除表
	 * @author 
	 * @date 2018/7/3 18:02
	 * @since 1.0.0
	 * @param tableName 表名
	 */
	public boolean deleteTable(String tableName){
		Admin admin = null;
		try {
			admin = connection.getAdmin();

			if(admin.tableExists(TableName.valueOf(tableName))){
				admin.disableTable(TableName.valueOf(tableName));
				admin.deleteTable(TableName.valueOf(tableName));
				log.debug(tableName + "is deleted!");
			}
		}catch (IOException e) {
			log.error(MessageFormat.format("删除指定的表失败,tableName:{0}"
					,tableName),e);
			return false;
		}finally {
			close(admin,null,null);
		}
		return true;
	}

	/**
	 * 关闭流
	 */
	private void close(Admin admin, ResultScanner rs, Table table){
		if(admin != null){
			try {
				admin.close();
			} catch (IOException e) {
				log.error("关闭Admin失败",e);
			}
		}

		if(rs != null){
			rs.close();
		}

		if(table != null){
			try {
				table.close();
			} catch (IOException e) {
				log.error("关闭Table失败",e);
			}
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716

测试

在这里插入图片描述

结果与shell终端查出来的一样，测试成功！！！

如果出现报错：Can not resolve docker-hbase, please check your network java.net.UnknownHostException: docker-hbase
在这里插入图片描述
需添加hosts映射地址

五、整合Phoenix

5.1、Phoenix 简介

Phoenix 是 HBase 的开源 SQL 皮肤。可以使用标准 JDBC API 代替 HBase 客户端 API来创建表，插入数据和查询 HBase 数据。就是像写sql一样，去插入查询HBase数据。

5.2、为什么使用 Phoenix

官方给的解释为：在 Client 和 HBase 之间放一个 Phoenix 中间层不会减慢速度，因为用户编写的数据处理代码和 Phoenix 编写的没有区别（更不用说你写的垃圾的多），不仅如此Phoenix 对于用户输入的 SQL 同样会有大量的优化手段（就像 hive 自带 sql 优化器一样）。

5.3、下载安装

1、下载
下载链接: https://phoenix.apache.org/download.html

2、解压

$ tar -zxvf phoenix-hbase-2.1-5.1.2-bin.tar.gz
1

3、拷贝server包到hbase的lib文件下,注意是hbase文件夹下的lib。

$ docker cp phoenix-server-hbase-2.1-5.1.2.jar 容器Id:、/hbase/lib
1

4、进入hbase 的bin目录,重启hbase

$ stop-hbase.sh
$ start-hbase.sh
1
2

5、到phoenix的bin目录，启动

$ ./sqlline.py 192.168.220.152:2181  // 后面跟zookeeperIP端口
1

在这里插入图片描述
这时候就可以输入sql了，例如：

// 创建一张t_user 表
CREATE TABLE IF NOT EXISTS t_user (
      name VARCHAR(20) NOT NULL,
      age integer(3) NOT NULL,
      address VARCHAR,
      updatetime timestamp,
      inserttime timestamp
      CONSTRAINT pk_name_age PRIMARY KEY (name, age)// 联合主键
);

// 添加和更新数据都用 upsert, 没有 insert 和 update
UPSERT INTO t_user (name, age, address) values ('test01',12,'上海市松江区');
UPSERT INTO t_user (name, age, address) values ('test02',13,'上海市金山区');

SELECT * FROM t_user;
SELECT COUNT(*) FROM t_user;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

在这里插入图片描述

用hbase终端查看，表也在，测试成功！！！
在这里插入图片描述

5.4、Phoenix Shell 相关命令

1)、显示所有表

$ !table 或 !tables
1

2)、创建表
直接指定单个列作为 RowKey

$ CREATE TABLE IF NOT EXISTS student(
		id VARCHAR primary key,
		name VARCHAR,
		age BIGINT,
		addr VARCHAR);
1
2
3
4
5

在 phoenix 中，表名等会自动转换为大写，若要小写，使用双引号，如"us_population"。

指定多个列的联合作为 RowKey

$ CREATE TABLE IF NOT EXISTS t_user (
      name VARCHAR(20) NOT NULL,
      age integer(3) NOT NULL,
      address VARCHAR,
      updatetime timestamp,
      inserttime timestamp
      CONSTRAINT pk_name_age PRIMARY KEY (name, age)// 联合主键
);
1
2
3
4
5
6
7
8

注：Phoenix 中建表，会在 HBase 中创建一张对应的表。为了减少数据对磁盘空间的占用，Phoenix 默认会对 HBase 中的列名做编码处理。具体规则可参考官网链接：https://phoenix.apache.org/columnencoding.html，若不想对列名编码，可在建表语句末尾加上 COLUMN_ENCODED_BYTES = 0;

3)、插入数据

// 添加和更新数据都用 upsert, 没有 insert 和 update
UPSERT INTO t_user (name, age, address) values ('test01',12,'上海市松江区');
1
2

4)、查询记录

SELECT * FROM t_user;

SELECT COUNT(*) FROM t_user;

SELECT * FROM t_user where name='test01';
1
2
3
4
5

5)、删除记录

delete from t_user where name='test01';
1

6)、删除表

drop table t_user;
1

7)、退出命令行

!quit
1

关于 Phoenix 的语法建议使用时直接查看官网：
链接: https://phoenix.apache.org/language/index.html

5.5、表的映射

默认情况下， HBase 中已存在的表，通过 Phoenix 是不可见的。如果要在 Phoenix 中操
作 HBase 中已存在的表，可以在 Phoenix 中进行表的映射。映射方式有两种：视图映射和表
映射。

例如创建hbase表test

 create 'test','info1','info2'
1

在这里插入图片描述

1)、视图映射

Phoenix 创建的视图是只读的，所以只能用来做查询，无法通过视图对数据进行修改等操作。在 phoenix 中创建关联 test 表的视图

create view "test"(id varchar primary key,
"info1"."name" varchar, 
"info2"."address" varchar);

// 删除视图
 drop view "test";
1
2
3
4
5
6

数字类型说明：

HBase 中的数字，底层存储为补码，而 Phoenix 中的数字，底层存储为在补码的基础上，将符号位反转。故当在 Phoenix 中建表去映射 HBase 中已存在的表，当 HBase 中有数字类型的字段时，会出现解析错误的现象。

例如：

//  Hbase 演示：
create 'test_number','info'

put 'test_number','1001','info:number',Bytes.toBytes(1000)
scan 'test_number',{COLUMNS => 'info:number:toLong'}

// phoenix 演示：
create view "test_number"(id varchar primary key,"info"."number" 
bigint);

select * from "test_number";
1
2
3
4
5
6
7
8
9
10
11

解决上述问题的方案有以下两种：

（1）Phoenix 种提供了 unsigned_int，unsigned_long 等无符号类型，其对数字的编码解码方式和 HBase 是相同的，如果无需考虑负数，那在 Phoenix 中建表时采用无符号类型是最合适的选择.

// phoenix 演示：
drop view "test_number";

create view "test_number"(id varchar primary key,"info"."number" 
unsigned_long);

select * from "test_number";
1
2
3
4
5
6
7

（2）如需考虑负数的情况，则可通过 Phoenix 自定义函数，将数字类型的最高位，即
符号位反转即可，自定义函数可参考如下链接：https://phoenix.apache.org/udf.html

5.6、 Hbase Phoenix 数据库连接工具

下载工具链接: https://github.com/dbeaver/dbeaver/releases/download/6.1.5/dbeaver-ce-6.1.5-x86_64-setup.exe

安装完成后, 在安装目录找到 dbeaver.ini 文件, 在首行添加本机jdk安装目录:

-vm
C:\Program Files\Java\jdk1.8.0_281\bin // 以本机为准
1
2

在这里插入图片描述
上方添加的文件为下载phoenix时的client包

在这里插入图片描述

输入sql测试，成功！！！
在这里插入图片描述

5.7、Springboot集成Phoenix

一直没成功，后续补上。。。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/神奇cpp/article/detail/907918