Python 通过thrift接口连接Hbase读取存储数据

待我称王封你为后i 2023-09-25 11:51 106阅读 0赞

原创 2013年03月14日 23:23:41

标签：thrift

/Hbase

/python

/并行拓展

/failover

介绍：

Hbase：开源的分布式数据库

资料介绍:http://www.oschina.net/p/hbase

Thrift:一个软件框架，用来进行可扩展且跨语言的服务的开发。最初由Facebook开发，作为Hadoop的一个工具，提供跨语言服务开发；

资料介绍：http://dongxicheng.org/search-engine/thrift-guide/

官方使用手册：http://download.csdn.net/detail/wyjzt999/5141006从安装到使用都很全面

我们项目里客户端是用python开发，因此需要Thrift提供server端，经过thrift对Hbase进行数据读写操作，性能非常不错，并且可以在Hadoop集群上做并行拓展，稳定性高，Facebook内部通信也是采用thrift来做；

首先学习一下Hbase的表结构：

Row Key

Row key行键 (Row key)可以是任意字符串(最大长度是 64KB，实际应用中长度一般为 10-100bytes)，在hbase内部，row key保存为字节数组。

列族 (column family)

hbase表中的每个列，都归属与某个列族。列族是表的chema的一部分(而列不是)，必须在使用表之前定义。列名都以列族作为前缀。例如courses:history ， courses:math 都属于 courses 这个列族。

时间戳

HBase中通过row和columns确定的为一个存贮单元称为cell。每个 cell都保存着同一份数据的多个版本。版本通过时间戳来索引。时间戳的类型是 64位整型。时间戳可以由hbase(在数据写入时自动 )赋值，此时时间戳是精确到毫秒的当前系统时间。时间戳也可以由客户显式赋值。如果应用程序要避免数据版本冲突，就必须自己生成具有唯一性的时间戳。每个

cell中，不同版本的数据按照时间倒序排序，即最新的数据排在最前面。

为了避免数据存在过多版本造成的的管理 (包括存贮和索引)负担，hbase提供了两种数据版本回收方式。一是保存数据的最后n个版本，二是保存最近一段时间内的版本(比如最近七天)。用户可以针对每个列族进行设置。

对Hbase而言，表结构设计会对系统的性能以及开销上造成很大的区别;

?1.首先建立与thriftserver端的连接

\[python\]

view plain

copy

from?thrift?import?Thriftfrom?thrift.transport?import?TSocket,?TTransportfrom?thrift.protocol?import?TBinaryProtocolfrom?hbase?import?Hbase\#server端地址和端口transport?=?TSocket.TSocket(host,?port)\#可以设置超时transport.setTimeout(5000)\#设置传输方式(TFramedTransport或TBufferedTransport)trans?=?TTransport.TBufferedTransport(transport)\#设置传输协议protocol?=?TBinaryProtocol.TBinaryProtocol(trans)\#确定客户端client?=?Hbase.Client(protocol)\#打开连接transport.open()

2.然后就可以做具体的操作，比如查表，删表，插入Row Key等等

\[python\]

view plain

copy

from?hbase.ttypes?import?ColumnDescriptor,?Mutation,?BatchMutation,?TRegionInfofrom?hbase.ttypes?import?IOError,?AlreadyExists\#获取表名client.getTableNames()\#创建新表\_TABLE?=?"keyword"

\[python\]

view plain

copy

demo?=?ColumnDescriptor(name='data:',maxVersions?=?10)\#列族data能保留最近的10个数据，每个列名后面要跟:号createTable(\_TABLE,?\[demo\])

\[python\]

view plain

copy

\#创建列名2个data:url?data:wordtmp1=?\[Mutation(column="data:url",?value="www.baidu.com")\]tmp2=?\[Mutation(column="data:word",?value="YaGer")\]\#新建2个列?(表名，行键，?列名)client.mutateRow(\_TABLE,?row,?tmp1)client.mutateRow(\_TABLE,?row,?tmp1)

\[python\]

view plain

copy

\#从表中取数据\#通过最大版本数取数据client.getByMaxver(\_TABLE,'00001','data:word',?10)\#一次取10个版本\#取列族内数据client.getColumns(\_TABLE,?'00001')

3.支持thrift并行拓展和失效转移Failover机制

\[python\]

view plain

copy

\#file?name：hbaseconn.py\#!?/usr/bin/env?python\#?-\*-?coding:?utf-8?-\*-\#提供建立连接的方法和取数据操作的方法import?loggingimport?tracebackimport?time,sysfrom?unittest?import?TestCase,?mainimport?socketfrom?thrift?import?Thriftfrom?thrift.transport?import?TSocketfrom?thrift.transport?import?TTransportfrom?thrift.protocol?import?TBinaryProtocolfrom?hbase?import?Hbasefrom?hbase.ttypes?import?IOError?as?HbaseIOError,?AlreadyExistsfrom?hbase.ttypes?import?ColumnDescriptor,?Mutation,?BatchMutation\_TABLE?=?'thrift\_check'\_ROW?=?'test\_for\_thrift'\_DATA?=?'data:word'\_VALUE?=?'Flag'class?DbConnError(Exception):"""Conn?Db?exception.?Timeout?or?any?except?for?connection?from?low?layer?api?"""passclass?Connection:def?\_\_init\_\_(self,?trans,?client,?addr,?port):self.trans?=?transself.client?=?clientself.hp?=?addrself.port?=?portpassclass?CenterDb:@classmethoddef?open(cls,?host\_port):cls.tc\_list?=?\[\]for?hp?in?host\_port:trans,?client?=?cls.\_open(\*hp)cls.tc\_list.append(Connection(trans,?client,?hp\[0\],?hp\[1\]))return?cls.tc\_list?@classmethoddef?\_open(cls,?host,?port):transport?=?TSocket.TSocket(host,?port)transport.setTimeout(5000)trans?=?TTransport.TBufferedTransport(transport)protocol?=?TBinaryProtocol.TBinaryProtocol(trans)client?=?Hbase.Client(protocol)ok?=?Falsetry:trans.open()ok?=?Trueexcept?TSocket.TTransportException,?e:logerr('CenterDb(Hbase)?Open?error(%s,?%d)'?%?(host,?port))ok?=?Falseelse:pass\#dbg('CenterDb?connected?(%s,?%d)'?%?(host,?port))return?trans,?client?@classmethoddef?\_initTable(cls,?client):dat?=?ColumnDescriptor(name?=?'data',?maxVersions?=?1)tmp?=?\[Mutation(column?=?\_DATA,?value?=?\_VALUE)\]try:?client.createTable(\_TABLE,?\[dat\])?client.mutateRow(\_TABLE,?\_ROW,?tmp)?dbg("Create?Table?For?Thrift?Test?Success!")?returnTrueexcept?AlreadyExists:return?Truereturn?False?@classmethoddef?\_get(cls,?client):client.getVer(\_TABLE,\_ROW,\_DATA,?1)return?True?@classmethoddef?\_reconnect(cls,?trans):trans.close()trans.open()def?\_\_init\_\_(self,?transport,?client):self.t?=?transportself.c?=?clientdef?\_\_str\_\_(self):return?'CenterDb.open(%s,?%d)'?%?(self.t,?self.c)def?\_\_del\_\_(self):self.t.close()def?getColumns(self,?table,?row):tr\_list?=?\[\]tr\_list?=?self.\_failover('getRow',?table,?row)if?(not?tr\_list):return?\{\}return?tr\_list\[0\].columns?

\[python\]

view plain

copy

\#file?name:?thriftmanage.py\#!?/usr/bin/python\#?-\*-?coding:?utf-8?-\*-\#提供管理thrift连接的方法\#建一个线程循环检测thrift的可使用性，非连接池import?loggingimport?time,sys,randomimport?threadingfrom?hbase?import?Hbasefrom?hbase.ttypes?import?IOError?as?HbaseIOErrorfrom?hbase.ttypes?import?ColumnDescriptor,?Mutation,?BatchMutationimport?hbaseconnfrom?hbaseconn?import?CenterDbclass?Failover(threading.Thread):def?\_\_init\_\_(self,serverid,?serverlist):threading.Thread.\_\_init\_\_(self)self.serverlist?=?serverlistself.serverid?=?serveridself.lenlist?=?len(serverlist)self.conn\_list?=?\[\]self.invalid\_con\_list?=?\[\]'''''?cur\_conn?:?now?oper?is?usingmy\_conn?:?this?schd?server?should?use?\[(trans1,client1),(trans2,client2),...\]'''self.cur\_conn?=?Noneself.my\_conn?=?Noneself.\_makeConn()self.\_getConn()self.\_init\_for\_check()self.switched\_flag?=?Falsedef?\_makeConn(self):'''''make?all?the?conntion?to?thrift?serverlist?'''self.conn\_list?=?CenterDb.open(self.serverlist)def?\_getConn(self):'''''get?only?one?connection?from?the?conn\_list,confirm?cur\_conn?and?my\_conn?'''self.cur\_conn?=?self.my\_conn?=?self.conn\_list\[(int(self.serverid)?%?self.lenlist)\]?self.other\_conn?=?self.conn\_list\[:\]self.other\_conn.remove(self.my\_conn)return?Truedef?\_init\_for\_check(self):'''''init?\_TABLLE?test\_for\_flag'''try:if?not?CenterDb.\_initTable(?self.my\_conn.client?):dbg("Error?In?Create?Table?For?Thrift?Test!")except?Exception,?e:dbg("init\_for\_check?thrift:%s"?%?e)'''''make?the?only?conn?for?check?proc?'''self.check\_conn?=?CenterDb.open(?((self.my\_conn.hp,?self.my\_conn.port),))\[0\]def?\_switch(self):'''''when?my\_conn?failed,?choose?the?other?client?randomly;when?my\_conn?is?reset?OK,?cur\_conn?will?use?my\_conn?again?'''\#print?'Schd%s?come?in?\_switch'?%?self.serveridif?0?==?len(self.other\_conn):return?Falsetrycount?=?0while?True:try:if?trycount?==?3\*self.lenlist?:\#try?3\*length?timesreturn?Falsetmp\_conn?=?random.choice(self.other\_conn)\#CenterDb.\_reconnect(tmp\_conn.trans)\#DEBUGif?self.\_checker(tmp\_conn):self.cur\_conn?=?tmp\_conndbg('Schd%s?\_switch?cur\_conn:?%s'?%?(self.serverid,?self.cur\_conn.hp))return?Trueelse:trycount?+=?1logerr('Schd%s?\_switch?for?%d?times'?%?(self.serverid,?trycount))CenterDb.\_reconnect(tmp\_conn.trans)\#close?this?trans?and?try?again;breakdown?earlyif?self.\_checker(tmp\_conn):self.cur\_conn?=?tmp\_connreturn?Trueelse:continue\#can't?be?usedexcept?Exception,?e:continuedef?\_failover(self,oper,?\*args,?\*\*kwargs):result?=?\[\]try?:result?=?getattr(self.cur\_conn.client,?oper)(\*args,?\*\*kwargs)return?resultexcept?HbaseIOError,?e:?logerr("\_failover?:?%s?"?%?e)except?Exception,?e:logerr(?'Schd%s?\_failover?:?Connect?to?%s?thrift?Server?closed!?Choose?another......Reason:%s?'?%?\\(self.serverid,?self.cur\_conn.hp,?e))self.cur\_conn.trans.close()self.switched\_flag?=?Trueif?self.\_switch():logerr('Schd%s?\_failover?:?Now?using?%s?server'?%?(self.serverid,?self.cur\_conn.hp))result?=?getattr(self.cur\_conn.client,?oper)(\*args,?\*\*kwargs)else:?logerr(?'Schd%s?\_failover?:?Switch?3?rotate?Find?No?Healthy?Thrift?server?!'?%?self.serverid)return?resultdef?getByMaxver(self,?table,?row,?column,?max\_ver):"""Get?cell?list?by?ver?Get?cell?list?no?more?than?max?versions?Args:?table:?table?name?row:?row?key?column:?column?name?max\_ver:?max?version?to?retrieve?cell?"""cell\_list?=?\[\]start\_time?=?time.time()cell\_list?=?self.\_failover('getVer',?table,?row,?column,?max\_ver)if?(not?cell\_list):return?\[\]take?=?time.time()?-?start\_timeif?take?>?0.015:logerr('Hbase?over?15ms:take?%s?ms'?%?str("%.3f"?%?(take\*1000)))return?map(lambda?x:?(x.value,?x.timestamp),?cell\_list)def?getColumns(self,?table,?row):tr\_list?=?\[\]tr\_list?=?self.\_failover('getRow',?table,?row)if?(not?tr\_list):return?\{\}return?tr\_list\[0\].columnsdef?\_checker(self,conn):'''''check?my\_conn?,?be?sure?it?is?connected?and?get?data?OK?'''try:if?conn.trans.isOpen():CenterDb.\_get(conn.client)return?Trueelse:?return?False?except?Exception,?e:logerr(?'Schd%s\_checker?:?Connect?to?%s?closed!?Please?Restart?now.....?Reason:?%s?'?%?\\(self.serverid,?conn.hp,?e))return?Falsedef?\_restart(self):'''''if?my\_conn?failed,?restart?it's?trans'''while?True:\#print?'come?in?\_restart?%s?'?%?self.my\_conn.hpself.my\_conn.trans.close()self.check\_conn.trans.close()try:time.sleep(2)self.my\_conn.trans.open()self.check\_conn.trans.open()if?self.my\_conn.trans.isOpen()?and?self.check\_conn.trans.isOpen():CenterDb.\_get(self.check\_conn.client)\#self.check\_conn.client.getVer('keywordurl','test\_for\_thrift','data:word',?1)return?Trueelse:?continueexcept?Exception,?e:logerr('Schd%s?\_restart?:?Connect?to?%s?is?not?restart?yet?...?Reason:%s?'?%?\\(self.serverid,?self.my\_conn.hp,?e))self.my\_conn.trans.close()self.check\_conn.trans.close()continuedef?run(self):while?True:time.sleep(1)if?self.\_checker(self.check\_conn)?and?(not?self.switched\_flag):continueelse:if?self.\_restart():logerr(?'Schd%s?Connection?from%s?to?my:%s?Recovered?!?'?%?\\(self.serverid,self.cur\_conn.hp,self.my\_conn.hp))self.cur\_conn?=?self.my\_connself.switched\_flag?=?Falsecontinue

使用thrift生成的代码中提供的方法有：

void enableTable(Bytes tableName)

void disableTable(Bytes tableName)

bool isTableEnabled(Bytes tableName)

void compact(Bytes tableNameOrRegionName)

void majorCompact(Bytes tableNameOrRegionName)

getTableNames()

getColumnDescriptors(Text tableName)

getTableRegions(Text tableName)

void createTable(Text tableName, columnFamilies)

void deleteTable(Text tableName)

get(Text tableName, Text row, Text column)

getVer(Text tableName, Text row, Text column, i32 numVersions)

getVerTs(Text tableName, Text row, Text column, i64 timestamp, i32 numVersions)

getRow(Text tableName, Text row)

getRowWithColumns(Text tableName, Text row, ?columns)

getRowTs(Text tableName, Text row, i64 timestamp)

getRowWithColumnsTs(Text tableName, Text row, ?columns, i64 timestamp)

getRows(Text tableName, ?rows)

getRowsWithColumns(Text tableName, ?rows, ?columns)

getRowsTs(Text tableName, ?rows, i64 timestamp)

getRowsWithColumnsTs(Text tableName, ?rows, ?columns, i64 timestamp)

void mutateRow(Text tableName, Text row, ?mutations)

void mutateRowTs(Text tableName, Text row, ?mutations, i64 timestamp)

void mutateRows(Text tableName, ?rowBatches)

void mutateRowsTs(Text tableName, ?rowBatches, i64 timestamp)

i64 atomicIncrement(Text tableName, Text row, Text column, i64 value)

void deleteAll(Text tableName, Text row, Text column)

void deleteAllTs(Text tableName, Text row, Text column, i64 timestamp)

void deleteAllRow(Text tableName, Text row)

void deleteAllRowTs(Text tableName, Text row, i64 timestamp)

ScannerID scannerOpenWithScan(Text tableName, TScan scan)

ScannerID scannerOpen(Text tableName, Text startRow, ?columns)

ScannerID scannerOpenWithStop(Text tableName, Text startRow, Text stopRow, ?columns)

ScannerID scannerOpenWithPrefix(Text tableName, Text startAndPrefix, ?columns)

ScannerID scannerOpenTs(Text tableName, Text startRow, ?columns, i64 timestamp)

ScannerID scannerOpenWithStopTs(Text tableName, Text startRow, Text stopRow, ?columns, i64 timestamp)

scannerGet(ScannerID id)

scannerGetList(ScannerID id, i32 nbRows)

void scannerClose(ScannerID id)

?参考：http://yannramin.com/2008/07/19/using-facebook-thrift-with-python-and-hbase/

\[python\]

view plain

copy

Python 通过thrift接口连接Hbase读取存储数据

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Python 通过thrift接口连接Hbase读取存储数据

相关 Redis 安装、启动、连接、存储和读取数据

相关用Python3 操作之HBase-Thrift

相关用Python操作HBase之HBase-Thrift

相关 python 通过thrift 简单操作hbase

相关 HBase thrift2 TIOError

相关 PHP通过Thrift操作Hbase

相关 Spark读取Hbase数据

相关 python hbase API (二) thrift2

相关 Hbase thrift-server