七、HBASE-06
HBaseAPI操作
1、环境准备
? 新建项目后再pom.xml中添加依赖:
<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>
2、HBaseAPI
(1)获取Configuration对象
public static Configuration conf;
static{
conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "192.168.36.102");
conf.set("hbase.zookeeper.property.clientPort", "2181");
}
(2)判断表是否存在
public static boolean isTableExist(String tableName) throws
MasterNotRunningException,
ZooKeeperConnectionException, IOException{
HBaseAdmin admin = new HBaseAdmin(conf);
return admin.tableExists(tableName);
}
(3)创建表
public static void createTable(String tableName, String...
columnFamily) throws
MasterNotRunningException, ZooKeeperConnectionException,
IOException{
HBaseAdmin admin = new HBaseAdmin(conf);
if(isTableExist(tableName)){
System.out.println("表" + tableName + "已存在");
}else{
HTableDescriptor descriptor = new
HTableDescriptor(TableName.valueOf(tableName));
for(String cf : columnFamily){
descriptor.addFamily(new HColumnDescriptor(cf));
}
admin.createTable(descriptor);
System.out.println("表" + tableName + "创建成功!");
} }
(4)删除表
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 + "不存在!");
} }
(5)向表中插入数据
public static void addRowData(String tableName, String rowKey, String
columnFamily, String
column, String value) throws IOException{
HTable hTable = new HTable(conf, tableName);
Put put = new Put(Bytes.toBytes(rowKey));
put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column),
Bytes.toBytes(value));
hTable.put(put);
hTable.close();
System.out.println("插入数据成功");
}
(6)删除多行数据
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();
}
(7)获取所有数据
public static void getAllRows(String tableName) throws IOException{
HTable hTable = new HTable(conf, tableName);
Scan scan = new Scan();
ResultScanner resultScanner = hTable.getScanner(scan);
for(Result result : resultScanner){
Cell[] cells = result.rawCells();
for(Cell cell : cells){
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)));
} } }
(8)获取某一行数据
public static void getRow(String tableName, String rowKey) throws
IOException{
HTable table = new HTable(conf, tableName);
Get get = new Get(Bytes.toBytes(rowKey));
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());
} }
(9)获取某一行指定“列族:列”的数据
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)));
} }
3、MapReduce
? 通过 HBase 的相关 JavaAPI,我们可以实现伴随 HBase 操作的MapReduce 过程,比如使用MapReduce 将数据从本地文件系统导入到HBase 的表中,比如我们从 HBase 中读取一些原始数据后使用 MapReduce 做数据分析。
(1)官方HBase-MapReduce
? 1、查看HBase的MapReduce任务的执行
$ bin/hbase mapredcp
? 2、环境变量的导入
export HBASE_HOME=/opt/module/hbase-1.3.1
export HADOOP_HOME=/opt/module/hadoop-2.7.2
? 并在Hadoop-env.sh中配置:
export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/opt/module/hbase/lib/*
? 3、运行官方的MapReduce任务
$ /opt/module/hadoop-2.7.2/bin/yarn jar lib/hbase-server-1.3.1.jar
rowcounter student
#案例二:使用MapReduce将本地数据导入HBase
#在本地创建一个tsv格式的文件:fruit.tsv
1001 Apple Red
1002 Pear Yellow
1003 Pineapple Yellow
hbase(main):001:0> create 'fruit','info'
$ /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/
$ /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
hbase(main):001:0> scan ‘fruit’
(2)自定义HBase-MapReduce 1
? 目标:将fruit表中的一部分数据,通过MR迁入到fruit_mr表中。
? 分步实现:
? 1、构建ReadFruitMapper类,用于读取fruit表中的数据
package com.lyinl;
import java.io.IOException;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
public class ReadFruitMapper extends
TableMapper<ImmutableBytesWritable, Put> {
@Override
protected void map(ImmutableBytesWritable key, Result value,
Context context)
throws IOException, InterruptedException {
对象中。
Put put = new Put(key.get());
for(Cell cell: value.rawCells()){
if("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))){
if("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell))
)){
put.add(cell);
}else
if("color".equals(Bytes.toString(CellUtil.cloneQualifier(cell))))
{
put.add(cell);
} } }
context.write(key, put);
} }
? 2、构建WriteFruitMRReducer类,用于将读取到的fruit表中的数据写入到fruit_mr表中。
package com.lyinl.hbase_mr;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.NullWritable;
public class WriteFruitMRReducer extends
TableReducer<ImmutableBytesWritable, Put, NullWritable> {
@Override
protected void reduce(ImmutableBytesWritable key, Iterable<Put>
values, Context context)
throws IOException, InterruptedException {
for(Put put: values){
context.write(NullWritable.get(), put);
} } }
? 3、构建Fruit2FruitMRRunner extends Configured implements Tool用于组装运行Job任务
public int run(String[] args) throws Exception {
Configuration conf = this.getConf();
Job job = Job.getInstance(conf,
this.getClass().getSimpleName());
job.setJarByClass(Fruit2FruitMRRunner.class);
Scan scan = new Scan();
scan.setCacheBlocks(false);
scan.setCaching(500);
是老版本
TableMapReduceUtil.initTableMapperJob(
"fruit",
scan,
ReadFruitMapper.class,
ImmutableBytesWritable.class,
Put.class,
job
);
TableMapReduceUtil.initTableReducerJob("fruit_mr",
WriteFruitMRReducer.class, job);
job.setNumReduceTasks(1);
boolean isSuccess = job.waitForCompletion(true);
if(!isSuccess){
throw new IOException("Job running with error");
}
return isSuccess ? 0 : 1;
}
? 4、主函数中调用运行该Job任务
public static void main( String[] args ) throws Exception{
Configuration conf = HBaseConfiguration.create();
int status = ToolRunner.run(conf, new Fruit2FruitMRRunner(), args);
System.exit(status);
}
? 5、打包运行任务
$ /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)
(2)自定义HBase-MapReduce 2
? 目标:实现将HDFS中的数据写入到HBase表中。
? 分步实现:
? 1、构建ReadFruitFromHDFSMapper于读取HDFS中的文件数据
package com.lyinl;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class ReadFruitFromHDFSMapper extends Mapper<LongWritable,
Text, ImmutableBytesWritable, Put> {
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String lineValue = value.toString();
String[] values = lineValue.split("\t");
String rowKey = values[0];
String name = values[1];
String color = values[2];
ImmutableBytesWritable rowKeyWritable = new
ImmutableBytesWritable(Bytes.toBytes(rowKey));
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);
} }
? 2、构建WriteFruitMRFromTxtReducer类
package com.z.hbase.mr2;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.NullWritable;
public class WriteFruitMRFromTxtReducer extends
TableReducer<ImmutableBytesWritable, Put, NullWritable> {
@Override
protected void reduce(ImmutableBytesWritable key, Iterable<Put>
values, Context context) throws IOException, InterruptedException {
for(Put put: values){
context.write(NullWritable.get(), put);
} } }
? 3、创建Txt2FruitRunner组装Job
public int run(String[] args) throws Exception {
Configuration conf = this.getConf();
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);
job.setMapperClass(ReadFruitFromHDFSMapper.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Put.class);
TableMapReduceUtil.initTableReducerJob("fruit_mr",
WriteFruitMRFromTxtReducer.class, job);
job.setNumReduceTasks(1);
boolean isSuccess = job.waitForCompletion(true);
if(!isSuccess){
throw new IOException("Job running with error");
}
return isSuccess ? 0 : 1;
}
? 4、调用执行Job
public static void main(String[] args) throws Exception {
Configuration conf = HBaseConfiguration.create();
int status = ToolRunner.run(conf, new Txt2FruitRunner(),
args);
System.exit(status);
}
? 5、打包运行
$ /opt/module/hadoop-2.7.2/bin/yarn jar hbase-0.0.1-SNAPSHOT.jar
com.lyinl.hbase.mr2.Txt2FruitRunner
4、与Hive的集成
(1)HBase与Hive的对比
? 1、Hive
? a、数据仓库
? Hive的本质其实就相当于将HDFS中已经存储的文件在Mysql中做了一个双射关系,以方便使用HQL去管理查询。
? b、用于数据分析、清洗
? Hive使用用离线的数据分析和清洗,延迟较高
? c、基于HDFS、MapReduce
? Hive存储的数据依旧在DataNode上,编写的HQL语句中将是转换为MapRedue代码执行。
? 2、HBase
? a、数据库
? 是一种面向列存储的非关系型数据库。
? b、用于存储结构化的数据
? 适用于单表非关系型数据的存储,不适合做关联查询,类似JOIN等操作。
? 3、基于HDFS
? 数据持久化存储的体现是Hfile,存放于DataNode中,被ResionServer以region的形式进行管理。
? 4、延迟较低,接入在西安业务使用
? 面都大量的企业数据,HBase可以直线单表大量数据的存储,同时提供了高效的数据访问速度。
?
|