[How to] 使用HBase协处理器---Endpoint客户端代码的实现

Posted 小马过溪

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了[How to] 使用HBase协处理器---Endpoint客户端代码的实现相关的知识,希望对你有一定的参考价值。

1.简介

  不同于Observer协处理器,EndPoint由于需要同region进行rpc服务的通信,以及客户端出数据的归并,需要自行实现客户端代码。

基于[How to] 使用HBase协处理器---Endpoint服务端的实现这篇文章,我们继续实现其客户端代码。

 

2.客户端代码实现方式介绍

  目前基于HBase1.0.0的版本客户端一共可以基于以下五个API来实现:

  1. Table.coprocessorService(byte[])

基于单个region的服务请求,参数为rowKey值,被请求的对象为此rowKey值参数所在的region

  2. Table.coprocessorService(Class, byte[], byte[],Batch.Call) 

基于rowKey范围的服务请求,第二和第三个参数为startKey和stopKey,被请求的region为rowkey范围内的region。
客户端并行的通过Class类型获取服务类型,通过Batch.Call调用服务接口,然后在客户端进行返回值的结果归并计算。

  3. Table.coprocessorService(Class, byte[], byte[], Batch.Call, Batch.Callback) 

与上述2类似,只是使用到了Batch.Callback方式,具体差异见后文分析。

  4.  Table.batchCoprocessorService(MethodDescriptor methodDescriptor, Message request, byte[] startKey, byte[] endKey, R responsePrototype)

批处理方式获取值,他将在regionserver上开启代理,代理将会调用当前regionserver上的目标region的服务接口。最后代理将归并这些值,然后返回给客户端。

 

  5.  Table.batchCoprocessorService(MethodDescriptor methodDescriptor, Message request, byte[] startKey, byte[] endKey, R responsePrototype, Callback<R> callback)

与4方法类似,具体差异见后文分析

 

3. 测试数据的准备

  在coprocessor_table表中我们插入如下数据:

hbase(main):038:0> put \'coprocessor_table\',\'row1\',\'F:A\',\'1\'
0 row(s) in 0.0100 seconds

hbase(main):039:0> put \'coprocessor_table\',\'row2\',\'F:A\',\'2\'
0 row(s) in 0.0050 seconds

hbase(main):040:0> put \'coprocessor_table\',\'row3\',\'F:A\',\'3\'
0 row(s) in 0.0070 seconds

hbase(main):041:0> put \'coprocessor_table\',\'row4\',\'F:A\',\'4\'
0 row(s) in 0.0050 seconds

hbase(main):042:0> put \'coprocessor_table\',\'row5\',\'F:A\',\'5\'
0 row(s) in 0.0060 seconds

hbase(main):006:0> put \'coprocessor_table\',\'row6\',\'F:A\',\'6\'
0 row(s) in 0.1370 seconds

hbase(main):013:0> scan \'coprocessor_table\'
ROW                   COLUMN+CELL                                               
 row1                 column=F:A, timestamp=1469853041043, value=1              
 row1                 column=F:B, timestamp=1469853041043, value=1              
 row2                 column=F:A, timestamp=1469853048597, value=2              
 row2                 column=F:B, timestamp=1469853048597, value=2              
 row3                 column=F:A, timestamp=1469853056871, value=3              
 row3                 column=F:B, timestamp=1469853056871, value=3              
 row4                 column=F:A, timestamp=1469853062544, value=4              
 row4                 column=F:B, timestamp=1469853062544, value=4              
 row5                 column=F:A, timestamp=1469853069914, value=5              
 row5                 column=F:B, timestamp=1469853069914, value=5              
 row6                 column=F:A, timestamp=1469859595584, value=6              
 row6                 column=F:B, timestamp=1469859595584, value=6 

 

 由于受到Observer协处理器的影响B列也会被插入值。

 为了后续测试的需要,我们将此表的region按照如下进行切割:

    

 

4. Table.coprocessorService(byte[])方式客户端代码实现

   其原理为通过参数的rowKey值进行region的定位,通过向此region请求服务进行计算,计算的数据范围仅限此region。

          

  其实现方法,注解见代码注解。

    /**
     * 通过CoprocessorRpcChannel coprocessorService(byte[] row); 请求单region服务
     * 
     * 客户端通过rowKey的指定,指向rowKey所在的region进行服务请求,所以从数据上来说只有这个region所包含的数据范围
     * 另外由于只向单个region请求服务,所以在客户端也没有必要在做归并操作。
     * 
     * @param config
     * @param tableName
     * @param rowkey
     * @param type
     * @param famillyName
     * @param columnName
     * @return
     * @throws IOException
     */
    private static long singleRegionStatistics(Configuration config,
            String tableName, String rowkey, String type, String famillyName,
            String columnName) throws IOException {
        long result = 0;
        Table table = null;
        Connection connection = null;
        try {
            connection = ConnectionFactory.createConnection(config);
            table = connection.getTable(TableName.valueOf(tableName));

            // 每一个region都加载了Endpoint协处理器,换句话说每一个region都能提供rpc的service服务,首先确定调用的范围
            // 这里只通过一个rowkey来确定,不管在此表中此rowkey是否存在,只要某个region的范围包含了这个rowkey,则这个region就为客户端提供服务
            CoprocessorRpcChannel channel = table.coprocessorService(rowkey
                    .getBytes());

            // 因为在region上可能会有很多不同rpcservice,所以必须确定你需要哪一个service
            MyStatisticsInterface.myStatisticsService.BlockingInterface service = MyStatisticsInterface.myStatisticsService
                    .newBlockingStub(channel);

            // 构建参数,设置 RPC 入口参数
            MyStatisticsInterface.getStatisticsRequest.Builder request = MyStatisticsInterface.getStatisticsRequest
                    .newBuilder();
            request.setType(type);
            if (null != famillyName) {
                request.setFamillyName(famillyName);
            }

            if (null != columnName) {
                request.setColumnName(columnName);
            }

            // 调用 RPC
            MyStatisticsInterface.getStatisticsResponse ret = service
                    .getStatisticsResult(null, request.build());

            // 解析结果,由于只向一个region请求服务,所以在客户端也就不存在去归并的操作
            result = ret.getResult();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != table) {
                table.close();
            }

            if (null != connection) {
                connection.close();
            }
        }
        return result;
    }

 

  测试代码:

        // 通过CoprocessorRpcChannel coprocessorService(byte[] row)
        // 请求单个region的rpc服务
        System.out.println("singleRegionStatistics COUNT = "
                + singleRegionStatistics(config, "coprocessor_table", "row2",
                        "COUNT", null, null));
        System.out.println("singleRegionStatistics SUM = "
                + singleRegionStatistics(config, "coprocessor_table", "row2",
                        "SUM", "F", "A"));

 

    测试结果:

singleRegionStatistics COUNT = 2
singleRegionStatistics SUM = 3

 结果分析: 

  因为我们指定的rowkey是row2,而这个rowkey是在第一个region上的,根据数据插入可知这个上面的数据如下(row3作为endkey并不属于当前region)。

  所以其一共有两行数据,对于A列的加和为 1+2=3,结果正确。

 row1                 column=F:A, timestamp=1469853041043, value=1              
 row1                 column=F:B, timestamp=1469853041043, value=1              
 row2                 column=F:A, timestamp=1469853048597, value=2              
 row2                 column=F:B, timestamp=1469853048597, value=2  

 

 

5. Table.coprocessorService(Class, byte[], byte[],Batch.Call) 方式客户端代码实现

  通过Table.coprocessorService(byte[])方式可以直接连到一个region,但是如果我们需要进行多个region的数据查询呢,甚至我们需要进行全表的数据查询呢?

我们可以简单的查找目标region的startKey然后在分别取调用Table.coprocessorService(byte[]),得到的结果值进行归并,这也可以,但是从效率和代码量维护性上来说都不是很好。HBase针对这种情况开放了Table.coprocessorService(Class, byte[], byte[],Batch.Call)这个API。

  在这个API参数列表中,

    Class代表所需要请求的服务,当前是我们定义在proto中的myStatisticsService服务。

    byte[], byte[]参数指明了startRowKey和endRowKey,当都为null的时候即进行全表的全region的数据计算。

    Batch.Call:需要自定义,API会根据如上参数信息并行的连接各个region,来执行这个参数中定义的call方法来执行接口方法的调用查询 

此API会将各个region上的信息放入Map<regionname,result>的结构中,搜集所有的结果后返回给调用者,调用者在循环进行归并计算。

其基本的运行原理如图:

         

其实现方法如下,具体解释见代码注解。(照葫芦画瓢即可)

    /**
     * 通过Table.coprocessorService(Class, byte[], byte[],Batch.Call) 请求多region服务
     * 
     * 在这个API参数列表中,
   * Class代表所需要请求的服务,当前是我们定义在proto中的myStatisticsService服务。
   * byte[], byte[]参数指明了startRowKey和endRowKey,当都为null的时候即进行全表的全region的数据计算。
   * Batch.Call:需要自定义,API会根据如上参数信息并行的连接各个region,来执行这个参数中定义的call方法来执行接口方法的调用查询 
     * 此API会将各个region上的信息放入Map<regionname,result>的结构中,搜集所有的结果后返回给调用者,调用者在循环进行归并计算。
     * 
     * @param config
     * @param tableName
     * @param startRowkey
     * @param endRowkey
     * @param type
     * @param famillyName
     * @param columnName
     * @return
     * @throws Throwable
     */
    private static long multipleRegionsStatistics(Configuration config,
            String tableName, String startRowkey, String endRowkey,
            final String type, final String famillyName, final String columnName)
            throws Throwable {
        long result = 0;
        Table table = null;
        Connection connection = null;
        
        // 返回值接收,Map<region名称,计算结果>
        Map<byte[], getStatisticsResponse> results = null;
        
        try {
            connection = ConnectionFactory.createConnection(config);
            table = connection.getTable(TableName.valueOf(tableName));

            // 第四个参数是接口类 Batch.Call。它定义了如何调用协处理器,用户通过重载该接口的 call() 方法来实现客户端的逻辑。在 call() 方法内,可以调用 RPC,并对返回值进行任意处理。
            Batch.Call<myStatisticsService, getStatisticsResponse> callable = new Batch.Call<myStatisticsService, getStatisticsResponse>() {
                ServerRpcController controller = new ServerRpcController();
                
                // 定义返回
                BlockingRpcCallback<getStatisticsResponse> rpcCallback = new BlockingRpcCallback<getStatisticsResponse>();

                // 下面重载 call 方法,API会连接到region后会运行call方法来执行服务的请求
                @Override
                public getStatisticsResponse call(myStatisticsService instance)
                        throws IOException {
                    // Server 端会进行慢速的遍历 region 的方法进行统计
                    MyStatisticsInterface.getStatisticsRequest.Builder request = MyStatisticsInterface.getStatisticsRequest
                            .newBuilder();
                    request.setType(type);
                    if (null != famillyName) {
                        request.setFamillyName(famillyName);
                    }

                    if (null != columnName) {
                        request.setColumnName(columnName);
                    }
                    // RPC 接口方法调用
                    instance.getStatisticsResult(controller, request.build(),
                            rpcCallback);
                    // 直接返回结果,即该 Region 的 计算结果
                    return rpcCallback.get();
                    
                }
            };
            
            /**
             * 通过Table.coprocessorService(Class, byte[], byte[],Batch.Call) 请求多region服务
             * 
             * 在这个API参数列表中,
           * Class代表所需要请求的服务,当前是我们定义在proto中的myStatisticsService服务。
           * byte[], byte[]参数指明了startRowKey和endRowKey,当都为null的时候即进行全表的全region的数据计算。
           * Batch.Call:需要自定义,API会根据如上参数信息并行的连接各个region,来执行这个参数中定义的call方法来执行接口方法的调用查询 
             * 此API会将各个region上的信息放入Map<regionname,result>的结构中,搜集所有的结果后返回给调用者,调用者在循环进行归并计算。
             */
            results = table.coprocessorService(myStatisticsService.class,
                    Bytes.toBytes(startRowkey), Bytes.toBytes(endRowkey),
                    callable);
            
            // 取得结果值后循环将结果合并,即得到最终的结果
            Collection<getStatisticsResponse> resultsc = results.values();
            for (getStatisticsResponse r : resultsc) {
                result += r.getResult();
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != table) {
                table.close();
            }

            if (null != connection) {
                connection.close();
            }
        }
        return result;
    }

 

测试代码:

        // 通过Table.coprocessorService(Class, byte[], byte[],Batch.Call)
        // 请求多个region的rpc服务
        // 计算指定rowkey范围所在的region上的数据行数
        System.out.println("multipleRegionsStatistics COUNT = "
                + multipleRegionsStatistics(config, "coprocessor_table", "row1","row3",
                        "COUNT", null, null));
        // 计算指定rowkey范围所在的region上的F列族下的A列的值得求和
        System.out.println("multipleRegionsStatistics SUM = "
                + multipleRegionsStatistics(config, "coprocessor_table", "row1","row3",
                        "SUM", "F", "A"));

 

解释结果:

multipleRegionsStatistics COUNT = 4
multipleRegionsStatistics SUM = 10

 

 结果分析:

  如上API中都指定的开始key为row1,结束key为3,根据region的key分布可知其包含两个rengion,这个两个region根据key分布可知其key范围为:[null,row5),

根据之前插入记录可知为如下数据:

 row1                 column=F:A, timestamp=1469853041043, value=1              
 row1                 column=F:B, timestamp=1469853041043, value=1              
 row2                 column=F:A, timestamp=1469853048597, value=2              
 row2                 column=F:B, timestamp=1469853048597, value=2              
 row3                 column=F:A, timestamp=1469853056871, value=3              
 row3                 column=F:B, timestamp=1469853056871, value=3              
 row4                 column=F:A, timestamp=1469853062544, value=4              
 row4                 column=F:B, timestamp=1469853062544, value=4  

 

  所以行数计算结果为4,sum计算结果为1+2+3+4=10,结果正确。

 

6. Table.coprocessorService(Class, byte[], byte[], Batch.Call, Batch.Callback) 客户端代码实现

   Table.coprocessorService(Class, byte[], byte[],Batch.Call)方式为我们提供了多region的调用实现,其这个API会搜集各个region上服务的返回数据然后加到Map中并返回,这是他的默认行为,Table.coprocessorService(Class, byte[], byte[], Batch.Call, Batch.Callback)可以让我们有机会通过实现Callback来改变这样的数据搜集行为,比如我们可以不放到Map中,直接进行加和,加和完毕后直接返回结果值即可,调用者也不必劳神去循环取得进行加和处理了,系统也不必额外的去为数据集进行内存开销。

  下面是这个API的运行原理图:

        

 

  其实现方法如下,具体解释见代码注解。(照葫芦画瓢即可)

    /**
     * 通过Table.coprocessorService(Class, byte[], byte[], Batch.Call,
     * Batch.Callback) 请求多region服务
     * 
     * 在这个API参数列表中,    * Class代表所需要请求的服务,当前是我们定义在proto中的myStatisticsService服务。
     *    * byte[],
     * byte[]参数指明了startRowKey和endRowKey,当都为null的时候即进行全表的全region的数据计算。
     *    * Batch.
     * Call:需要自定义,API会根据如上参数信息并行的连接各个region,来执行这个参数中定义的call方法来执行接口方法的调用查询 
     * 此API会将各个region上的信结果信息放入Map通过Callback定义进行处理,处理完所有所有的结果后返回给调用者。
     * 
     * @param config
     * @param tableName
     * @param startRowkey
     * @param endRowkey
     * @param type
     * @param famillyName
     * @param columnName
     * @return
     * @throws Throwable
     */
    private static long multipleRegionsCallBackStatistics(Configuration config,
            String tableName, String startRowkey, String endRowkey,
            final String type, final String famillyName, final String columnName)
            throws Throwable {
        final AtomicLong atoResult = new AtomicLong();
        Table table = null;
        Connection connection = null;

        try {
            connection = ConnectionFactory.createConnection(config);
            table = connection.getTable(TableName.valueOf(tableName));

            // 第四个参数是接口类 Batch.Call。它定义了如何调用协处理器,用户通过重载该接口的 call() 方法来实现客户端的逻辑。在
            // call() 方法内,可以调用 RPC,并对返回值进行任意处理。
            Batch.Call<myStatisticsService, getStatisticsResponse> callable = new Batch.Call<myStatisticsService, getStatisticsResponse>() {
                ServerRpcController controller = new ServerRpcController();

                // 定义返回
                BlockingRpcCallback<getStatisticsResponse> rpcCallback = new BlockingRpcCallback<getStatisticsResponse>();

                // 下面重载 call 方法,API会连接到region后会运行call方法来执行服务的请求
                @Override
                public getStatisticsResponse call(myStatisticsService instance)
                        throws IOException {
                    // Server 端会进行慢速的遍历 region 的方法进行统计
                    MyStatisticsInterface.getStatisticsRequest.Builder request = MyStatisticsInterface.getStatisticsRequest
                            .newBuilder();
                    request.setType(type);
                    if (null != famillyName) {
                        request.setFamillyName(famillyName);
                    }

                    if (null != columnName) {
                        request.setColumnName(columnName);
                    }
                    // RPC 接口方法调用
                    instance.getStatisticsResult(controller, request.build(),
                            rpcCallback);
                    // 直接返回结果,即该 Region 的 计算结果
                    return rpcCallback.get();

                }
            };

            // 定义 callback
            Batch.Callback<getStatisticsResponse> callback = new Batch.Callback<getStatisticsResponse>() {
                @Override
                public void update(byte[] region, byte[] row,
                        getStatisticsResponse result) {
                    // 直接将 Batch.Call 的结果,即单个 region 的 计算结果 累加到 atoResult
                    atoResult.getAndAdd(result.getResult());
                }
            };

            /**
             * 通过Table.coprocessorService(Class, byte[],
             * byte[],Batch.Call,Batch.Callback) 请求多region服务
             * 
             * 在这个API参数列表中,
             *    * Class代表所需要请求的服务,当前是我们定义在proto中的myStatisticsService服务。
             *    * byte[],
             * byte[]参数指明了startRowKey和endRowKey,当都为null的时候即进行全表的全region的数据计算。
             *    * Batch.Call:需要自定义,API会根据如上参数信息并行的连接各个region,
             * 来执行这个参数中定义的call方法来执行接口方法的调用查询 
             * 此API会将各个region上的信结果信息放入Map通过Callback定义进行处理,处理完所有所有的结果后返回给调用者。
             */
            table.coprocessorService(myStatisticsService.class,
                    Bytes.toBytes(startRowkey), Bytes.toBytes(endRowkey),
                    callable, callback);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != table) {
                table.close();
            }

            if (null != connection) {
                connection.close();
            }
        }
        return atoResult.longValue();
    }

 

  上述代码和非call方式上差别不大,唯一的差别是定义Batch.Callback实例,通过此实例进行结果值得加和处理,由于直接进行了处理,所以这个方法是没有返回值的,调用者需要在调用侧开启一个保存结果的实例,调用完毕后直接返回这个结果实例即可。

测试和验证方法同 Table.coprocessorService(Class, byte[], byte[],Batch.Call)

 

6.  Table.batchCoprocessorService(MethodDescriptor methodDescriptor, Message request, byte[] startKey, byte[] endKey, R responsePrototype)实现

7.  Table.batchCoprocessorService(MethodDescriptor methodDescriptor, Message request, byte[] startKey, byte[] endKey, R responsePrototype, Callback<R> callback)实现

  如前所述,客户端会直接通过网络调用region进行服务请求,如果表比较大,region比较多,这个时候势必会对网络IO造成压力,上述两个方法的原理即是通过在目标region所在的regionserver上开启调用代理,由代理来调用region上的服务,并归并结果后返回客户端,好处在于客户端只要少量的网络请求即可。减小带宽压力。

基本原理:

          

 

实现方法:

  略。

 

8.总结:

  使用框架则必须准守框架规定,框架是固定的,所以上述代码中对于框架的代码都可重用。

  HBase的协处理器为客户端在不修改和重编译源码的情况下增加定制功能,还是很强大的,但是协处理器的增加势必湖造成性能的降低,如Observer协处理器过多的话那么对于插入读取都会有影响,所以不能再协处理器中加入过多逻辑,简单化是必须的。

 

 

参考:

  http://www.ibm.com/developerworks/cn/opensource/os-cn-hbase-coprocessor2/index.html

 

代码下载:

  https://github.com/xufeng79x/Test-HBase-Endpoint

  

 

以上是关于[How to] 使用HBase协处理器---Endpoint客户端代码的实现的主要内容,如果未能解决你的问题,请参考以下文章

hbase协处理器Coprocessor(简介)

HBase协处理器实战

HBase-协处理器详解及实现

2021年大数据HBase(十六):HBase的协处理器(Coprocessor)

2021年大数据HBase(十六):HBase的协处理器(Coprocessor)

hbase协处理器编码实例