python hbase读取数据发送kafka的方法

本例子实现从hbase获取数据，并发送kafka。

使用

#!/usr/bin/env python
#coding=utf-8

import sys
import time
import json

sys.path.append('/usr/local/lib/python3.5/site-packages')
from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocol
from hbase1 import Hbase #调用hbase thrif1
from hbase1.ttypes import *
from kafka import KafkaConsumer
from kafka import KafkaProducer
from kafka.errors import KafkaError
import unittest

class HbaseOpreator:
def __init__(self,host,port,table='test'):
self.tableName=table
self.transport=TTransport.TBufferedTransport(TSocket.TSocket(host,port))
self.protocol=TBinaryProtocol.TBinaryProtocol(self.transport)
self.client=Hbase.Client(self.protocol)
self.transport.open()

def __del__(self):
self.transport.close()

def scanTablefilter(self,table,*args):
d=dict()
L=[]
try:
tableName=table
# scan = Hbase.TScan(startRow, stopRow)
scan=TScan()
#主键首字母123
# filter = "PrefixFilter('123_')"
# filter = "RowFilter(=,'regexstring:.aaa')"
#过滤条件，当前为 statis_date 字段，值为20170223
# fitler = "SingleColumnValueFilter(tableName,'f','statis_date','20170223')"
# filter="SingleColumnValueFilter('f','statis_date',=,'binary:20170223') AND SingleColumnValueFilter('f','name',=,'binary:LXS')"
filter="SingleColumnValueFilter('info','name',=,'binary:lilei') OR SingleColumnValueFilter('info','name',=,'binary:lily')"
scan.filterString=filter
id=self.client.scannerOpenWithScan(tableName,scan,None)
result=self.client.scannerGet(id)
# result=self.client.scannerGetList(id,100)
while result:
for r in result:
key=r.row
name=r.columns.get('info:name').value
age=r.columns.get('info:age').value
phone=r.columns.get('info:phone').value
d['key']=key
d['name']=name
d['age']=age
d['phone']=phone
# encode_result_json=json.dumps(d).encode(encoding="utf-8")
# print(encode_result_json)
L.append(d)
result=self.client.scannerGet(id)
return json.dumps(L).encode(encoding="utf-8")
finally:
# self.client.scannerClose(scan)
print("scan finish")

def sendKfafkaProduct(data):
# self.host_port='localhost:9092'
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
for d in data:
producer.send('test', key=b'lxs', value=d)
time.sleep(5)
print(d)

while True:
producer.send('test', key=b'lxs', value=data)
time.sleep(5)
print(data)

if __name__== '__main__':
# unittest.main()

B=HbaseOpreator('10.27.1.138',9090)
value=B.scanTablefilter('ns_lbi:test_hbase_student')
print(value)
#sendKfafkaProduct(value)

以上这篇python hbase读取数据发送kafka的方法就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

• 在python环境下运用kafka对数据进行实时传输的方法
• kafka-python批量发送数据的实例
• 对python操作kafka写入json数据的简单demo分享
• python消费kafka数据批量插入到es的方法