hbase 分享笔记

import com.google.common.io.Closeables;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.*;

import org.apache.hadoop.hbase.client.*;

import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HbaseClient {

// 声明静态配置

static
Configuration conf =
null;

static {

conf = HBaseConfiguration.create();

conf.set("hbase.zookeeper.quorum","localhost");

}

/*

* 创建表

*

* @tableName表名

*

* @family
列族列表

*/

public static voidcreatTable(String tableName, String[] family)

throws Exception {

HBaseAdmin admin = new HBaseAdmin(conf);

HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));

for (inti =0; i < family.length; i++) {

desc.addFamily(new HColumnDescriptor(family[i]).setMaxVersions(10));

}

if (admin.tableExists(tableName)) {

System.out.println("table Exists!");

} else {

admin.createTable(desc);

System.out.println("create table Success!");

}

Closeables.close(admin,false);

}

/**

* 获取表等于list命令

* @throwsException

*/

public static voidgetTables()

throws Exception {

HBaseAdmin admin = new HBaseAdmin(conf);

TableName[] tableNames = admin.listTableNames();

for (TableName tableName : tableNames) {

HTableDescriptor tableDescriptor = admin.getTableDescriptor(tableName);

System.out.println(tableName+"--"+tableDescriptor);

}

Closeables.close(admin,false);

}

/*

* 根据rwokey查询

*

* @rowKey rowKey

*

* @tableName表名

*/

public staticResult getResult(String tableName, String rowKey)

throws IOException {

Get get = new Get(Bytes.toBytes(rowKey));

HTable table = new HTable(conf, Bytes.toBytes(tableName));//获取表

Result result = table.get(get);

for (KeyValue kv : result.list()) {

System.out.println("family:"+ Bytes.toString(kv.getFamily()));

System.out

.println("qualifier:"+ Bytes.toString(kv.getQualifier()));

System.out.println("value:"+ Bytes.toString(kv.getValue()));

System.out.println("Timestamp:"+ kv.getTimestamp());

System.out.println("-------------------------------------------");

}

Closeables.close(table,false);

return result;

}

/*

* 遍历查询hbase表

*

* @tableName表名

*/

public static voidgetResultScann(String tableName)throwsIOException {

Scan scan = new Scan();

ResultScanner rs = null;

HTable table = new HTable(conf, Bytes.toBytes(tableName));

try {

rs = table.getScanner(scan);

for (Result r : rs) {

for (KeyValue kv : r.list()) {

System.out.println("row:"+ Bytes.toString(kv.getRow()));

System.out.println("family:"

+ Bytes.toString(kv.getFamily()));

System.out.println("qualifier:"

+ Bytes.toString(kv.getQualifier()));

System.out

.println("value:"+ Bytes.toString(kv.getValue()));

System.out.println("timestamp:"+ kv.getTimestamp());

System.out

.println("-------------------------------------------");

}

}

} finally {

rs.close();

}

}

/*

* 遍历查询hbase表

*

* @tableName表名

*/

public static voidgetResultScann(String tableName, String start_rowkey,

String stop_rowkey) throwsIOException {

Scan scan = new Scan();

scan.setStartRow(Bytes.toBytes(start_rowkey));

scan.setStopRow(Bytes.toBytes(stop_rowkey));

ResultScanner rs = null;

HTable table = new HTable(conf, Bytes.toBytes(tableName));

try {

rs = table.getScanner(scan);

for (Result r : rs) {

for (KeyValue kv : r.list()) {

System.out.println("row:"+ Bytes.toString(kv.getRow()));

System.out.println("family:"

+ Bytes.toString(kv.getFamily()));

System.out.println("qualifier:"

+ Bytes.toString(kv.getQualifier()));

System.out

.println("value:"+ Bytes.toString(kv.getValue()));

System.out.println("timestamp:"+ kv.getTimestamp());

System.out

.println("-------------------------------------------");

}

}

} finally {

Closeables.close(rs,false);

}

}

/*

* 查询表中的某一列

*

* @tableName表名

*

* @rowKey rowKey

*/

public static voidgetResultByColumn(String tableName, String rowKey,

String familyName, String columnName)throwsIOException {

HTable table = new HTable(conf, Bytes.toBytes(tableName));

Get get = new Get(Bytes.toBytes(rowKey));

get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));//获取指定列族和列修饰符对应的列

Result result = table.get(get);

System.out.println("查询表中的某一列:");

for (KeyValue kv : result.list()) {

System.out.println("family:"+ Bytes.toString(kv.getFamily()));

System.out

.println("qualifier:"+ Bytes.toString(kv.getQualifier()));

System.out.println("value:"+ Bytes.toString(kv.getValue()));

System.out.println("Timestamp:"+ kv.getTimestamp());

System.out.println("---------------");

}

System.out.println("-------------------------------------------");

Closeables.close(table,false);

}

/*

* 更新表中的某一列

*

* @tableName表名

*

* @rowKey rowKey

*

* @familyName列族名

*

* @columnName列名

*

* @value
更新后的值

*/

public static voidupdateTable(String tableName, String rowKey,

String familyName, String columnName, String value)

throws IOException {

HTable table = new HTable(conf, Bytes.toBytes(tableName));

Put put = new Put(Bytes.toBytes(rowKey));

put.add(Bytes.toBytes(familyName), Bytes.toBytes(columnName),

Bytes.toBytes(value));

table.put(put);

System.out.println("update table Success!");

Closeables.close(table,false);

}

/*

* 查询某列数据的多个版本

*

* @tableName表名

*

* @rowKey rowKey

*

* @familyName列族名

*

* @columnName列名

*/

public static voidgetResultByVersion(String tableName, String rowKey,

String familyName, String columnName)throwsIOException {

HTable table = new HTable(conf, Bytes.toBytes(tableName));

Get get = new Get(Bytes.toBytes(rowKey));

get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));

get.setMaxVersions(5);

Result result = table.get(get);

System.out.println("查询某列数据的多个版本");

for (KeyValue kv : result.list()) {

System.out.println("family:"+ Bytes.toString(kv.getFamily()));

System.out

.println("qualifier:"+ Bytes.toString(kv.getQualifier()));

System.out.println("value:"+ Bytes.toString(kv.getValue()));

System.out.println("Timestamp:"+ kv.getTimestamp());

System.out.println("--------------------");

}

System.out.println("-------------------------------------------");

Closeables.close(table,false);

}

/*

* 查询某列数据的时间磋范围

*

* @tableName
表名

*

* @rowKey rowKey

*

* @familyName
列族名

*

* @columnName
列名

*/

public static voidgetResultByTimeRange(String tableName, String rowKey,

String familyName, String columnName,longstart,longend)throwsIOException
{

HTable table = new HTable(conf, Bytes.toBytes(tableName));

Get get = new Get(Bytes.toBytes(rowKey));

get.setTimeRange(start, end);

get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));

Result result = table.get(get);

System.out.println("查询某列数据TimeRange");

System.out.println("start:"+start);

System.out.println("start:"+end);

for (KeyValue kv : result.list()) {

System.out.println("family:"+ Bytes.toString(kv.getFamily()));

System.out

.println("qualifier:"+ Bytes.toString(kv.getQualifier()));

System.out.println("value:"+ Bytes.toString(kv.getValue()));

System.out.println("Timestamp:"+ kv.getTimestamp());

System.out.println("--------------------");

}

System.out.println("-------------------------------------------");

Closeables.close(table,false);

}

/*

* 删除指定的列

*

* @tableName表名

*

* @rowKey rowKey

*

* @familyName列族名

*

* @columnName列名

*/

public static voiddeleteColumn(String tableName, String rowKey,

String falilyName, String columnName)
throws IOException {

HTable table = new HTable(conf, Bytes.toBytes(tableName));

Delete deleteColumn = new Delete(Bytes.toBytes(rowKey));

deleteColumn.deleteColumns(Bytes.toBytes(falilyName),

Bytes.toBytes(columnName));

table.delete(deleteColumn);

System.out.println(falilyName +":"+ columnName +"is deleted!");

Closeables.close(table,false);

}

/*

* 删除指定的列

*

* @tableName表名

*

* @rowKey rowKey

*/

public static voiddeleteAllColumn(String tableName, String rowKey)

throws IOException {

HTable table = new HTable(conf, Bytes.toBytes(tableName));

Delete deleteAll = new Delete(Bytes.toBytes(rowKey));

table.delete(deleteAll);

System.out.println("all columns are deleted!");

Closeables.close(table,false);

}

/*

* 删除表

*

* @tableName表名

*/

public static voiddeleteTable(String tableName)throwsIOException {

HBaseAdmin admin = new HBaseAdmin(conf);

admin.disableTable(tableName);

admin.deleteTable(tableName);

System.out.println(tableName +"is deleted!");

Closeables.close(admin,false);

}

public static void main(String[] args)throwsException {

String tableName = "hcr_tb";

String[] family = { "dcf","cfc"};

//创建表

creatTable(tableName, family);

//查看表

getTables();

//删除表

//        deleteTable(tableName);

//
为表添加数据

String[] column1 = { "name","age","tag"};

String[] value1 = {"侯哥","20","hadoop"};

String[] column2 = { "modifycount","searchcount"};

String[] value2 = { "3","5"};

long a2=System.currentTimeMillis();

addData("rowkey1",tableName,family[0], column1, value1);

addData("rowkey1", tableName, family[1],column2, value2);

addData("rowkey2", tableName, family[0], column1, value1);

addData("rowkey2", tableName, family[1], column2, value2);

// 遍历查询

getResultScann(tableName,
"rowkey1", "rowkey2");

// 根据row key范围遍历查询

getResultScann(tableName,
"rowkey4", "rowkey5");

// 查询

getResult(tableName,
"rowkey1");

// 查询某一列的值

getResultByColumn(tableName,
"rowkey1", "dcf","name");

long a=System.currentTimeMillis();

// 更新列

updateTable(tableName,
"rowkey1", "dcf","name","大师兄");

// 查询某一列的值

getResultByColumn(tableName,
"rowkey1", "dcf","name");

// 查询某列的多版本

getResultByVersion(tableName,"rowkey1","dcf","name");

//时间错范围查询

getResultByTimeRange(tableName,"rowkey1","dcf","name",a2,a);

// 删除一列

deleteColumn(tableName,
"rowkey1", "dcf","name");

// 删除所有列

deleteAllColumn(tableName,
"rowkey1");

// 删除表

deleteTable(tableName);

}

private static void addData(String rowKey, String tableName, String s, String[] column1, String[] value1)throwsIOException {

Put put = new Put(Bytes.toBytes(rowKey));//设置rowkey

HTable table = newHTable(conf, Bytes.toBytes(tableName));// HTabel负责跟记录相关的操作如增删改查等//

for
(int j =
0; j < column1.length; j++) {

put.add(Bytes.toBytes(s),

Bytes.toBytes(column1[j]), Bytes.toBytes(value1[j]));

}

table.put(put);

Closeables.close(table,false);

System.out.println("add data Success!");

}

}

<beansxmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:p="http://www.springframework.org/schema/p"

xmlns:hdp="http://www.springframework.org/schema/hadoop"

xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd

http://www.springframework.org/schema/hadoop http://www.springframework.org/schema/hadoop/spring-hadoop-2.0.xsd">

<hdp:configurationresources="classpath:/hbase-site.xml"/>

<hdp:hbase-configurationconfiguration-ref="hadoopConfiguration"/>

<beanid="hbaseTemplate"class="org.springframework.data.hadoop.hbase.HbaseTemplate"p:configuration-ref="hbaseConfiguration"/>

</beans>

<configuration>

<property>

<name>hbase.zookeeper.quorum</name>

<value>localhost</value>

</property>

<property>

<name>hbase.zookeeper.property.clientPort</name>

<value>2181</value>

</property>

</configuration>

HbaseTemplate bean = classPathXmlApplicationContext.getBean(HbaseTemplate.class);

bean.execute("t1", newTableCallback<Object>() {

@Override

public
Object doInTable(HTableInterface table) throwsThrowable {

HTableDescriptor tableDescriptor = table.getTableDescriptor();

System.out.println(tableDescriptor);

return null;

}

});

bean.get("t1", "r1",newRowMapper<Object>() {

@Override

public
Object mapRow(Result result, int
rowNum) throws Exception {

for (KeyValue kv : result.list()) {

System.out.println("family:"+ Bytes.toString(kv.getFamily()));

System.out.println("qualifier:"+ Bytes.toString(kv.getQualifier()));

System.out.println("value:"+ Bytes.toString(kv.getValue()));

System.out.println("Timestamp:"+ kv.getTimestamp());

System.out.println("-------------------------------------------");

}

return null;

}

});

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

import java.util.StringTokenizer;

/**

* Created by hcr on 16-6-11.

*/

public class MapReduceTest {

public static void main(String[] args)throwsIOException, ClassNotFoundException, InterruptedException {

Configuration conf = HBaseConfiguration.create();

conf.addResource("yarn-site.xml");

conf.addResource("mapred-site.xml");

conf.set("tmpjars","file:/app/project/myself/hbasetTest/target/hbaseTest-1.0-SNAPSHOT.jar");

Job job =Job.getInstance(conf);

job.setJobName("hcr-job name");

job.setJarByClass(MapReduceTest.class);

Scan scan = new Scan();

TableMapReduceUtil.initTableMapperJob("hcr_test", scan, WordCountHbaseMapper.class, Text.class,
IntWritable.class, job);

TableMapReduceUtil.initTableReducerJob("hcr_test2", WordCountHbaseReducer.class, job);

boolean b = job.waitForCompletion(true);

System.out.println("run:"+b);

}

public static class WordCountHbaseMapperextends

TableMapper<Text, IntWritable> {

private final static IntWritableone=newIntWritable(1);

private Text
word = newText();

@Override

protected voidmap(ImmutableBytesWritable key, Result value, Context context)throwsIOException, InterruptedException
{

StringTokenizer itr = new
StringTokenizer(value.toString());

System.out.println("key:"+newString(key.get()));

System.out.println("value:"+value);

while (itr.hasMoreTokens()) {

word.set(itr.nextToken());

context.write(word,one);//输出<key,value>为<word,one>

}

}

}

public static class WordCountHbaseReducerextends

TableReducer<Text, IntWritable, ImmutableBytesWritable> {

@Override

protected voidreduce(Text key, Iterable<IntWritable> values, Context context)throwsIOException, InterruptedException
{

int sum =
0;

for (IntWritable val : values) {//遍历求和

sum += val.get();

}

Put put = new Put(key.getBytes());//put实例化，每一个词存一行

//列族为content,列修饰符为count，列值为数目

put.add(Bytes.toBytes("cf"), Bytes.toBytes("count"), Bytes.toBytes(String.valueOf(sum)));

context.write(new ImmutableBytesWritable(key.getBytes()), put);//输出求和后的<key,value>

}

}

}

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.Cell;

import org.apache.hadoop.hbase.client.Durability;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;

import org.apache.hadoop.hbase.coprocessor.ObserverContext;

import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;

import org.apache.hadoop.hbase.regionserver.wal.WALEdit;

import java.io.IOException;

import java.util.Iterator;

import java.util.List;

/**

* Created by hcr on 16-6-13.

*/

public class MyTestCoprocessorextendsBaseRegionObserver {

@Override

public voidprePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability)throwsIOException
{

Configuration conf = new Configuration();

conf.set("hbase.zookeeper.quorum","localhost");

String colName = "url";

HTable table = new HTable(conf,"idx_tb");

List<Cell> kv = put.get("cf".getBytes(), colName.getBytes());

Iterator<Cell> kvItor = kv.iterator();

while (kvItor.hasNext()) {

Cell tmp = kvItor.next();

Put indexPut = new Put(tmp.getValue());

indexPut.add("cf".getBytes(),colName.getBytes(),

tmp.getRow());

table.put(indexPut);

}

table.close();

}

}

option java_outer_classname = "CounterService";

option java_generic_services = true;

option java_generate_equals_and_hash = true;

option optimize_for = SPEED;

message CountReq {

required string condition = 1;

}

message CountRes {

required int64 ret = 1;

}

service Counter{

rpc count(CountReq)

returns (CountRes);

}

import com.google.common.io.Closeables;

import com.google.protobuf.RpcCallback;

import com.google.protobuf.RpcController;

import com.google.protobuf.Service;

import com.ruishenh.hbase.cp.proto.CounterService;

import org.apache.hadoop.hbase.Cell;

import org.apache.hadoop.hbase.Coprocessor;

import org.apache.hadoop.hbase.CoprocessorEnvironment;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.coprocessor.CoprocessorException;

import org.apache.hadoop.hbase.coprocessor.CoprocessorService;

import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;

import org.apache.hadoop.hbase.regionserver.RegionScanner;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

/**

* Created by hcr on 16-6-14.

*/

public class MyTestCoprocessorEndPointextendsCounterService.Counter

implements Coprocessor, CoprocessorService {

Logger LOGGER =LoggerFactory.getLogger(MyTestCoprocessorEndPoint.class);

private RegionCoprocessorEnvironmentenv;

@Override

public voidstart(CoprocessorEnvironment coprocessorEnvironment)throwsIOException {

if (coprocessorEnvironmentinstanceofRegionCoprocessorEnvironment)

this.env= (RegionCoprocessorEnvironment) coprocessorEnvironment;

else throw new CoprocessorException("Must be loaded on a table region!!");

}

@Override

public voidstop(CoprocessorEnvironment env)throwsIOException {

}

@Override

public
Service getService() {

return this;

}

@Override

public voidcount(RpcController controller, CounterService.CountReq request, RpcCallback<CounterService.CountRes> done) {

RegionScanner scanner = null;

CounterService.CountRes.Builder respBuilder = CounterService.CountRes.newBuilder();

long count =
0;

try {

Scan scan = new Scan();

scan.setMaxVersions(1);

scanner = env.getRegion().getScanner(scan);

List<Cell> list = new
ArrayList<Cell>();

while (scanner.next(list))

count += 1;

respBuilder.setRet(count);

} catch (IOException e) {

LOGGER.error("io error",e);

} finally {

try {

Closeables.close(scanner,false);

} catch (IOException e) {

LOGGER.error("close io error",e);

}

}

done.run(respBuilder.build());

}

}

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.coprocessor.Batch;

import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;

import org.apache.hadoop.hbase.ipc.ServerRpcController;

import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

import java.util.concurrent.atomic.AtomicLong;

/**

* Created by hcr on 16-6-14.

*/

public class EndPointTest {

public static void main(String[] args)throwsThrowable {

Configuration conf = HBaseConfiguration.create();

conf.set("hbase.zookeeper.quorum","localhost");

HTable table = new HTable(conf,"hcr_test");

final CounterService.CountReq req = CounterService.CountReq.newBuilder().setCondition("xxxx").build();

final AtomicLong ret =newAtomicLong();

table.coprocessorService(CounterService.Counter.class,null,null,newBatch.Call<CounterService.Counter,
Long>() {

@Override

publicLong call(CounterService.Counter instance)throwsIOException {

ServerRpcController controller = newServerRpcController();

BlockingRpcCallback<CounterService.CountRes> rpc =
new BlockingRpcCallback<CounterService.CountRes>();

instance.count(controller, req, rpc);

CounterService.CountRes resp = rpc.get();

return resp.getRet();

}

}, new Batch.Callback<Long>() {

@Override

public voidupdate(byte[] region,byte[] row, Long result) {

ret.getAndAdd(result);

System.out.println(Bytes.toString(row)+": "+result);

}

});

System.out.println("lines: "+ ret.get());

}

}