HBase赠书|云HBase Thrift使用最佳实践

Posted 2021-04-18 HBase技术社区

社区福利

《HBase原理与实践》

↓

参与方式

截止到2019.09.20 00:00

https://www.douban.com/doubanapp/dispatch/book/34819650

短评点赞量排名第一

获赠

胡争&范欣欣签名版《HBase原理与实践》纸质书

HBase原生只提供了Java API客户端，针对诸如Python、php、C++等非Java语言一般都是通过Thrift代理的方式访问HBase服务，本文从Thrift架构、HBase Thrift API使用以及如何监控Thrift等几个方面详细介绍云HBase Thrift使用最佳实践。

Thrift是一种接口描述语言和二进制通讯协议，它被用来定义和创建跨语言的服务。通常被当作RPC框架来使用，最初是Facebook为了解决大规模的跨语言服务调用而设计开发出来的，Thrift架构图如下： 

从上图可以看出Thrift采用了分层架构，概念非常清晰，便于大家理解使用，从下至上依次为传输层、协议层和处理层；

• 传输层：对底层IO进行了简单的抽象，解决数据在网络中的传输问题，通过字节流的方式发送和接受Message，支持TSocket、TFileTransport、TZlibTransport、TBufferedTranspor、TFramedTransport等多种传输层协议。

• 协议层：将底层的字节流转换成数据流，对字节流数据进行序列化和反序列化，协议层决定了到底被传输的数据是什么，支持包括TBinaryProtocol、TCompactProtocal、TCompactProtocal在内的多种协议；

• 处理层：包含用户的完整逻辑，决定数据应该被如何进行处理；

• TServer：Thrift同时还提供了多种服务端模型，TSimpleServer(单线程服务模型，常用于测试)、TThreadPoolServer(多线程服务模型，使用标准的阻塞式IO)、TNonblockingServer(多线程服务模型，使用非阻塞式IO,需同TFramedTransport配合使用)、THttpServer(支持HTTP协议)。

了解完Thrift架构，我们通过一个简单的例子来了解Thrift是如何工作的。Thrift定义了一种接口描述语言IDL，IDL文件可以被Thrift代码生成期处理生成目标语言的代码，我们定义了一个简单的IDL文件tutorial.thrift文件，内容如下：

#cat tutorial.thrift
server Calculator{
  void ping(),
  i32 add(i:i32 num1, 2:i32 num2)
}

文件中我们定义了一个名为calculator的service，包含两个接口ping和add，围绕这个文件我们分别实现客户端和服务端代码,在编写代码之前需要使用如下命令生成目标语言代码：

thrift -gen py(替换py为其他语言可以生成对应语言的) tutorial.thrift

以Python为例实现服务端代码：

import sys, glob
sys.path.append('gen-py')
sys.path.insert(0, glob.glob('../../lib/py/build/lib.*')[0])

from tutorial import Calculator
from tutorial.ttypes import *

from shared.ttypes import SharedStruct

from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from thrift.server import TServer

class CalculatorHandler:
 def __init__(self):
 self.log = {}

  def ping(self):
    print 'ping()'

  def add(self, n1, n2):
    print 'add(%d,%d)' % (n1, n2)
    return n1+n2

handler = CalculatorHandler()
processor = Calculator.Processor(handler)
transport = TSocket.TServerSocket(port=9090)
tfactory = TTransport.TBufferedTransportFactory()
pfactory = TBinaryProtocol.TBinaryProtocolFactory()

server = TServer.TSimpleServer(processor, transport, tfactory, pfactory)

# You could do one of these for a multithreaded server
#server = TServer.TThreadedServer(processor, transport, tfactory, pfactory)
#server = TServer.TThreadPoolServer(processor, transport, tfactory, pfactory)

print 'Starting the server...'
server.serve()
print 'done.'

整个服务端需要我们实现的就是处理层，决定如果处理客户群传输过来的数据，对应CalculatorHandler实现了接口中定义的ping和add方法，分别输出"ping()"和实现整数加法，将handler作为参数生成Processor，传输层使用TServerSocket监听9090端口，传输层使用TBinaryProtocol同时使用Buffer进行读写(用TBufferedTransport进行装饰),可以选用不同的Server模型，测试使用TSimpleServer，将传输层、协议层、处理层以及Server组合到一起就完成全部服务端开发工作；在看下客户端实现：

import sys, glob
sys.path.append('gen-py')
sys.path.insert(0, glob.glob('../../lib/py/build/lib.*')[0])

from tutorial import Calculator
from tutorial.ttypes import *

from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol

try:

  # Make socket
  transport = TSocket.TSocket('localhost', 9090)

  # Buffering is critical. Raw sockets are very slow
  transport = TTransport.TBufferedTransport(transport)

  # Wrap in a protocol
  protocol = TBinaryProtocol.TBinaryProtocol(transport)

  # Create a client to use the protocol encoder
  client = Calculator.Client(protocol)

  # Connect!
  transport.open()

  client.ping()
  print 'ping()'
  
  sum = client.add(1,1)
  print '1+1=%d' % (sum)

  # Close!
  transport.close()

except Thrift.TException, tx:
  print '%s' % (tx.message)

客户端的实现也很简单传输层使用TSocket访问本地9090端口同时使用buffer进行读写(TBufferedTransport进行装饰)，协议层使用TBinaryProtocol，初始化Calculator.Client(根据IDL自动生成代码)，我就可以使用client进行方法的调用；
上面这个例子完整的说明了如何通过Thrift实现自己的业务逻辑，通过Thrift访问HBase同上面本质上是完全一样的，只不过HBase提供了更多的接口而已，HBase提供了相应的IDL文件Hbase.thrift，通过该文件可以生成对应语言的客户端代码，还是已python为例实现基本的CURD操作，代码如下：

from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol

# hbase 客户端代码是由 thrift -gen py hbase.thrift 生成，拷贝到工程目录下
from hbase.THBaseService import Client
from hbase.ttypes import TScan, TColumn, TGet, TPut, TColumnValue, TableDescriptor, ColumnDescriptor, TDelete

class HbaseClient(object):
 def __init__(self, host='hbasetest02.et2sqa.tbsite.net', port=9090):
        self.transport = TTransport.TBufferedTransport(TSocket.TSocket(host, port))
        protocol = TBinaryProtocol.TBinaryProtocol(self.transport)
        self.client = Client(protocol)

    def list_tables(self):
        self.transport.open()
        tables = self.client.listTables()
        self.transport.close()
        return tables

    def scan_data(self, table, startRow, stopRow, limit, columns=None):
        self.transport.open()
        if columns:
            columns = [TColumn(*i.split(':')) for i in columns]
        scan = TScan(startRow=startRow, stopRow=stopRow, columns=columns)
        rows = self.client.scan(table, scan, limit)
        self.transport.close()
        return rows

    def get(self, table, row=None, columns=None):
        self.transport.open()
        if columns:
            columns = [TColumn(*i.split(':')) for i in columns]
        get = TGet(row=row, columns=columns)
        res = self.client.get(table, get)
        self.transport.close()
        return res

    def put(self, table, row, columnValues):
        self.transport.open()
        put = TPut(row, columnValues)
        self.client.put(table, put)
        self.transport.close()

    def table_exists(self, table):
        self.transport.open()
        res = self.client.tableExists(table)
        self.transport.close()
        return res

    def delete_single(self, table, delete):
        self.transport.open()
        self.client.deleteSingle(table, delete)
        self.transport.close()

    def create_table(self, tableDescriptor, splitKeys):
        self.transport.open()
        self.client.createTable(tableDescriptor, splitKeys)
        self.transport.close()

    def enable_table(self, table):
        self.transport.open()
        self.client.enableTable(table)
        self.transport.close()

    def disable_table(self, table):
        self.transport.open()
        self.client.disableTable(table)
        self.transport.close()


if __name__ == '__main__':
    client = HbaseClient()
    print client.list_tables()

    print client.scan_data('bear_test', startRow='row1', stopRow='row5', limit=10, columns=['f1:name'])
    print client.get('bear_test', 'row1', ['f1:age'])

    columnValues = [TColumnValue('f1', 'name', 'bear3')]
    client.put('bear_test', 'row4', columnValues)

    print client.table_exists('bear_test')

    delete = TDelete('row4')
    client.delete_single('bear_test', delete)
    print client.scan_data('bear_test', startRow='row1', stopRow='row5', limit=10, columns=['f1:name'])

    tableDescriptor = TableDescriptor('bear_test1', families=[ColumnDescriptor(name='f1', maxVersions=1)])
    client.create_table(tableDescriptor, None)
    print client.table_exists('bear_test1')

点击Ganglia进入到主页，在主页最下面可以找到hbase_cluster。

点击hbase_cluster进入到hbase监控详情：

详情页的最上面为时间设置，可以选择hour、2hr、4hr等不同的时间纬度，也可以在from、to中设置自定义时间

详情页的下面为指标选择：

Thrift相关的指标都是以thrift-one和thrift-two开头的，分别对应thrift1和thrift2，thrift默认使用TThreadPoolServer服务模型，使用Thrift主要关注3种监控指标：

• thrift-one.Thrift.numActiveWorkers:表示当前活跃的工作线程，默认的最大线程数为1000，通过该指标可以判断当前thrift server的负载情况

• thrift-one.Thrift.callQueueLen:表示请求的队列大小，默认队列长度为1000，该指标同thrift-one.Thrift.numActiveWorkers一同反应thrift server的负载情况

• thrift-one.Thrift.方法名_开头的各种指标：表示对应方法的thrift server耗时，以batchGet方法为例thrift-one.Thrift.batchGet_max、thrift-one.Thrift.batchGet_mean以及thrift-one.Thrift.batchGet_min分别表示batchGet的最大、平均以及最小耗时。

本群为HBase+Spark技术交流讨论，整合最优质的专家资源和技术资料会定期开展线下技术沙龙，专家技术直播，专家答疑活动。点击链接钉钉入群：

https://dwz.cn/Fvqv066s或扫码进群

本群为Cassandra技术交流讨论，整合最优质的专家资源和技术资料会定期开展线下技术沙龙，专家技术直播，专家答疑活动。Cassandra 社区钉钉大群：

https://c.tb.cn/F3.ZRTY0o或扫码进群