'雲HBase Thrift使用最佳實踐'

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

對於上述例子介紹下幾個關鍵的點，HBase Thrift Server默認使用TThreadPoolServer服務模型(在併發數可控的情況下使用該服務模型性能最好)，相應的客戶端傳輸層使用TTransport.TBufferedTransport(TSocket.TSocket(host, port))以及協議層TBinaryProtocol同服務端進行通信；同時另外一個需要注意的點就是Client本身並不是線程安全的，每一個線程要使用單獨的Client；

阿里雲HBase本身提供了高可用版本的ThriftServer，通過負載均衡將流量分散到不同的Thrift Server上，最大程度上保證服務的高可用和高性能，使用戶可以不必關係資源、部署以及運維，將精力都放在業務邏輯的實現上，雲HBase的連接地址以及使用可以參考(https://help.aliyun.com/document_detail/87068.html);學會了如何使用客戶端，同時雲HBase Thrift Server免運維，那麼唯一需要關心的就是監控了，每一個雲HBase實例多部署了ganglia監控，可以在雲HBase實例-》數據庫連接-》UI訪問中找到：

點擊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的最大、平均以及最小耗時。

作者：修者

本文為雲棲社區內容，未經允許不得轉載。