1. 程式人生 > >Hbase實戰教程之happybase

Hbase實戰教程之happybase

本文基於實驗室已經搭建好的Hadoop平臺而寫,使用Python呼叫happybase庫。

1.thrift 是facebook開發並開源的一個二進位制通訊中介軟體,通過thrift,我們可以用Python來操作Hbase

        首先開啟Hadoop平臺的HadoopMaster的thrift服務,用Xshell連線HadoopMaster,用root使用者登入,

如果想關閉終端之後,thrift服務繼續執行,可以用daemon模式執行

 

2.安裝happybase和thrift

    pip install happybase

    pip install thrift

 

3.嘗試連線Hbase

import happybase


connection = happybase.Connection('10.1.13.111')

print connection.tables()

此時會出現下面的錯誤:

thriftpy.parser.exc.ThriftParserError: ThriftPy does not support generating module with path in protocol 'c'

解決的辦法請參考這個連線:

http://stackoverflow.com/questions/39220102/error-import-impyla-library-on-windows

即將   C:\Python27\Lib\site-packages\thriftpy\parser\parser.py , line 488        

if url_scheme == '':

修改為

if len(url_scheme) <= 1:

 

上面的方法不好用,應該改成下面就可以

 

if url_scheme == '' or url_scheme == 'c':

 

 

 

 

4.happybase的使用

請參考http://happybase.readthedocs.io/en/latest/index.html

在此做一下簡單的使用介紹

(1)建立連線

import happybase

connection = happybase.Connection('10.1.13.111')

當connection被建立的時候,預設自動與Hbase建立socket連線的。

若不想自動與Hbase建立socket連線,可以將autoconnect引數設定為False

connection = happybase.Connection('10.1.13.111', autoconnect=False)

然後手動與Hbase建立socket連線

connection.open()

 

(2)連線建立好之後檢視可以使用的table

print connection.tables()

 因為還沒有建立table,所以返回結果是 []

 

(3)建立一個table

connection.create_table(
    'my_table',
    {
        'cf1': dict(max_versions=10),
        'cf2': dict(max_versions=1, block_cache_enabled=False),
        'cf3': dict(),  # use defaults
    }
)

此時,我們再通過connection.tables()檢視可以使用的table,結果為['my_table']

建立的table即my_table包含3個列族:cf1、cf2、cf3

 

(4)獲取一個table例項

一個table被建立好之後,要想對其進行操作,首先要獲取這個table例項

table = connection.table('my_table')

(5)使用table的名稱空間

        因為一個Hbase會被多個專案共同使用,所以就會導致table的命名衝突,為了解決這個問題,可以在建立table的時候,手動加上專案的名字作為table名字的字首,例如myproject_xyz。

        但是這樣做比較麻煩,happybase幫我們做好了工作,我們可以在與Hbase建立連線的時候,通過設定table_prefix引數來實現這個功能

connection = happybase.Connection('10.1.13.111', table_prefix='myproject')

    此時connection.tables()只會返回包含在該名稱空間裡的tables,且返回的tables的名字會以簡單的形式顯示,即不包含字首。

 

(6)儲存資料:Hbase裡 儲存的資料都是原始的位元組字串

cloth_data = {'cf1:content': u'牛仔褲', 'cf1:price': '299', 'cf1:rating': '98%'}

hat_data = {'cf1:content': u'鴨舌帽', 'cf1:price': '88', 'cf1:rating': '99%'}

shoe_data = {'cf1:content': u'耐克', 'cf1:price': '988', 'cf1:rating': '100%'}

author_data = {'cf2:name': u'LiuLin', 'cf2:date': '2017-03-09'}


table.put(row='www.test1.com', data=cloth_data)

table.put(row='www.test2.com', data=hat_data)

table.put(row='www.test3.com', data=shoe_data)

table.put(row='www.test4.com', data=author_data)

使用put一次只能儲存一行資料

如果row key已經存在,則變成了修改資料

 

(7)更好的儲存資料

table.put()方法會立即給Hbase Thrift server傳送一條命令。其實這種方法的效率並不高,我們可以使用更高效的table.batch()方法。

# 使用batch一次插入多行資料

bat = table.batch()

bat.put('www.test5.com', {'cf1:price': 999, 'cf2:title': 'Hello Python', 'cf2:length': 34, 'cf3:code': 'A43'})

bat.put('www.test6.com', {'cf1:content': u'剃鬚刀', 'cf1:price': 168, 'cf1:rating': '97%'})

bat.put('www.test7.com', {'cf3:function': 'print'})

bat.send()

更有用的方法是使用上下文管理器來管理batch,這樣就不用手動傳送資料了,即不再需要bat.send()

# 使用with來管理batch

with table.batch() as bat:
 
    bat.put('www.test5.com', {'cf1:price': '999', 'cf2:title': 'Hello Python', 'cf2:length': '34', 'cf3:code': 'A43'})

    bat.put('www.test6.com', {'cf1:content': u'剃鬚刀', 'cf1:price': '168', 'cf1:rating': '97%'})

    bat.put('www.test7.com', {'cf3:function': 'print'})

還可以刪除資料

# 在batch中刪除資料

with table.batch() as bat:

    bat.put('www.test5.com', {'cf1:price': '999', 'cf2:title': 'Hello Python', 'cf2:length': '34', 'cf3:code': 'A43'})

    bat.put('www.test6.com', {'cf1:content': u'剃鬚刀', 'cf1:price': '168', 'cf1:rating': '97%'})

    bat.put('www.test7.com', {'cf3:function': 'print'})

    bat.delete('www.test1.com')

 batch將資料儲存在記憶體中,知道資料被send,第一種send資料的方法是顯示地傳送,即bat.send(),第二種send資料的方法是到達with上下文管理器的結尾自動傳送。這樣就存在一個問題,萬一資料量很大,就會佔用太多的記憶體。所以我們在使用table.batch()的時候要通過batch_size引數來設定batch的大小

# 通過batch_size引數來設定batch的大小
with table.batch(batch_size=10) as bat:
    for i in range(16):
        bat.put('www.test{}.com'.format(i), {'cf1:price': '{}'.format(i)})

(8)掃描一個table裡的資料

# 全域性掃描一個table
for key, value in table.scan():
    print key, value

結果如下:

這種全域性掃描一個表格其實代價是很大的,尤其是當資料量很大的時候。我們可以通過設定開始的row key 或結束的row key或者同時設定開始和結束的row key來進行區域性查詢

# 通過row_start引數來設定開始掃描的row key
for key, value in table.scan(row_start='www.test2.com'):
    print key, value
# 通過row_stop引數來設定結束掃描的row key
for key, value in table.scan(row_stop='www.test3.com'):
    print key, value
# 通過row_start和row_stop引數來設定開始和結束掃描的row key
for key, value in table.scan(row_start='www.test2.com', row_stop='www.test3.com'):
    print key, value

另外,還可以通過設定row key的字首來進行區域性掃描

# 通過row_prefix引數來設定需要掃描的row key
for key, value in table.scan(row_prefix='www.test'):
    print key, value

(9)檢索資料

# 檢索一行資料
row = table.row('www.test4.com')
print row

直接返回該row key的值(以字典的形式),結果為:

{'cf2:name': 'LiuLin', 'cf2:date': '2017-03-09'}
# 檢索多行資料
rows = table.rows(['www.test1.com', 'www.test4.com'])
print rows

返回的是一個list,list的一個元素是一個tuple,tuple的第一個元素是row key,第二個元素是row key的值

如果想使檢索多行資料即table.rows()返回的結果是一個字典,可以這樣處理

# 檢索多行資料,返回字典
rows_dict = dict(table.rows(['www.test1.com', 'www.test4.com']))
print rows_dict

如果想使table.rows()返回的結果是一個有序字典,即OrderedDict,可以這樣處理

# 檢索多行資料,返回有序字典

from collection import OrderedDict

rows_ordered_dict = OrderedDict(table.rows(['www.test1.com', 'www.test4.com']))

print rows_ordered_dict

(10)更好地檢索資料

# 通過指定列族來檢索資料
row = table.row('www.test1.com', columns=['cf1'])
print row
# 通過指定列族中的列來檢索資料
row = table.row('www.test1.com', columns=['cf1:price', 'cf1:rating'])
print row
print row['cf1:price']

在Hbase裡,每一個cell都有一個時間戳timestamp,可以通過時間戳來檢索資料

# 通過指定時間戳來檢索資料,時間戳必須是整數
row = table.row('www.test1.com', timestamp=1489070666)
print row

預設情況下,返回的資料並不會包含時間戳,如果你想獲取時間戳,這樣就可以了

# 在返回的資料裡面包含時間戳

row = table.row(row='www.test1.com', columns=['cf1:rating', 'cf1:price'], include_timestamp=True)

print row

對於同一個單元的值,Hbase儲存了多個版本,在建立表的時候可以通過max_versions引數來設定一個列族的最大版本號,如果想檢索某一cell所有的版本,可以這樣

# 檢索某一個cell所有的版本
cells = table.cells(b'www.test1.com', column='cf1:price')
print cells

也可以通過version引數來指定需要檢索的前n個版本,如下

# 通過設定version引數來檢索前n個版本
cells = table.cells(b'www.test1.com', column='cf1:price', versions=3)
print cells

(11)刪除資料

# 刪除一整行資料
table.delete('www.test4.com')
# 刪除一個列族的資料
table.delete('www.test2.com', columns=['cf1'])
# 刪除一個列族中幾個列的資料
table.delete('www.test2.com', columns=['cf1:name', 'cf1:price'])

(12)使用連線池

Hbase自帶有執行緒安全的連線池,踏允許多個執行緒共享和重用已經開啟的連線。這對於多執行緒的應用是非常有用的。當一個執行緒申請一個連線,它將獲得一個租賃憑證,在此期間,這個執行緒單獨享有這個連線。當這個執行緒使用完該連線之後,它將該連線歸還給連線池以便其他的執行緒可以使用

# 建立連線,通過引數size來設定連線池中連線的個數
pool = happybase.ConnectionPool(size=3, host='10.1.13.111', table_prefix='myProject')
# 獲取連線
with pool.connection() as connection:
    print connection.tables()