今天小编就为大家分享一篇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的方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持爱安网。

最新资讯
美参议员:举行听证会是耻辱 共和党一直在欺负科技公司CEO

美参议员:举行听证会是

针对美国参议院商务委员会今日举行的科技公司CEO听证
TCL科技:前三季度净利润20.25亿元,同比下降21.42%

TCL科技:前三季度净利

TCL科技发布三季报,前三季度净利润20.25亿元,同比下降21
iPhone 12除了性价比和 5G,它还剩下什么?

iPhone 12除了性价比

相信很多人都已经收到不久前订购的iPhone12,有些应该也
中兴通讯:第三季度实现净利润8.55亿元,同比下降67.83%

中兴通讯:第三季度实现

中兴通讯公告,第三季度实现净利润8.55亿元,同比下降67.8
中兴通讯:拟以发行股份方式购买中兴微电子18.8219%股权

中兴通讯:拟以发行股份

中兴通讯发布公告称,拟以发行股份方式购买恒健欣芯、汇
携程升级会员权益体系 年轻用户占比超7成

携程升级会员权益体系

携程近期宣布再次升级会员权益体系,全新的黑钻会员正式
最新文章
在pycharm中为项目导入anacodna环境的操作方法

在pycharm中为项目导

这篇文章主要介绍了在pycharm中为项目导入anacodna环
tensorflow的ckpt及pb模型持久化方式及转化详解

tensorflow的ckpt及pb

今天小编就为大家分享一篇tensorflow的ckpt及pb模型持
PyTorch笔记之scatter()函数的使用

PyTorch笔记之scatter

这篇文章主要介绍了PyTorch笔记之scatter()函数的使用
python3实现网页版raspberry pi(树莓派)小车控制

python3实现网页版ras

这篇文章主要为大家详细介绍了python3实现网页版raspb
完美解决pycharm导入自己写的py文件爆红问题

完美解决pycharm导入

今天小编就为大家分享一篇完美解决pycharm导入自己写
pycharm内无法import已安装的模块问题解决

pycharm内无法import

今天小编就为大家分享一篇pycharm内无法import已安装