当前位置:首页 > 技术文章 > 编程技术 > Python

超全的使用Python操作Kafka详细教程(2)

发布时间:2018-02-20 06:34:04 作者:匿名 来源:网络 点击量:
1、创建Produer
1)、命令行方式---普通的发送方式
[python] view plain copy
[root@node1 python_app]# python   
Python 2.6.6 (r266:84292, Nov 22 2013, 12:16:22)   
[GCC 4.4.7 20120313 (Red Hat 4.4.7-4)] on linux2  
Type "help", "copyright", "credits" or "license" for more information.  
>>> from kafka import KafkaProducer  
>>> producer = KafkaProducer(bootstrap_servers='192.168.120.11:9092')  
>>> for _ in range(100):  
...     producer.send('world',b'some_message_bytes')  
...   
<kafka.producer.future.FutureRecordMetadata object at 0xddb5d0>  
<kafka.producer.future.FutureRecordMetadata object at 0xddb750>  
<kafka.producer.future.FutureRecordMetadata object at 0xddb790>  
上面的几行功能分别是:
 
导入KafkaProducer
 
创建连接到192.168.120.11:9092这个Broker的Producer,
 
循环向world这个Topic发送100个消息,消息内容都是some_message_bytes',这种发送方式不指定Partition,kafka会均匀大把这些消息分别写入5个Partiton里面,
 
 
更详细的说明可以参考 https://kafka-python.readthedocs.io/en/master/index.html
 
 
 
2)、命令行方式---发送json字符串
json作为一种强大的文本格式,已经得到非常普遍的应用,kafak-python也支持发送json格式的消息
 
其实如果你参考https://kafka-python.readthedocs.io/en/master/index.html这里的KafkaProducer里面的发送json
 
 
 
一定会报错的,这应该是这个文档的一个bug,
 
[python] view plain copy
>>> producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'))  
Traceback (most recent call last):  
  File "<stdin>", line 1, in <module>  
  File "/usr/lib/python2.6/site-packages/kafka_python-1.3.4-py2.6.egg/kafka/producer/kafka.py", line 347, in __init__  
    **self.config)  
  File "/usr/lib/python2.6/site-packages/kafka_python-1.3.4-py2.6.egg/kafka/client_async.py", line 220, in __init__  
    self.config['api_version'] = self.check_version(timeout=check_timeout)  
  File "/usr/lib/python2.6/site-packages/kafka_python-1.3.4-py2.6.egg/kafka/client_async.py", line 841, in check_version  
    raise Errors.NoBrokersAvailable()  
kafka.errors.NoBrokersAvailable: NoBrokersAvailable  
经过测试,应该是这样写才是OK的
[python] view plain copy
>>> producer = KafkaProducer(bootstrap_servers='192.168.120.11:9092',value_serializer=lambda v: json.dumps(v).encode('utf-8'))  
>>> producer.send('world', {'key1': 'value1'})  
<kafka.producer.future.FutureRecordMetadata object at 0x2a9ebd0>  
>>>   
 
3)、命令行方式---发送普通字符串
[python] view plain copy
>>> producer.send('world', key=b'foo', value=b'bar')  
<kafka.producer.future.FutureRecordMetadata object at 0x29dcd90>  
>>>   
4)、命令行方式--发送压缩字符串
[python] view plain copy
>>> producer = KafkaProducer(bootstrap_servers='192.168.120.11:9092',compression_type='gzip')  
>>> producer.send('world', b'msg 1')  
 
经过测试这种方式发送的内容,在接收方收到的消息仍然是普通的字符串,也许是没有安装python-lz4,原文中有这样的内容:
kafka-python supports gzip compression/decompression natively. To produce or consume lz4 compressed messages, you should install python-lz4 (pip install lz4). To enable snappy, install python-snappy (also requires snappy library). See Installation for more information.
 
 
 
上面都是测试各个命令的使用,接下来,我们写一个完整的脚本,这个脚本的功能是把指定目录下的文件名发送到world这个topic
 
file_monitor.py脚本
 
[python] view plain copy
#-*- coding: utf-8 -*-  
  
from kafka import KafkaProducer  
import json  
import os  
import time  
from sys import argv  
  
producer = KafkaProducer(bootstrap_servers='192.168.120.11:9092')  
  
def log(str):  
    t = time.strftime(r"%Y-%m-%d_%H-%M-%S",time.localtime())  
    print("[%s]%s"%(t,str))  
  
def list_file(path):  
    dir_list = os.listdir(path);  
    for f in dir_list:  
         producer.send('world',f)  
         producer.flush()  
         log('send: %s' % (f))    
  
list_file(argv[1])  
producer.close()  
log('done')  
 
 
假如我们要监控/opt/jdk1.8.0_91/lib/missioncontrol/features这个目录下的文件,可以这样执行
 
python file_monitor.py  /opt/jdk1.8.0_91/lib/missioncontrol/features
 
执行结果如下:
 
[python] view plain copy
[root@node2 python_app]# python file_monitor.py  /opt/jdk1.8.0_91/lib/missioncontrol/features  
[2017-11-07_17-41-04]send: org.eclipse.ecf.filetransfer.ssl.feature_1.0.0.v20140827-1444  
[2017-11-07_17-41-04]send: org.eclipse.emf.common_2.10.1.v20140901-1043  
[2017-11-07_17-41-04]send: com.jrockit.mc.feature.rcp.ja_5.5.0.165303  
[2017-11-07_17-41-04]send: com.jrockit.mc.feature.console_5.5.0.165303  
[2017-11-07_17-41-04]send: org.eclipse.ecf.core.feature_1.1.0.v20140827-1444  
[2017-11-07_17-41-04]send: org.eclipse.equinox.p2.core.feature_1.3.0.v20140523-0116  
[2017-11-07_17-41-04]send: org.eclipse.ecf.filetransfer.httpclient4.ssl.feature_1.0.0.v20140827-1444  
[2017-11-07_17-41-04]send: com.jrockit.mc.feature.rcp_5.5.0.165303  
[2017-11-07_17-41-04]send: org.eclipse.babel.nls_eclipse_zh_4.4.0.v20140623020002  
[2017-11-07_17-41-04]send: com.jrockit.mc.rcp.product_5.5.0.165303  
[2017-11-07_17-41-04]send: org.eclipse.help_2.0.102.v20141007-2301  
[2017-11-07_17-41-04]send: org.eclipse.ecf.core.ssl.feature_1.0.0.v20140827-1444  
[2017-11-07_17-41-04]send: org.eclipse.ecf.filetransfer.httpclient4.feature_3.9.1.v20140827-1444  
[2017-11-07_17-41-04]send: org.eclipse.e4.rcp_1.3.100.v20141007-2033  
[2017-11-07_17-41-04]send: org.eclipse.babel.nls_eclipse_ja_4.4.0.v20140623020002  
[2017-11-07_17-41-04]send: com.jrockit.mc.feature.flightrecorder_5.5.0.165303  
[2017-11-07_17-41-04]send: org.eclipse.emf.ecore_2.10.1.v20140901-1043  
[2017-11-07_17-41-04]send: org.eclipse.equinox.p2.rcp.feature_1.2.0.v20140523-0116  
[2017-11-07_17-41-04]send: org.eclipse.ecf.filetransfer.feature_3.9.0.v20140827-1444  
[2017-11-07_17-41-04]send: com.jrockit.mc.feature.core_5.5.0.165303  
[2017-11-07_17-41-04]send: org.eclipse.rcp_4.4.0.v20141007-2301  
[2017-11-07_17-41-04]send: com.jrockit.mc.feature.rcp.zh_CN_5.5.0.165303  
[2017-11-07_17-41-04]done  
 
 
 
在consumer上看到的内容是这样:
 
[plain] view plain copy
world:4:93: key=None value=org.eclipse.ecf.filetransfer.ssl.feature_1.0.0.v20140827-1444  
world:1:112: key=None value=org.eclipse.emf.common_2.10.1.v20140901-1043  
world:3:119: key=None value=com.jrockit.mc.feature.console_5.5.0.165303  
world:1:113: key=None value=com.jrockit.mc.feature.rcp.ja_5.5.0.165303  
world:0:86: key=None value=org.eclipse.ecf.core.feature_1.1.0.v20140827-1444  
world:1:114: key=None value=org.eclipse.equinox.p2.core.feature_1.3.0.v20140523-0116  
world:4:94: key=None value=org.eclipse.ecf.filetransfer.httpclient4.ssl.feature_1.0.0.v20140827-1444  
world:0:87: key=None value=com.jrockit.mc.feature.rcp_5.5.0.165303  
world:4:95: key=None value=org.eclipse.babel.nls_eclipse_zh_4.4.0.v20140623020002  
world:2:66: key=None value=com.jrockit.mc.rcp.product_5.5.0.165303  
world:4:96: key=None value=org.eclipse.ecf.core.ssl.feature_1.0.0.v20140827-1444  
world:2:67: key=None value=org.eclipse.help_2.0.102.v20141007-2301  
world:1:115: key=None value=org.eclipse.ecf.filetransfer.httpclient4.feature_3.9.1.v20140827-1444  
world:4:97: key=None value=org.eclipse.e4.rcp_1.3.100.v20141007-2033  
world:0:88: key=None value=org.eclipse.babel.nls_eclipse_ja_4.4.0.v20140623020002  
world:4:98: key=None value=com.jrockit.mc.feature.flightrecorder_5.5.0.165303  
world:3:120: key=None value=org.eclipse.emf.ecore_2.10.1.v20140901-1043  
world:1:116: key=None value=org.eclipse.equinox.p2.rcp.feature_1.2.0.v20140523-0116  
world:4:99: key=None value=org.eclipse.ecf.filetransfer.feature_3.9.0.v20140827-1444  
world:2:68: key=None value=com.jrockit.mc.feature.core_5.5.0.165303  
world:4:100: key=None value=com.jrockit.mc.feature.rcp.zh_CN_5.5.0.165303  
world:2:69: key=None value=org.eclipse.rcp_4.4.0.v20141007-2301  
 
 
2、创建Consumer
 
通常使用Kafka时会创建不同的Topic,并且在Topic里面创建多个Partiton,因此作为Consumer,通常是连接到指定的Broker,指定的Topic来消费消息。
 
完整的python 脚本
 
consumer.py
 
[python] view plain copy
#-*- coding: utf-8 -*-  
  
from kafka import KafkaConsumer  
import time  
  
def log(str):  
        t = time.strftime(r"%Y-%m-%d_%H-%M-%S",time.localtime())  
        print("[%s]%s"%(t,str))  
  
log('start consumer')  
#消费192.168.120.11:9092上的world 这个Topic,指定consumer group是consumer-20171017  
consumer=KafkaConsumer('world',group_id='consumer-20171017',bootstrap_servers=['192.168.120.11:9092'])  
for msg in consumer:  
        recv = "%s:%d:%d: key=%s value=%s" %(msg.topic,msg.partition,msg.offset,msg.key,msg.value)  
        log(recv)  
 
 
重新启动file_monitor.py脚本
 
[python] view plain copy
[root@node2 python_app]# python file_monitor.py  /opt/jdk1.8.0_91/lib/missioncontrol/features  
[2017-11-07_18-00-31]send: org.eclipse.ecf.filetransfer.ssl.feature_1.0.0.v20140827-1444  
[2017-11-07_18-00-31]send: org.eclipse.emf.common_2.10.1.v20140901-1043  
[2017-11-07_18-00-31]send: com.jrockit.mc.feature.rcp.ja_5.5.0.165303  
[2017-11-07_18-00-31]send: com.jrockit.mc.feature.console_5.5.0.165303  
[2017-11-07_18-00-31]send: org.eclipse.ecf.core.feature_1.1.0.v20140827-1444  
[2017-11-07_18-00-31]send: org.eclipse.equinox.p2.core.feature_1.3.0.v20140523-0116  
[2017-11-07_18-00-31]send: org.eclipse.ecf.filetransfer.httpclient4.ssl.feature_1.0.0.v20140827-1444  
[2017-11-07_18-00-31]send: com.jrockit.mc.feature.rcp_5.5.0.165303  
[2017-11-07_18-00-31]send: org.eclipse.babel.nls_eclipse_zh_4.4.0.v20140623020002  
[2017-11-07_18-00-31]send: com.jrockit.mc.rcp.product_5.5.0.165303  
[2017-11-07_18-00-31]send: org.eclipse.help_2.0.102.v20141007-2301  
[2017-11-07_18-00-31]send: org.eclipse.ecf.core.ssl.feature_1.0.0.v20140827-1444  
[2017-11-07_18-00-31]send: org.eclipse.ecf.filetransfer.httpclient4.feature_3.9.1.v20140827-1444  
[2017-11-07_18-00-31]send: org.eclipse.e4.rcp_1.3.100.v20141007-2033  
[2017-11-07_18-00-31]send: org.eclipse.babel.nls_eclipse_ja_4.4.0.v20140623020002  
[2017-11-07_18-00-31]send: com.jrockit.mc.feature.flightrecorder_5.5.0.165303  
[2017-11-07_18-00-31]send: org.eclipse.emf.ecore_2.10.1.v20140901-1043  
[2017-11-07_18-00-31]send: org.eclipse.equinox.p2.rcp.feature_1.2.0.v20140523-0116  
[2017-11-07_18-00-31]send: org.eclipse.ecf.filetransfer.feature_3.9.0.v20140827-1444  
[2017-11-07_18-00-31]send: com.jrockit.mc.feature.core_5.5.0.165303  
[2017-11-07_18-00-31]send: org.eclipse.rcp_4.4.0.v20141007-2301  
[2017-11-07_18-00-31]send: com.jrockit.mc.feature.rcp.zh_CN_5.5.0.165303  
[2017-11-07_18-00-31]done  
 
然后启动consumer.py脚本
 
[python] view plain copy
[root@node1 python_app]# python consumer.py   
[2017-09-23_11-34-00]start consumer  
[2017-09-23_11-34-10]world:3:121: key=None value=org.eclipse.ecf.filetransfer.ssl.feature_1.0.0.v20140827-1444  
[2017-09-23_11-34-10]world:2:70: key=None value=org.eclipse.emf.common_2.10.1.v20140901-1043  
[2017-09-23_11-34-10]world:3:122: key=None value=com.jrockit.mc.feature.rcp.ja_5.5.0.165303  
[2017-09-23_11-34-10]world:2:71: key=None value=com.jrockit.mc.feature.console_5.5.0.165303  
[2017-09-23_11-34-10]world:0:89: key=None value=org.eclipse.ecf.core.feature_1.1.0.v20140827-1444  
[2017-09-23_11-34-10]world:4:101: key=None value=org.eclipse.equinox.p2.core.feature_1.3.0.v20140523-0116  
[2017-09-23_11-34-10]world:1:117: key=None value=org.eclipse.ecf.filetransfer.httpclient4.ssl.feature_1.0.0.v20140827-1444  
[2017-09-23_11-34-10]world:2:72: key=None value=com.jrockit.mc.feature.rcp_5.5.0.165303  
[2017-09-23_11-34-10]world:4:102: key=None value=org.eclipse.babel.nls_eclipse_zh_4.4.0.v20140623020002  
[2017-09-23_11-34-10]world:2:73: key=None value=com.jrockit.mc.rcp.product_5.5.0.165303  
[2017-09-23_11-34-10]world:3:123: key=None value=org.eclipse.help_2.0.102.v20141007-2301  
[2017-09-23_11-34-10]world:3:124: key=None value=org.eclipse.ecf.core.ssl.feature_1.0.0.v20140827-1444  
[2017-09-23_11-34-10]world:0:90: key=None value=com.jrockit.mc.feature.flightrecorder_5.5.0.165303  
[2017-09-23_11-34-10]world:3:125: key=None value=org.eclipse.ecf.filetransfer.httpclient4.feature_3.9.1.v20140827-1444  
[2017-09-23_11-34-10]world:3:126: key=None value=org.eclipse.e4.rcp_1.3.100.v20141007-2033  
[2017-09-23_11-34-10]world:3:127: key=None value=org.eclipse.babel.nls_eclipse_ja_4.4.0.v20140623020002  
[2017-09-23_11-34-10]world:2:74: key=None value=org.eclipse.emf.ecore_2.10.1.v20140901-1043  
[2017-09-23_11-34-10]world:3:128: key=None value=org.eclipse.equinox.p2.rcp.feature_1.2.0.v20140523-0116  
[2017-09-23_11-34-10]world:0:91: key=None value=com.jrockit.mc.feature.core_5.5.0.165303  
[2017-09-23_11-34-10]world:3:129: key=None value=org.eclipse.ecf.filetransfer.feature_3.9.0.v20140827-1444  
[2017-09-23_11-34-11]world:3:130: key=None value=org.eclipse.rcp_4.4.0.v20141007-2301  
[2017-09-23_11-34-11]world:3:131: key=None value=com.jrockit.mc.feature.rcp.zh_CN_5.5.0.165303  
 
可以看到file_monitor.py脚本发送了一批文件名到word这个topic,并且consumer.py收到了这些文件名。

本文链接:http://www.topjishu.net/article/94.html,如有转载,请注明出处!


关键词:Python,Kafka


发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表