Project description

This is a client python API for HBase thrift2 service. Added exception handling and autoretry, reconnection functions.

Install

Use pip to install the package (recommand).

pip install thbase

Usage

Single thread

from thbase.thrift2.client import Client
from thbase.config import ClientConfig, TransportType, ProtocolType
from thbase.thrift2.operation import Delete, Scan, Put, Get

if __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)
    if client.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 operation
        p = Put(row="your_row_key",
                family="column_family",
                qualifier="column_qualifier",
                value="your_value")
        table.put(p)
        
        # example for a batch operation
        put_list = []
        for i in range(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 using
        client.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:

from thbase.thrift2.client import Client
from thbase.config import ClientConfig, TransportType, ProtocolType
from thbase.thrift2.operation import Delete, Scan, Put, Get
import threading
import logging
 
 
# initialize the logger to check runtime log information for more details about logger usage please refer: https://docs.python.org/2.7/library/logging.html
logging.basicConfig()
 
 
host = your_host
port = your_port
 
 
def demo_func(conf):
    # get the Client object
    client = Client(conf)
 
    # Open the connection
    if client.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 name
        table = client.get_table("your_table_name")
        # single put operation
        p = Put(row="your_row_key",
                family="your_column_family",
                qualifier="your_column_qualifier",
                value="your_data")
        if table.put(p):
            # do sth
        else:
            # do sth
        
        
        # batch put operation
        put_list = []
        for i in range(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)
        if table.put_batch(put_list):
            # do sth
        else:
            # do sth
        
        # single get operation
        g = 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 operation
        get_list = []
        for i in range(10):
            get_list.append(Get(row=row_key,
                                family='0',
                                qualifier=None,
                                max_versions=1))
        table.get_batch(get_list)
        
        # single delete operation
        delete = Delete(row='row10',
                        family='0')
        if table.delete(delete):
            # do sth.
             
        else:
            # do sth.
            
        # delete batch operation
        delete_list = []
        for i in range(10):
            delete_list.append(Delete(row='row{}'.format(i)))
        if table.delete_batch(delete_list):
            # do sth.
        else:
            # do sth.
        
        # scan operation
        scan = 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) for r in results]
        # don't forget to close the connection after using.
        client.close_connection()
 
 
if __name__ == '__main__':
 
    # initialize the client configuration
    conf = ClientConfig(thrift_host=host,  # thrift server address type: str
                        port=9090,  # thrift server port type: int, default 9090
                        retry_times=10,
                        # retry times for reconnection when client lose connnection with the server, type: int, default: 10
                        retry_timeout=10,  # seconds between two reconnection tries, type: int, default: 10
                        transport_type=TransportType.FRAMED,
                        # Use the Enum class, default: TransportType.BUFFERED
                        protocol_type=ProtocolType.BINARY,
                        # Use the relative Enum class, default: ProtocolType.BINARY
                        use_ssl=True,
                        # If True, the Client will use SSL Socket to transport requests to the thrift server
                        batch_size=10,  # The max size of the batch operations
                        use_http=True,
                        )
    # initialize thread list
    thread_list = []
    for _ in range(10):
        x = threading.Thread(target=demo_func, args=(conf,))
        thread_list.append(x)
        x.start()
    for thread in thread_list:
        thread.join()

Source

The github repository is:
https://github.com/YutSean/thbase