快速学习-HBaseAPI操作

2020-02-21 15:33:46 浏览数 (1)

第6章 HBase API操作

6.1 环境准备

新建项目后在pom.xml中添加依赖：

代码语言：javascript复制
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-server</artifactId>
    <version>1.3.1</version>
</dependency>

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>1.3.1</version>
</dependency>

<dependency>
	<groupId>jdk.tools</groupId>
	<artifactId>jdk.tools</artifactId>
	<version>1.8</version>
	<scope>system</scope>
	<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
</dependency>

6.2 HBaseAPI

6.2.1 获取Configuration对象

代码语言：javascript复制
public static Configuration conf;
static{
	//使用HBaseConfiguration的单例方法实例化
	conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "192.168.9.102");
conf.set("hbase.zookeeper.property.clientPort", "2181");
}

6.2.2 判断表是否存在

代码语言：javascript复制
public static boolean isTableExist(String tableName) throws MasterNotRunningException,
 ZooKeeperConnectionException, IOException{
	//在HBase中管理、访问表需要先创建HBaseAdmin对象
//Connection connection = ConnectionFactory.createConnection(conf);
//HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
	HBaseAdmin admin = new HBaseAdmin(conf);
	return admin.tableExists(tableName);
}

6.2.3 创建表

代码语言：javascript复制
public static void createTable(String tableName, String... columnFamily) throws
 MasterNotRunningException, ZooKeeperConnectionException, IOException{
	HBaseAdmin admin = new HBaseAdmin(conf);
	//判断表是否存在
	if(isTableExist(tableName)){
		System.out.println("表"   tableName   "已存在");
		//System.exit(0);
	}else{
		//创建表属性对象,表名需要转字节
		HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
		//创建多个列族
		for(String cf : columnFamily){
			descriptor.addFamily(new HColumnDescriptor(cf));
		}
		//根据对表的配置，创建表
		admin.createTable(descriptor);
		System.out.println("表"   tableName   "创建成功！");
	}
}

6.2.4 删除表

代码语言：javascript复制
public static void dropTable(String tableName) throws MasterNotRunningException,
 ZooKeeperConnectionException, IOException{
	HBaseAdmin admin = new HBaseAdmin(conf);
	if(isTableExist(tableName)){
		admin.disableTable(tableName);
		admin.deleteTable(tableName);
		System.out.println("表"   tableName   "删除成功！");
	}else{
		System.out.println("表"   tableName   "不存在！");
	}
}

6.2.5 向表中插入数据

代码语言：javascript复制
public static void addRowData(String tableName, String rowKey, String columnFamily, String
 column, String value) throws IOException{
	//创建HTable对象
	HTable hTable = new HTable(conf, tableName);
	//向表中插入数据
	Put put = new Put(Bytes.toBytes(rowKey));
	//向Put对象中组装数据
	put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
	hTable.put(put);
	hTable.close();
	System.out.println("插入数据成功");
}

6.2.6 删除多行数据

代码语言：javascript复制
public static void deleteMultiRow(String tableName, String... rows) throws IOException{
	HTable hTable = new HTable(conf, tableName);
	List<Delete> deleteList = new ArrayList<Delete>();
	for(String row : rows){
		Delete delete = new Delete(Bytes.toBytes(row));
		deleteList.add(delete);
	}
	hTable.delete(deleteList);
	hTable.close();
}

6.2.7 获取所有数据

代码语言：javascript复制
public static void getAllRows(String tableName) throws IOException{
	HTable hTable = new HTable(conf, tableName);
	//得到用于扫描region的对象
	Scan scan = new Scan();
	//使用HTable得到resultcanner实现类的对象
	ResultScanner resultScanner = hTable.getScanner(scan);
	for(Result result : resultScanner){
		Cell[] cells = result.rawCells();
		for(Cell cell : cells){
			//得到rowkey
			System.out.println("行键:"   Bytes.toString(CellUtil.cloneRow(cell)));
			//得到列族
			System.out.println("列族"   Bytes.toString(CellUtil.cloneFamily(cell)));
			System.out.println("列:"   Bytes.toString(CellUtil.cloneQualifier(cell)));
			System.out.println("值:"   Bytes.toString(CellUtil.cloneValue(cell)));
		}
	}
}

6.2.8 获取某一行数据

代码语言：javascript复制
public static void getRow(String tableName, String rowKey) throws IOException{
	HTable table = new HTable(conf, tableName);
	Get get = new Get(Bytes.toBytes(rowKey));
	//get.setMaxVersions();显示所有版本
    //get.setTimeStamp();显示指定时间戳的版本
	Result result = table.get(get);
	for(Cell cell : result.rawCells()){
		System.out.println("行键:"   Bytes.toString(result.getRow()));
		System.out.println("列族"   Bytes.toString(CellUtil.cloneFamily(cell)));
		System.out.println("列:"   Bytes.toString(CellUtil.cloneQualifier(cell)));
		System.out.println("值:"   Bytes.toString(CellUtil.cloneValue(cell)));
		System.out.println("时间戳:"   cell.getTimestamp());
	}
}

6.2.9 获取某一行指定“列族:列”的数据

代码语言：javascript复制
public static void getRowQualifier(String tableName, String rowKey, String family, String
 qualifier) throws IOException{
	HTable table = new HTable(conf, tableName);
	Get get = new Get(Bytes.toBytes(rowKey));
	get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
	Result result = table.get(get);
	for(Cell cell : result.rawCells()){
		System.out.println("行键:"   Bytes.toString(result.getRow()));
		System.out.println("列族"   Bytes.toString(CellUtil.cloneFamily(cell)));
		System.out.println("列:"   Bytes.toString(CellUtil.cloneQualifier(cell)));
		System.out.println("值:"   Bytes.toString(CellUtil.cloneValue(cell)));
	}
}

6.3 MapReduce

通过HBase的相关JavaAPI，我们可以实现伴随HBase操作的MapReduce过程，比如使用MapReduce将数据从本地文件系统导入到HBase的表中，比如我们从HBase中读取一些原始数据后使用MapReduce做数据分析。

6.3.1 官方HBase-MapReduce

  1. 查看HBase的MapReduce任务的执行
代码语言：javascript复制
$ bin/hbase mapredcp
  1. 环境变量的导入 （1）执行环境变量的导入（临时生效，在命令行执行下述操作）
代码语言：javascript复制
$ export HBASE_HOME=/opt/module/hbase-1.3.1
$ export HADOOP_HOME=/opt/module/hadoop-2.7.2
$ export HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`

（2）永久生效：在/etc/profile配置

代码语言：javascript复制
export HBASE_HOME=/opt/module/hbase-1.3.1
export HADOOP_HOME=/opt/module/hadoop-2.7.2

并在hadoop-env.sh中配置：（注意：在for循环之后配）

代码语言：javascript复制
export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/opt/module/hbase/lib/*
  1. 运行官方的MapReduce任务 – 案例一：统计Student表中有多少行数据
代码语言：javascript复制
$ /opt/module/hadoop-2.7.2/bin/yarn jar lib/hbase-server-1.3.1.jar rowcounter student

– 案例二：使用MapReduce将本地数据导入到HBase 1）在本地创建一个tsv格式的文件：fruit.tsv

代码语言：javascript复制
1001	Apple	Red
1002	Pear		Yellow
1003	Pineapple	Yellow

2）创建HBase表

代码语言：javascript复制
hbase(main):001:0> create 'fruit','info'

3）在HDFS中创建input_fruit文件夹并上传fruit.tsv文件

代码语言：javascript复制
$ /opt/module/hadoop-2.7.2/bin/hdfs dfs -mkdir /input_fruit/
$ /opt/module/hadoop-2.7.2/bin/hdfs dfs -put fruit.tsv /input_fruit/

4）执行MapReduce到HBase的fruit表中

代码语言：javascript复制
$ /opt/module/hadoop-2.7.2/bin/yarn jar lib/hbase-server-1.3.1.jar importtsv 
-Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:color fruit 
hdfs://hadoop102:9000/input_fruit

5）使用scan命令查看导入后的结果

代码语言：javascript复制
hbase(main):001:0> scan ‘fruit’

6.3.2 自定义HBase-MapReduce1

目标：将fruit表中的一部分数据，通过MR迁入到fruit_mr表中。 分步实现：

  1. 构建ReadFruitMapper类，用于读取fruit表中的数据
代码语言：javascript复制
public class ReadFruitMapper extends TableMapper<ImmutableBytesWritable, Put> {

	@Override
	protected void map(ImmutableBytesWritable key, Result value, Context context) 
	throws IOException, InterruptedException {
	//将fruit的name和color提取出来，相当于将每一行数据读取出来放入到Put对象中。
		Put put = new Put(key.get());
		//遍历添加column行
		for(Cell cell: value.rawCells()){
			//添加/克隆列族:info
			if("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))){
				//添加/克隆列：name
				if("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
					//将该列cell加入到put对象中
					put.add(cell);
					//添加/克隆列:color
				}else if("color".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
					//向该列cell加入到put对象中
					put.add(cell);
				}
			}
		}
		//将从fruit读取到的每行数据写入到context中作为map的输出
		context.write(key, put);
	}
}
  1. 构建WriteFruitMRReducer类，用于将读取到的fruit表中的数据写入到fruit_mr表中
代码语言：javascript复制
public class WriteFruitMRReducer extends TableReducer<ImmutableBytesWritable, Put, NullWritable> {
	@Override
	protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) 
	throws IOException, InterruptedException {
		//读出来的每一行数据写入到fruit_mr表中
		for(Put put: values){
			context.write(NullWritable.get(), put);
		}
	}
}
  1. 构建Fruit2FruitMRRunner extends Configured implements Tool用于组装运行Job任务
代码语言：javascript复制
//组装Job
	public int run(String[] args) throws Exception {
		//得到Configuration
		Configuration conf = this.getConf();
		//创建Job任务
		Job job = Job.getInstance(conf, this.getClass().getSimpleName());
		job.setJarByClass(Fruit2FruitMRRunner.class);

		//配置Job
		Scan scan = new Scan();
		scan.setCacheBlocks(false);
		scan.setCaching(500);

		//设置Mapper，注意导入的是mapreduce包下的，不是mapred包下的，后者是老版本
		TableMapReduceUtil.initTableMapperJob(
		"fruit", //数据源的表名
		scan, //scan扫描控制器
		ReadFruitMapper.class,//设置Mapper类
		ImmutableBytesWritable.class,//设置Mapper输出key类型
		Put.class,//设置Mapper输出value值类型
		job//设置给哪个JOB
		);
		//设置Reducer
		TableMapReduceUtil.initTableReducerJob("fruit_mr", WriteFruitMRReducer.class, job);
		//设置Reduce数量，最少1个
		job.setNumReduceTasks(1);

		boolean isSuccess = job.waitForCompletion(true);
		if(!isSuccess){
			throw new IOException("Job running with error");
		}
		return isSuccess ? 0 : 1;
	}
  1. 主函数中调用运行该Job任务
代码语言：javascript复制
public static void main( String[] args ) throws Exception{
	Configuration conf = HBaseConfiguration.create();
	int status = ToolRunner.run(conf, new Fruit2FruitMRRunner(), args);
	System.exit(status);
}
  1. 打包运行任务
代码语言：javascript复制
$ /opt/module/hadoop-2.7.2/bin/yarn jar ~/softwares/jars/hbase-0.0.1-SNAPSHOT.jar
 com.z.hbase.mr1.Fruit2FruitMRRunner

提示：运行任务前，如果待数据导入的表不存在，则需要提前创建。 提示：maven打包命令：-P local clean package或-P dev clean package install（将第三方jar包一同打包，需要插件：maven-shade-plugin）

6.3.3 自定义HBase-MapReduce2

目标：实现将HDFS中的数据写入到HBase表中。 分步实现：

  1. 构建ReadFruitFromHDFSMapper于读取HDFS中的文件数据
代码语言：javascript复制
public class ReadFruitFromHDFSMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		//从HDFS中读取的数据
		String lineValue = value.toString();
		//读取出来的每行数据使用t进行分割，存于String数组
		String[] values = lineValue.split("t");
		
		//根据数据中值的含义取值
		String rowKey = values[0];
		String name = values[1];
		String color = values[2];
		
		//初始化rowKey
		ImmutableBytesWritable rowKeyWritable = new ImmutableBytesWritable(Bytes.toBytes(rowKey));
		
		//初始化put对象
		Put put = new Put(Bytes.toBytes(rowKey));
		
		//参数分别:列族、列、值  
        put.add(Bytes.toBytes("info"), Bytes.toBytes("name"),  Bytes.toBytes(name)); 
        put.add(Bytes.toBytes("info"), Bytes.toBytes("color"),  Bytes.toBytes(color)); 
        
        context.write(rowKeyWritable, put);
	}
}
  1. 构建WriteFruitMRFromTxtReducer类
代码语言：javascript复制
public class WriteFruitMRFromTxtReducer extends TableReducer<ImmutableBytesWritable, Put, NullWritable> {
	@Override
	protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
		//读出来的每一行数据写入到fruit_hdfs表中
		for(Put put: values){
			context.write(NullWritable.get(), put);
		}
	}
}
  1. 创建Txt2FruitRunner组装Job
代码语言：javascript复制
public int run(String[] args) throws Exception {
	//得到Configuration
	Configuration conf = this.getConf();
	
	//创建Job任务
	Job job = Job.getInstance(conf, this.getClass().getSimpleName());
	job.setJarByClass(Txt2FruitRunner.class);
	Path inPath = new Path("hdfs://hadoop102:9000/input_fruit/fruit.tsv");
	FileInputFormat.addInputPath(job, inPath);
	
	//设置Mapper
	job.setMapperClass(ReadFruitFromHDFSMapper.class);
	job.setMapOutputKeyClass(ImmutableBytesWritable.class);
	job.setMapOutputValueClass(Put.class);
	
	//设置Reducer
	TableMapReduceUtil.initTableReducerJob("fruit_mr", WriteFruitMRFromTxtReducer.class, job);
	
	//设置Reduce数量，最少1个
	job.setNumReduceTasks(1);
	
	boolean isSuccess = job.waitForCompletion(true);
	if(!isSuccess){
		throw new IOException("Job running with error");
	}
	
	return isSuccess ? 0 : 1;
}
  1. 调用执行Job
代码语言：javascript复制
public static void main(String[] args) throws Exception {
		Configuration conf = HBaseConfiguration.create();
	    int status = ToolRunner.run(conf, new Txt2FruitRunner(), args);
	    System.exit(status);
}
  1. 打包运行
代码语言：javascript复制
$ /opt/module/hadoop-2.7.2/bin/yarn jar hbase-0.0.1-SNAPSHOT.jar com.atguigu.hbase.mr2.Txt2FruitRunner

提示：运行任务前，如果待数据导入的表不存在，则需要提前创建之。 提示：maven打包命令：-P local clean package或-P dev clean package install（将第三方jar包一同打包，需要插件：maven-shade-plugin）

6.4 与Hive的集成

6.4.1 HBase与Hive的对比

  1. Hive (1) 数据仓库 Hive的本质其实就相当于将HDFS中已经存储的文件在Mysql中做了一个双射关系，以方便使用HQL去管理查询。 (2) 用于数据分析、清洗 Hive适用于离线的数据分析和清洗，延迟较高。 (3) 基于HDFS、MapReduce Hive存储的数据依旧在DataNode上，编写的HQL语句终将是转换为MapReduce代码执行。
  2. HBase (1) 数据库 是一种面向列存储的非关系型数据库。 (2) 用于存储结构化和非结构化的数据 适用于单表非关系型数据的存储，不适合做关联查询，类似JOIN等操作。 (3) 基于HDFS 数据持久化存储的体现形式是Hfile，存放于DataNode中，被ResionServer以region的形式进行管理。 (4) 延迟较低，接入在线业务使用 面对大量的企业数据，HBase可以直线单表大量数据的存储，同时提供了高效的数据访问速度。

6.4.2 HBase与Hive集成使用

尖叫提示：HBase与Hive的集成在最新的两个版本中无法兼容。所以，我们只能含着泪勇敢的重新编译：hive-hbase-handler-1.2.2.jar！！好气！！

环境准备 因为我们后续可能会在操作Hive的同时对HBase也会产生影响，所以Hive需要持有操作HBase的Jar，那么接下来拷贝Hive所依赖的Jar包（或者使用软连接的形式）。

代码语言：javascript复制
export HBASE_HOME=/opt/module/hbase
export HIVE_HOME=/opt/module/hive

ln -s $HBASE_HOME/lib/hbase-common-1.3.1.jar  $HIVE_HOME/lib/hbase-common-1.3.1.jar
ln -s $HBASE_HOME/lib/hbase-server-1.3.1.jar $HIVE_HOME/lib/hbase-server-1.3.1.jar
ln -s $HBASE_HOME/lib/hbase-client-1.3.1.jar $HIVE_HOME/lib/hbase-client-1.3.1.jar
ln -s $HBASE_HOME/lib/hbase-protocol-1.3.1.jar $HIVE_HOME/lib/hbase-protocol-1.3.1.jar
ln -s $HBASE_HOME/lib/hbase-it-1.3.1.jar $HIVE_HOME/lib/hbase-it-1.3.1.jar
ln -s $HBASE_HOME/lib/htrace-core-3.1.0-incubating.jar $HIVE_HOME/lib/htrace-core-3.1.0-incubating.jar
ln -s $HBASE_HOME/lib/hbase-hadoop2-compat-1.3.1.jar $HIVE_HOME/lib/hbase-hadoop2-compat-1.3.1.jar
ln -s $HBASE_HOME/lib/hbase-hadoop-compat-1.3.1.jar $HIVE_HOME/lib/hbase-hadoop-compat-1.3.1.jar

同时在hive-site.xml中修改zookeeper的属性，如下：

代码语言：javascript复制
<property>
  <name>hive.zookeeper.quorum</name>
  <value>hadoop102,hadoop103,hadoop104</value>
  <description>The list of ZooKeeper servers to talk to. This is only needed for read/write locks.</description>
</property>
<property>
  <name>hive.zookeeper.client.port</name>
  <value>2181</value>
  <description>The port of ZooKeeper servers to talk to. This is only needed for read/write locks.</description>
</property>
  1. 案例一 目标：建立Hive表，关联HBase表，插入数据到Hive表的同时能够影响HBase表。 分步实现： (1) 在Hive中创建表同时关联HBase
代码语言：javascript复制
CREATE TABLE hive_hbase_emp_table(
empno int,
ename string,
job string,
mgr int,
hiredate string,
sal double,
comm double,
deptno int)
STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,info:ename,info:job,info:mgr,info:hiredate,info:sal,info:comm,info:deptno")
TBLPROPERTIES ("hbase.table.name" = "hbase_emp_table");

提示：完成之后，可以分别进入Hive和HBase查看，都生成了对应的表

(2) 在Hive中创建临时中间表，用于load文件中的数据 提示：不能将数据直接load进Hive所关联HBase的那张表中

代码语言：javascript复制
CREATE TABLE emp(
empno int,
ename string,
job string,
mgr int,
hiredate string,
sal double,
comm double,
deptno int)
row format delimited fields terminated by 't';

(3) 向Hive中间表中load数据

代码语言：javascript复制
hive> load data local inpath '/home/admin/softwares/data/emp.txt' into table emp;

(4) 通过insert命令将中间表中的数据导入到Hive关联HBase的那张表中

代码语言：javascript复制
hive> insert into table hive_hbase_emp_table select * from emp;

(5) 查看Hive以及关联的HBase表中是否已经成功的同步插入了数据

代码语言：javascript复制
Hive：
hive> select * from hive_hbase_emp_table;
HBase：
hbase> scan ‘hbase_emp_table’
  1. 案例二 目标：在HBase中已经存储了某一张表hbase_emp_table，然后在Hive中创建一个外部表来关联HBase中的hbase_emp_table这张表，使之可以借助Hive来分析HBase这张表中的数据。
代码语言：javascript复制
注：该案例2紧跟案例1的脚步，所以完成此案例前，请先完成案例1。

分步实现： (1) 在Hive中创建外部表

代码语言：javascript复制
CREATE EXTERNAL TABLE relevance_hbase_emp(
empno int,
ename string,
job string,
mgr int,
hiredate string,
sal double,
comm double,
deptno int)
STORED BY 
'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
WITH SERDEPROPERTIES ("hbase.columns.mapping" = 
":key,info:ename,info:job,info:mgr,info:hiredate,info:sal,info:comm,info:deptno") 
TBLPROPERTIES ("hbase.table.name" = "hbase_emp_table");

(2) 关联后就可以使用Hive函数进行一些分析操作了

代码语言：javascript复制
hive (default)> select * from relevance_hbase_emp;
hbase TDSQLMySQL版 hive mapreduce 数据库

0 人点赞