用Python操做HBase之HBase-Thrift

安裝Thrift

安裝Thrift的具體操做，請點擊連接

Python操做Hbase

• 安裝依賴包

pip install thrift
pip install hbase-thrift

 

• 鏈接與操做代碼以下：

from thrift.transport import TSocket,TTransport
from thrift.protocol import TBinaryProtocol
from hbase import Hbase

# thrift默認端口是9090
socket = TSocket.TSocket('192.168.0.156',9090)
socket.setTimeout(5000)

transport = TTransport.TBufferedTransport(socket)
protocol = TBinaryProtocol.TBinaryProtocol(transport)

client = Hbase.Client(protocol)
socket.open()

print client.getTableNames()
print client.get('test','row1','cf:a')

 

經常使用方法說明

• createTable(tbaleName,columnFamilies)：建立表，無返回值 
  • tableName：表名
  • columnFamilies：列族信息，爲一個ColumnDescriptor列表

from hbase.ttypes import ColumnDescriptor

# 定義列族
column = ColumnDescriptor(name='cf')
# 建立表
client.createTable('test4',[column])

 

• enabledTable(tbaleName)：啓用表，無返回值 
  • tableName：表名

# 啓用表，若表以前未被禁用將會引起IOError錯誤
client.enabledTable('test')

 

• disableTable(tbaleName)：禁用表，無返回值 
  • tableName：表名

# 禁用表，若表以前未被啓用將會引起IOError錯誤
client.disableTable('test')

 

• isTableEnabled(tbaleName)：驗證表是否被啓用，返回一個bool值 
  • tableName：表名

client.isTableEnabled('test')

 

• getTableNames(tbaleName)：獲取表名列表，返回一個str列表 
  • tableName：表名

client.getTableNames()

• 1

• getColumnDescriptors(tbaleName)：獲取全部列族信息，返回一個字典 
  • tableName：表名

client.getColumnDescriptors('test')

 

• getTableRegions(tbaleName)：獲取全部與表關聯的regions，返回一個TRegionInfo對象列表 
  • tableName：表名

client.getTableRegions('test')

 

• deleteTable(tbaleName)：刪除表，無返回值 
  • tableName：表名

# 表不存在將會引起IOError(message='java.io.IOException: table does not exist...)錯誤
# 表未被禁用將會引起IOError(message='org.apache.hadoop.hbase.TableNotDisabledException:...)錯誤

client.deleteTable('test5')

 

• get(tableName,row,column)：獲取數據列表，返回一個hbase.ttypes.TCell對象列表 
  • tableName：表名
  • row：行
  • column：列

result = client.get('test','row1','cf:a')       # 爲一個列表，其中只有一個hbase.ttypes.TCell對象的數據
print result[0].timestamp
print result[0].value

 

• getVer(tableName,row,column,numVersions)：獲取數據列表，返回一個hbase.ttypes.TCell對象列表 
  • tableName：表名
  • row：行
  • column：列
  • numVersions：要檢索的版本數量

result = client.get('test','row1','cf:a',2)       # 爲一個列表，其中只有一個hbase.ttypes.TCell對象的數據
print result[0].timestamp
print result[0].value

 

• getVerTs(tableName,row,column,timestamp,numVersions)：獲取小於當前時間戳的數據列表(彷佛只獲取前一個)，返回一個hbase.ttypes.TCell對象列表 
  • tableName：表名
  • row：行
  • column：列
  • timestamp：時間戳
  • numVersions：要檢索的版本數量

result = client.get('test','row1','cf:a',2)       # 爲一個列表，其中只有一個hbase.ttypes.TCell對象的數據
print result[0].timestamp
print result[0].value

 

• getRow(tableName,row)：獲取表中指定行在最新時間戳上的數據。返回一個hbase.ttypes.TRowResult對象列表，若是行號不存在返回一個空列表 
  • tableName：表名
  • row：行

# 行
row = 'row1'
# 列
column = 'cf:a'
# 查詢結果
result = client.getRow('test',row)      # result爲一個列表
for item in result:                     # item爲hbase.ttypes.TRowResult對象
    print item.row
    print item.columns.get('cf:a').value        # 獲取值。item.columns.get('cf:a')爲一個hbase.ttypes.TCell對象
    print item.columns.get('cf:a').timestamp    # 獲取時間戳。item.columns.get('cf:a')爲一個hbase.ttypes.TCell對象

 

• getRowWithColumns(tableName,row,columns)：獲取表中指定行與指定列在最新時間戳上的數據。返回一個hbase.ttypes.TRowResult對象列表，若是行號不存在返回一個空列表 
  • tableName：表名
  • row：行
  • columns：列，list

result = client.getRowWithColumns('test','row1',['cf:a','df:a'])
for item in result:
    print item.row
    print item.columns.get('cf:a').value
    print item.columns.get('cf:a').timestamp

    print item.columns.get('df:a').value
    print item.columns.get('df:a').timestamp

 

• getRowTs(tableName,row,timestamp)：獲取表中指定行而且小於這個時間戳的全部數據。返回一個hbase.ttypes.TRowResult對象列表，若是行號不存在返回一個空列表 
  • tableName：表名
  • row：行
  • timestamp：時間戳

result = client.getRowTs('test','row1',1513069831512)

 

• getRowWithColumnsTs(tableName,row,columns,timestamp)：獲取指定行與指定列，而且小於這個時間戳的全部數據。返回一個hbase.ttypes.TRowResult對象列表，若是行號不存在返回一個空列表 
  • tableName：表名
  • row：行
  • columns：列，list
  • timestamp：時間戳

result = client.getRowWithColumnsTs('test','row1',['cf:a','cf:b','df:a'],1513069831512)

 

• mutateRow(tableName,row,mutations)：在表中指定行執行一系列的變化操做。若是拋出異常，則事務被停止。使用默認的當前時間戳，全部條目將具備相同的時間戳。無返回值 
  • tableName：表名
  • row：行
  • mutations：變化,list

from hbase.ttypes import Mutation

mutation = Mutation(name='cf:a',value='1')

# 插入數據。若是在test表中row行cf:a列存在，將覆蓋
client.mutateRow('test','row1',[mutation])

 

• mutateRowTs(tableName,row,mutations,timestamp)：在表中指定行執行一系列的變化操做。若是拋出異常，則事務被停止。使用指定的時間戳，全部條目將具備相同的時間戳。若是是更新操做時，若是指定時間戳小於原來數據的時間戳，將被忽略。無返回值 
  • tableName：表名
  • row：行
  • mutations：變化,list
  • timestamp：時間戳

from hbase.ttypes import Mutation
# value必須爲字符串格式，不然將報錯
mutation = Mutation(column='cf:a',value='2')
client.mutateRowTs('test','row1',[mutation],1513070735669)

 

• mutateRows(tableName,rowBatches)：在表中執行一系列批次(單個行上的一系列突變)。若是拋出異常，則事務被停止。使用默認的當前時間戳，全部條目將具備相同的時間戳。無返回值 
  • tableName：表名
  • rowBatches：一系列批次

from hbase.ttypes import Mutation,BatchMutation
mutation = Mutation(column='cf:a',value='2')
batchMutation = BatchMutation('row1',[mutation])
client.mutateRows('test',[batchMutation])

 

• mutateRowsTs(tableName,rowBatches,timestamp)：在表中執行一系列批次(單個行上的一系列突變)。若是拋出異常，則事務被停止。使用指定的時間戳，全部條目將具備相同的時間戳。若是是更新操做時，若是指定時間戳小於原來數據的時間戳，將被忽略。無返回值 
  • tableName：表名
  • rowBatches：一系列批次，list
  • timestamp：時間戳

mutation = Mutation(column='cf:a',value='2')
batchMutation = BatchMutation('row1',[mutation])
client.mutateRowsTs('cx',[batchMutation],timestamp=1513135651874)

 

• atomicIncrement(tableName,row,column,value)：原子遞增的列。返回當前列的值 
  • tableName：表名
  • row：行
  • column：列
  • value：原子遞增的值

result = client.atomicIncrement('cx','row1','cf:b',1)
print result    # 若是以前的值爲2，此時值爲3

 

• deleteAll(tableName,row,column)：刪除指定表指定行與指定列的全部數據，無返回值 
  • tableName：表名
  • row：行
  • column：列

client.deleteAll('cx','row1','cf:a')

 

• deleteAllTs(tableName,row,column,timestamp)：刪除指定表指定行與指定列中，小於等於指定時間戳的全部數據，無返回值 
  • tableName：表名
  • row：行
  • column：列
  • timestamp：時間戳

client.deleteAllTs('cx','row1','cf:a',timestamp=1513569725685)

 

• deleteAllRow(tableName,row)：刪除整行數據，無返回值 
  • tableName：表名
  • row：行

client.deleteAllRow('cx','row1')

 

• deleteAllRowTs(tableName,row,timestamp)：刪除指定表指定行中，小於等於此時間戳的全部數據，無返回值 
  • tableName：表名
  • row：行
  • timestamp：時間戳

client.deleteAllRowTs('cx','row1',timestamp=1513568619326)

 

• scannerOpen(tableName,startRow,columns)：在指定表中，從指定行開始掃描，到表中最後一行結束，掃描指定列的數據。返回一個ScannerID，int類型 
  • tableName：表名
  • startRow：起始行
  • columns：列名列表,list類型

scannerId = client.scannerOpen('cx','row2',["cf:b","cf:c"])

 

• scannerOpenTs(tableName,startRow,columns,timestamp)：在指定表中，從指定行開始掃描，獲取全部小於指定時間戳的全部數據，掃描指定列的數據。返回一個ScannerID，int類型 
  • tableName：表名
  • startRow：起始行
  • columns：列名列表,list類型
  • timestamp：時間戳

scannerId = client.scannerOpenTs('cx','row1',["cf:a","cf:b","cf:c"],timestamp=1513579065365)

 

• scannerOpenWithStop(tableName,startRow,stopRow,columns)：在指定表中，從指定行開始掃描，掃描到結束行結束(並不獲取指定行的數據)，掃描指定列的數據。返回一個ScannerID，int類型 
  • tableName：表名
  • startRow：起始行
  • stopRow：結束行
  • columns：列名列表,list類型

scannerId = client.scannerOpenWithStop('cx','row1','row2',["cf:b","cf:c"])

 

• scannerOpenWithStopTs(tableName,startRow,stopRow,columns,timestamp)：在指定表中，從指定行開始掃描，掃描到結束行結束(並不獲取指定行的數據)，獲取全部小於指定時間戳的全部數據，掃描指定列的數據。返回一個ScannerID，int類型 
  • tableName：表名
  • startRow：起始行
  • stopRow：結束行
  • columns：列名列表,list類型
  • timestamp：時間戳

scannerId = client.scannerOpenWithStopTs('cx','row1','row2',["cf:a","cf:b","cf:c"],timestamp=1513579065365)

 

• scannerOpenWithPrefix(tableName,startAndPrefix,columns)：在指定表中，掃描具備指定前綴的行，掃描指定列的數據。返回一個ScannerID，int類型 
  • tableName：表名
  • startAndPrefix：行前綴
  • columns：列名列表,list類型

scannerId = client.scannerOpenWithPrefix('cx','row',["cf:b","cf:c"])

 

• scannerGet(id)：根據ScannerID來獲取結果，返回一個hbase.ttypes.TRowResult對象列表 
  • id：ScannerID

scannerId = client.scannerOpen('cx','row1',["cf:b","cf:c"])
while True:
    result = client.scannerGet(scannerId)
    if not result:
        break
    print result

 

• scannerGetList(id,nbRows)：根據ScannerID來獲取指定數量的結果，返回一個hbase.ttypes.TRowResult對象列表 
  • id：ScannerID
  • nbRows：指定行數

scannerId = client.scannerOpen('cx','row1',["cf:b","cf:c"])
result = client.scannerGetList(scannerId,2)

 

• scannerClose(id)：關閉掃描器，無返回值 
  • id：ScannerID

client.scannerClose(scannerId)