HBase设计与开发

HBase设计与开发

@(HBase)

适合HBase应用的场景

基本概念

行健：是hbase表自带的，每个行健对应一条数据。

列族：是创建表时指定的，为列的集合，每个列族作为一个文件单独存储，存储的数据都是字节数组，其中的数据可以有很多，通过时间戳来区分。

物理模型：整个hbase表会拆分为多个region，每个region记录着行健的起始点保存在不同的节点上，查询时就是对各个节点的并行查询，当region很大时使用.META表存储各个region的起始点，-ROOT又可以存储.META的起始点。

rowkey的设计原则：长度原则、相邻原则，各个列簇数据平衡，创建表的时候设置表放入regionserver缓存中，避免自动增长和时间，使用字节数组代替string，最大长度64kb，最好16字节以内。

列族的设计原则：尽可能少（按照列族进行存储，按照region进行读取，不必要的io操作），经常和不经常使用的两类数据放入不同列族中，列族名字尽可能短，注意列族的数量级，避免数据倾斜。

rowkey设计可以参考：http://blog.chedushi.com/archives/9720

HBase开发

写入HBase

maper程序不需要改动，还是继承Mapper类读取文件。

//reducer要extends Tablereducer，使用方式和reducer一样，泛型的最后一个参数是输出的key类型，value类型固定为Put
public class ExtractReducer extends TableReducer<Text, Text, ImmutableBytesWritable> {
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) {
//Put对象new的时候参数为Bytes.toBytes(行健的值)，一个put就是一行
Put put = new Put(key.getBytes());
for (Text v : values) {
String[] vs = v.toString().split("\t");
for (int i = 1; i < vs.length; i++) {
//put.add(Bytes.toBytes(列簇),Bytes.toBytes(列),Bytes.toBytes(值));
put.add(Bytes.toBytes("trade"), Bytes.toBytes(""), Bytes.toBytes(vs[i]));
}
}
try {
context.write(new ImmutableBytesWritable(key.getBytes()), put);
} catch (IOException | InterruptedException e) {
e.printStackTrace();
}
}
}

提交mr程序的main方法和普通的mr程序差不多，只是conf中要多加几个hbase的设置：

conf.set("hbase.rootdir","/hbase");
conf.set("hbase.zookeeper.quorum","zookeeper地址");
conf.set("dfs.socket.timeout", "18000");

main程序示例：

public class ExtractDriver extends BaseDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Job job = new Job(config, "Extract");
job.setJarByClass(ExtractDriver.class);
FileInputFormat.setInputPaths(job, "/oracle/TYZXJS/T_CUSTOMER_AGREEMENT");
job.setInputFormatClass(TextInputFormat.class);

job.setMapperClass(ExtractMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
//设置reduce类和操作的hbase表名
TableMapReduceUtil.initTableReducerJob("CUSTOMER", ExtractReducer.class, job);

job.waitForCompletion(true);
}
}

读取hbase数据

map程序示例：

//需要使用到TableMapper这个类
public static class MyMapper extends TableMapper<Text, IntWritable>  {

private final IntWritable ONE = new IntWritable(1);
private Text text = new Text();

public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
String ip = Bytes.toString(row.get()).split("-")[0];
//列族，列名
String url = new String(value.getValue(Bytes.toBytes("info"), Bytes.toBytes("url")));
text.set(ip+"&"+url);
context.write(text, ONE);
}
}

main函数示例：

public static void main(String[] args) {
try{
Configuration config = HBaseConfiguration.create();
Job job = new Job(config,"ExampleSummary");
job.setJarByClass(MapReduce.class);
//设置scan读取hbase
Scan scan = new Scan();
scan.setCaching(500);
scan.setCacheBlocks(false);
TableMapReduceUtil.initTableMapperJob(
"access-log",        // 读取的表
scan,               // scan对象
MyMapper.class,     // mapper类
Text.class,         // mapper输出key类型
IntWritable.class,  // mapper输出value类型
job);
job.setNumReduceTasks(1);

//reduce代码和设置不变

job.waitForCompletion(true);
} catch(Exception e){
e.printStackTrace();
}
}

读写hbase示例

此处代码来自CSDN

public class ExampleTotalMapReduce{
public static void main(String[] args) {
try{
Configuration config = HBaseConfiguration.create();
Job job = new Job(config,"ExampleSummary");
job.setJarByClass(ExampleTotalMapReduce.class);     // class that contains mapper and reducer

Scan scan = new Scan();
scan.setCaching(500);        // 1 is the default in Scan, which will be bad for MapReduce jobs
scan.setCacheBlocks(false);  // don't set to true for MR jobs
// set other scan attrs
//scan.addColumn(family, qualifier);
TableMapReduceUtil.initTableMapperJob(
"access-log",        // input table
scan,               // Scan instance to control CF and attribute selection
MyMapper.class,     // mapper class
Text.class,         // mapper output key
IntWritable.class,  // mapper output value
job);
TableMapReduceUtil.initTableReducerJob(
"total-access",        // output table
MyTableReducer.class,    // reducer class
job);
job.setNumReduceTasks(1);   // at least one, adjust as required

boolean b = job.waitForCompletion(true);
if (!b) {
throw new IOException("error with job!");
}
} catch(Exception e){
e.printStackTrace();
}
}

public static class MyMapper extends TableMapper<Text, IntWritable>  {

private final IntWritable ONE = new IntWritable(1);
private Text text = new Text();

public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
String ip = Bytes.toString(row.get()).split("-")[0];
String url = new String(value.getValue(Bytes.toBytes("info"), Bytes.toBytes("url")));
text.set(ip+"&"+url);
context.write(text, ONE);
}
}

public static class MyTableReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable>  {
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}

Put put = new Put(key.getBytes());
put.add(Bytes.toBytes("info"), Bytes.toBytes("count"), Bytes.toBytes(String.valueOf(sum)));

context.write(null, put);
}
}
}

操作HBase过程中遇到的异常信息

1、hbase测试程序运行失败，情况一。

症状：提示zookeeper无法连接。

**问题所在：**zk地址错误。

解决方法：使用正确的zk地址。

2、hbase测试程序运行失败，情况二

**症状：**java程序访问hbase 卡住不动 不报异常。

问题插排：使用shell创建表失败，转异常3。

解决方式：异常3解决之后即可正常运行。

3、使用hbase shell创建表失败。

症状：提示ERROR: java.io.IOException: Table Namespace Manager not ready yet, try again later。

**问题插排：**hbase集群没有启动，转异常4。

解决方式：异常4解决之后正常启动集群即可。

4、hbase集群无法正常启动。

**症状：**hbase无法正常启动，子节点进程自动退出。

问题插排：查看节点日志发现异常，集群时间没有同步，待解决。

日志如下： org.apache.hadoop.hbase.ClockOutOfSyncException: org.apache.hadoop.hbase.ClockOutOfSyncException: Server zx-hadoop-210-27,60020,1451550407914 has been rejected; Reported time is too far out of sync with master. Time difference of 158391ms > max allowed of 30000ms

解决方式：同步集群时间。

作者：@小黑