title: Python 讀寫(xiě) hbase 數(shù)據(jù)的正確姿勢(shì)（一）
tags:
- hbase
- happybase
- python
categories:
- ?Hbase
comments: true
date: 2017-09-09 19:00:00
之前操作 hbase 大都是用 java 寫(xiě)贷祈，或者偶爾用 python 寫(xiě)幾個(gè)一些簡(jiǎn)單的 put梢莽、get 操作满粗。最近在使用 happybase 庫(kù)批量向 hbase 導(dǎo)入數(shù)據(jù)笆豁，并通過(guò) java 實(shí)現(xiàn)查詢的一些復(fù)雜的搜索時(shí)（scan+filter），遇到了一些有趣的問(wèn)題。
實(shí)驗(yàn)版本
Hbase 版本：1.0.0
Happybase 版本：1.1.0
Python 版本：2.7.13
問(wèn)題1：filter 過(guò)濾失敗
問(wèn)題重現(xiàn)
hbase 的使用場(chǎng)景大概是這樣的：
有一個(gè) hbase table，存儲(chǔ)一些文章的基本信息售貌，包括創(chuàng)建時(shí)間、文章ID疫萤、文章類別ID等颂跨，同屬于一個(gè)column family，"article"扯饶。
查詢的場(chǎng)景則是查找"指定的時(shí)間范圍"恒削，"文章類型ID為N" 的所有文章數(shù)據(jù)。
根據(jù)以上場(chǎng)景帝际，設(shè)計(jì)如下 table：
- hbase table 為 article 蔓同。
- rowkey 是 "ARTICLE" + 微秒級(jí)時(shí)間戳（類似OpenTSDB 的rowkey，便于按時(shí)間序列查到某一段時(shí)間創(chuàng)建的 articles）蹲诀，即 "ARTICLE1504939752000000"斑粱。
- family 為 "basic"，包含 "ArticleID"脯爪， "ArticleTypeID"则北， "Created"矿微， 三個(gè) column。
查詢時(shí)通過(guò)指定 rowkey start 和 rowkey stop尚揣，可以 scan 某一個(gè)時(shí)間段的數(shù)據(jù)(因?yàn)?rowkey 中包含數(shù)值型的時(shí)間戳)涌矢，通過(guò) hbase filter 實(shí)現(xiàn)"ArticleTypeID" == N 的過(guò)濾條件。
開(kāi)始導(dǎo)入數(shù)據(jù)快骗、準(zhǔn)備查詢娜庇，以下是導(dǎo)入數(shù)據(jù)部分代碼 demo：
def save_batch_events(datas, table=None):
with get_connetion_pool().connection() as conn:
if table is not None:
t = conn.table(table)
else:
t = conn.table(TABLE)
b = t.batch(transaction=False)
for row, data in datas.items():
b.put(row, data)
b.send()
def save_main_v1():
datas = dict()
for i in range(100):
article_type_id = i % 2
timestamp = time.time() + i
rowkey = "ARTICLE" + str(timestamp * 1000000)
data = {
"basic:" + "ArticleID": str(i),
"basic:" + "ArticleTypeID": str(article_type_id),
"basic:" + "Created": str(timestamp),
}
datas[rowkey] = data
save_batch_events(datas)
查看一下 hbase 的數(shù)據(jù)，100 條數(shù)據(jù)全部正常導(dǎo)入方篮，其中50條數(shù)據(jù) "ArticleTypeID" 為0名秀，50條為1 ：
接下來(lái)就是用 hbase filter 過(guò)濾的過(guò)程了，假設(shè)查詢 "ArticleTypeID" 為 0 的數(shù)據(jù)藕溅，使用 java 客戶端實(shí)現(xiàn)查詢：
public static void test_hbase_filter() throws IOException {
TableName tableName = TableName.valueOf("test_article_1");
Configuration conf = HBaseConfiguration.create();
Connection conn = ConnectionFactory.createConnection(conf);
Table table = conn.getTable(tableName);
// Scan python table `test_article_1`
System.out.println("Prepare to scan !");
FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ONE);
SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("basic"),
Bytes.toBytes("ArticleTypeID"), CompareOp.EQUAL, Bytes.toBytes(1L));
list.addFilter(filter1);
Scan s = new Scan();
s.addFamily(Bytes.toBytes("basic"));
s.setFilter(list);
ResultScanner scanner = table.getScanner(s);
int num = 0;
for (Result rr = scanner.next(); rr != null; rr = scanner.next()) {
num++;
}
System.out.println("Found row: " + num);// 預(yù)期 50匕得，結(jié)果為 0
問(wèn)題出現(xiàn)：使用 java 期望的查詢結(jié)果為 50 條，但是查出的結(jié)果卻是 0 條!
使用 python 查詢卻可以得到正確的結(jié)果：
def recent_events_v1(start, end, table=None, filter_str=None, limit=2000):
with get_connetion_pool().connection() as conn:
if table is not None:
t = conn.table(table)
else:
t = conn.table(TABLE)
start_row = 'ARTICLE' + str(start * 1000000)
end_row = 'ARTICLE' + str(end * 1000000)
return t.scan(row_start=start_row, row_stop=end_row, filter=filter_str, limit=limit)
if __name__ == '__main__':
filter_str = "SingleColumnValueFilter('basic', 'ArticleTypeID', =, 'binary:1')"
results = recent_events_v1(start=0, end=1505023900, filter_str=filter_str)
print len([i for i in results]) # 期望值為50, 實(shí)際值為 50巾表，正確
尋找原因
經(jīng)過(guò) N 次確認(rèn)汁掠，java 的讀操作是沒(méi)有問(wèn)題的，python 實(shí)現(xiàn)的讀寫(xiě)也得到了預(yù)期的效果集币。進(jìn)一步探究考阱，特意用 java 完整的實(shí)現(xiàn)的數(shù)據(jù)的導(dǎo)入和查詢：
public static void test_hbase_filter1() throws IOException {
tableName = TableName.valueOf("test_article_java_1");
table = conn.getTable(tableName);
System.out.println("Prepare create table !");
Admin admin = conn.getAdmin();
if (!admin.tableExists(tableName)) {
HTableDescriptor td = new HTableDescriptor(tableName);
HColumnDescriptor basic = new HColumnDescriptor("basic");
td.addFamily(basic);
admin.createTable(td);
System.out.println("Created !");
}
// Put value to test_article_java_1
System.out.println("Prepare to write data to: " + table.getName().toString());
for (int i = 0; i < 100; i++) {
Put p = new Put(Bytes.toBytes("ARTICLE" + (System.currentTimeMillis() + 1000) * 1000));
p.addColumn(Bytes.toBytes("basic"), Bytes.toBytes("ArticleTypeID"), Bytes.toBytes(Long.valueOf(i % 2)));
table.put(p);
}
// scan test_article_java_1
scanner = table.getScanner(s);
num = 0;
for (Result rr = scanner.next(); rr != null; rr = scanner.next()) {
num++;
}
System.out.println("Found row: " + num);// 預(yù)期 50，結(jié)果為 50
}
可見(jiàn)惠猿，用 java 寫(xiě)的數(shù)據(jù)羔砾，用 java 讀是沒(méi)問(wèn)題的，用 python 寫(xiě)的數(shù)據(jù)用 python 讀也沒(méi)問(wèn)題偶妖。但 java 讀 python 寫(xiě)的數(shù)據(jù)就存在異常，難道是 python 寫(xiě)的數(shù)據(jù)和 java 寫(xiě)的數(shù)據(jù)不一樣政溃？為此分別對(duì)比一下 python 和 java 寫(xiě)入 hbase 的數(shù)據(jù)：
仔細(xì)觀察圖 1 和圖 2 中的數(shù)據(jù)可以發(fā)現(xiàn)趾访，python 寫(xiě)入的數(shù)據(jù)中對(duì)應(yīng)的 ArticleTypeID 值為 0 或 1，而 java 則是一串 bytes董虱。突然意識(shí)到一個(gè)問(wèn)題扼鞋，hbase 讀寫(xiě)的時(shí)候要求傳入的數(shù)據(jù)類型為 bytes，而使用 python 傳輸?shù)倪^(guò)程中這種整形數(shù)據(jù)是直接通過(guò) str() 方法轉(zhuǎn)成字符串存儲(chǔ)到 hbase 中的愤诱，并不是以 bytes 的形式存于 hbase云头，所以使用 java 用轉(zhuǎn)化成 bytes 的 filter 讀才沒(méi)能得到預(yù)期的結(jié)果。
正確的 filter 姿勢(shì)
既然找到了原因淫半，解決問(wèn)題就比較簡(jiǎn)單了溃槐，存儲(chǔ)的時(shí)候?qū)⒄蛿?shù)據(jù)全部都通過(guò) struct.pack 方法轉(zhuǎn)成 bytes 存入，這樣就可以被通用的查詢了科吭，同時(shí) 使用 python 查詢的時(shí)候也將 filter 中的整型數(shù)值替換成 bytes 格式昏滴。
使用 struct.pack 方法將整型轉(zhuǎn)成 bytes 時(shí)猴鲫，注意選擇使用 big-endian 的 Byte order，即 pack 方法的第一個(gè)參數(shù)使用 >谣殊。因?yàn)?java 官方 client 采用這種字節(jié)序拂共，下面是 Bytes.toBytes 的實(shí)現(xiàn)源碼，可見(jiàn)采用的是 big-endian：
/**
* Convert a long value to a byte array using big-endian.
*
* @param val value to convert
* @return the byte array
*/
public static byte[] toBytes(long val) {
byte [] b = new byte[8];
for (int i = 7; i > 0; i--) {
b[i] = (byte) val;
val >>>= 8;
}
b[0] = (byte) val;
return b;
寫(xiě)入的代碼：
def save_main_v2():
datas = dict()
for i in range(100):
article_type_id = i % 2
timestamp = time.time() + i
rowkey = "ARTICLE" + str(timestamp * 1000000)
data = {
"basic:" + "ArticleID": str(i),
"basic:" + "ArticleTypeID": struct.pack('>q', article_type_id),
"basic:" + "Created": str(timestamp),
}
datas[rowkey] = data
save_batch_events(datas, table="test_article_2")
查詢是的filter：
filter_str = "SingleColumnValueFilter('basic', 'ArticleTypeID', =, 'binary:{value}')".format(value=struct.pack('>q', 1))
這樣就沒(méi)有問(wèn)題了~
總結(jié)
使用 python 讀寫(xiě) hbase 數(shù)據(jù)姻几，直接傳輸整型參數(shù)時(shí)宜狐，hbase 的 thrift 接口會(huì)拋出 TDecodeException: Field 'value(3)' of 'Mutation' needs type 'STRING' 異常，被告知只接受 string 類型的數(shù)據(jù)蛇捌。這時(shí)注意將整型數(shù)據(jù)轉(zhuǎn)化成 bytes 形式的 str肌厨，而不要直接使用 str() 方法強(qiáng)轉(zhuǎn)，否則難以避免的會(huì)出現(xiàn)一些非預(yù)期的結(jié)果豁陆。
以為這樣就沒(méi)問(wèn)題了柑爸？ 請(qǐng)關(guān)注看下文~
