pykafka常见问题分析与记录

柔光的暖阳◎ 2022-06-09 05:54 449阅读 0赞

python 操作kafka主要有两个库，pykafka和kafka-python, 网上有些文章说明两者的差异。如果不是特殊的需求，估计基本的操作两者都能满足，因为用过pymysql，所以这里就选择一个pykafka开坑吧。

官方文档：  
[http://pykafka.readthedocs.io/en/latest][http_pykafka.readthedocs.io_en_latest]  
github:  
[https://github.com/Parsely/pykafka][https_github.com_Parsely_pykafka]

私以为这个文档写的对中国人不怎么友好啊。反正我是看的一脸懵逼，上面的demo一些也敲不过去。好多问题还是从github的issue上一点点挖出来结果。

## 安装 ##

pip install pykafka

安装基本没有什么幺蛾子，等一会就安装好了。

## kafka配置 ##

操作kafka相信你看到这篇文章，基本的环境是有了。  
说下我的环境，一台机器安装了kafka和zookeeper(也可以用kafka自带的zookeeper)。  
另一台机器远程生产或消费

如果在安装机器上搞，问题会少一点，如果是在远程的机器上生产和消费，对kafka的配置就要多一点。

不少文章都表示，需要修改/etc/hosts文件，好让kafka解析到对应的ip地址上。老版本可能需要这个操作，至少我没配置也正常使用了。

vim config/server.properties

advertised.listeners=PLAINTEXT://xxx.xxx.xxx.xxx:9092

修改kafka 的server配置文件。  
xxx.xxx.xxx.xxxy一定要填入具体的ip地址，我直接填了0.0.0.0导致后面远程连接一直出错。

修改后重启下服务。

还有些外网访问的配置，端口开放等，这里不细说，那是另一个话题。

我们还是重点说说pykafka的坑。

## 连接问题 ##

建议先用ipython 进去敲几个试试手，好，大家跟着我的步骤来，  
进入 ipython (或者其他解释器)

from pykafka import KafkaClient
    client = KafkaClient(hosts="ip:9092")
    print client.topics

应该输出的是你的kafka中的topic。  
但是如果你看到下面的类似信息，就有问题了

No handlers could be found for logger "pykafka.broker"
    NoBrokersAvailableError: Unable to connect to a broker to fetch metadata. See logs

首先保证连接不上不是服务没开，或者你以为开了，实际kafka本身启动有问题，我们的一切前提都是在kafka正常工作的基础之上。

在解释器输入命令打开调试模式

import logging as log
    log.basicConfig(level=log.DEBUG)

再次执行

from pykafka import KafkaClient
    client = KafkaClient(hosts="ip:9092")
    print client.topics

输出的log中有几句

INFO:pykafka.cluster:Discovered 1 brokers
    DEBUG:pykafka.cluster:Discovered broker id 0: 0.0.0.0:9092
    DEBUG:pykafka.connection:Connecting to 0.0.0.0:9092
    INFO:pykafka.connection:Failed to connect to 0.0.0.0:9092
    WARNING:pykafka.broker:Failed to connect newly created broker for 0.0.0.0:9092

连接broker失败。就是我说的kafka的server.properties中不能设置0.0.0.0必须为一个具体的ip地址。

参考的资料  
[https://github.com/Parsely/pykafka/issues/418][https_github.com_Parsely_pykafka_issues_418]  
[https://github.com/Shopify/sarama/issues/754][https_github.com_Shopify_sarama_issues_754]

## 脚本执行错误 ##

我们解决了上面的问题，直接在解释器执行下

from pykafka import KafkaClient
    
    client = KafkaClient(hosts="ip:9092")
    topic=client.topics['wk']
    producer = topic.get_producer()
    producer.produce('test')

此时你在安装kafka的机器上订阅该topic就可以看到输出了test文本。

那我们搞到脚本中去，

python test.py

ReferenceError: weakly-referenced object no longer exists

报错了，

> Thanks very much for reporting this. Your finding confirms \#352 - which said I had a hunch that this kind of error would be possible. The reason it works fine in the interpreter is that there you keep the producer instance alive, whereas in the scripted run the producer may have been finalised before the worker thread gets to empty the queue.

因为在解释器能运行的正常因为，解释器保证了生产者的存活，而脚本中执行完就退出了，队列中的内容却没有发送完成

解决办法

from pykafka import KafkaClient
    
    client = KafkaClient(hosts="ip:9092")
    topic=client.topics['wk']
    producer = topic.get_producer()
    producer.produce('test')
    producer.stop()

在最后添加stop()退出生产者实例。

参考  
[https://github.com/Parsely/pykafka/issues/422][https_github.com_Parsely_pykafka_issues_422]  
[https://github.com/Parsely/pykafka/issues/352][https_github.com_Parsely_pykafka_issues_352]

## 性能提升 ##

都没问题写入还是有些慢。  
参考文章  
[http://blog.csdn.net/liuxingen/article/details/71746218][http_blog.csdn.net_liuxingen_article_details_71746218]

这里需要安装librdkafka

这个我使用了好几种方法，晚上也比较多的方法。  
所以最后我也不确定是哪种方法成功的。

pip install librdkafka  
pip install librdkafka1

还有的需要编译的。都尝试下。

[http_pykafka.readthedocs.io_en_latest]: http://pykafka.readthedocs.io/en/latest
[https_github.com_Parsely_pykafka]: https://github.com/Parsely/pykafka
[https_github.com_Parsely_pykafka_issues_418]: https://github.com/Parsely/pykafka/issues/418
[https_github.com_Shopify_sarama_issues_754]: https://github.com/Shopify/sarama/issues/754
[https_github.com_Parsely_pykafka_issues_422]: https://github.com/Parsely/pykafka/issues/422
[https_github.com_Parsely_pykafka_issues_352]: https://github.com/Parsely/pykafka/issues/352
[http_blog.csdn.net_liuxingen_article_details_71746218]: http://blog.csdn.net/liuxingen/article/details/71746218