python hbase读取数据发送kafka的方法
程序员文章站
2022-03-20 15:05:53
本例子实现从hbase获取数据,并发送kafka。
使用
#!/usr/bin/env python
#coding=utf-8
import sys...
本例子实现从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从文件中读取数据的方法讲解
-
对python3 Serial 串口助手的接收读取数据方法详解
-
spark读取hbase数据速度(spark读取hbase提速的方法)
-
使用python读取.text文件特定行的数据方法
-
Python读取txt文件数据的方法(用于接口自动化参数化数据)
-
python读取json文件并将数据插入到mongodb的方法
-
Python实现读取SQLServer数据并插入到MongoDB数据库的方法示例
-
python读取excel指定列数据并写入到新的excel方法
-
Python简单实现TCP包发送十六进制数据的方法
-
python使用socket向客户端发送数据的方法