This is a client python API for HBase thrift2 service. Added exception handling and autoretry, reconnection functions.
Use pip to install the package (recommand).
pip install thbase
fromthbase.thrift2.clientimportClientfromthbase.configimportClientConfig, TransportType, ProtocolTypefromthbase.thrift2.operationimportDelete, Scan, Put, Getif__name__=='__main__':
conf=ClientConfig(thrift_host=host,
port=port,
retry_times=10,
retry_timeout=10,
transport_type=TransportType.BUFFERED,
protocol_type=ProtocolType.BINARY,
use_ssl=True,
batch_size=10,
use_http=True)
client=Client(conf)
ifclient.open_connection(): # if use SASL client, this value will always be None, so that should not use if here.table=client.get_table("your_table_name")
# example for a single operationp=Put(row="your_row_key",
family="column_family",
qualifier="column_qualifier",
value="your_value")
table.put(p)
# example for a batch operationput_list= []
foriinrange(100):
row_key="row{}".format(i)
p=Put(row=row_key,
family="column_family",
qualifier="column_qualifier",
value="your_value")
put_list.append(p)
table.put_batch(put_list)
# do not forget to close the connection after usingclient.close_connection()###Multi-threaded The thrift basic transport is not thread-safe. In this case, if you want to parallelize your program, you should create a new connection object for each thread. The sample code is:
fromthbase.thrift2.clientimportClientfromthbase.configimportClientConfig, TransportType, ProtocolTypefromthbase.thrift2.operationimportDelete, Scan, Put, Getimportthreadingimportlogging# initialize the logger to check runtime log information for more details about logger usage please refer: https://docs.python.org/2.7/library/logging.htmllogging.basicConfig()
host=your_hostport=your_portdefdemo_func(conf):
# get the Client objectclient=Client(conf)
# Open the connectionifclient.open_connection(): # if use SASL client, this value will always be None, so that should not use if here.# get a table object with given table nametable=client.get_table("your_table_name")
# single put operationp=Put(row="your_row_key",
family="your_column_family",
qualifier="your_column_qualifier",
value="your_data")
iftable.put(p):
# do sthelse:
# do sth# batch put operationput_list= []
foriinrange(100):
row_key="row{}".format(i)
p=Put(row=row_key,
family="your_column_family",
qualifier="your_column_qualifier",
value="your_data")
put_list.append(p)
iftable.put_batch(put_list):
# do sthelse:
# do sth# single get operationg=Get(row=row_key,
family="your_column_family",
qualifier="your_coloumn_qualifier",
max_versions=your_max_version,
filter_string="your_filter_string")
result=table.get(g)
# batch get operationget_list= []
foriinrange(10):
get_list.append(Get(row=row_key,
family='0',
qualifier=None,
max_versions=1))
table.get_batch(get_list)
# single delete operationdelete=Delete(row='row10',
family='0')
iftable.delete(delete):
# do sth.else:
# do sth.# delete batch operationdelete_list= []
foriinrange(10):
delete_list.append(Delete(row='row{}'.format(i)))
iftable.delete_batch(delete_list):
# do sth.else:
# do sth.# scan operationscan=Scan(start_row="your_start_row_key",
family="your_column_family",
qualifier="your_column_qualifier",
max_versions="your_max_version",
reversed="if_reverse_results",
filter_string="your_filter_string")
results=table.scan(scan=scan)
print [str(r) forrinresults]
# don't forget to close the connection after using.client.close_connection()
if__name__=='__main__':
# initialize the client configurationconf=ClientConfig(thrift_host=host, # thrift server address type: strport=9090, # thrift server port type: int, default 9090retry_times=10,
# retry times for reconnection when client lose connnection with the server, type: int, default: 10retry_timeout=10, # seconds between two reconnection tries, type: int, default: 10transport_type=TransportType.FRAMED,
# Use the Enum class, default: TransportType.BUFFEREDprotocol_type=ProtocolType.BINARY,
# Use the relative Enum class, default: ProtocolType.BINARYuse_ssl=True,
# If True, the Client will use SSL Socket to transport requests to the thrift serverbatch_size=10, # The max size of the batch operationsuse_http=True,
)
# initialize thread listthread_list= []
for_inrange(10):
x=threading.Thread(target=demo_func, args=(conf,))
thread_list.append(x)
x.start()
forthreadinthread_list:
thread.join()The github repository is:
https://github.com/YutSean/thbase