hbase 分享笔记

package com.ruishenh.hbase.client.java;

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!");

    }

}

<?xml version="1.0"encoding="UTF-8"?>

<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>

<?xml version="1.0"encoding="UTF-8"?>

<configuration>

    <property>

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

        <value>localhost</value>

    </property>

    <property>

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

        <value>2181</value>

    </property>

</configuration>

ClassPathXmlApplicationContext classPathXmlApplicationContext = newClassPathXmlApplicationContext("hbase.xml");

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;

    }

});

package com.ruishenh.hbase.mr;

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>

        }

    }

}

package com.ruishenh.hbase.cp;

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_package = "com.ruishenh.hbase.cp.proto";

 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); 

} 

package com.ruishenh.hbase.cp;

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());

    }

}

package com.ruishenh.hbase.cp.proto;

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());

        }

}