Scrapy Clusters kafka_monitor.py 中断

Question

Scrapy Clusters 是一种扩展大型、连续的 scrapy 项目的绝佳新方法。它在 VM 中设置，并与来自不同工具的多个实例一起运行。

我已经构建了虚拟机并安装了必要的工具，例如 Redis、Kafka 和 zookeeper。

在我第一次测试爬行之前，我只是停留在一条简单的线上，我不知道这是 python 问题、安装问题还是什么。

当我尝试时

python kafka_monitor.py run

我明白了：

Traceback（最近一次调用最后一次）：文件“kafka_monitor.py”，行 512，在 sys.exit(main())
文件“kafka_monitor.py”，第 497 行，在 main kafka_monitor.run()
运行中的文件“kafka_monitor.py”，第 413 行 self._setup_kafka()
_setup_kafka 中的文件“kafka_monitor.py”，第 232 行 ret_val = _hidden_​​setup()
文件“/usr/local/lib/python2.7/dist-packages/scutils/method_timer.py”，行 46，在 f2 retval = f(*args)
_hidden_​​setup 中的文件“kafka_monitor.py”，第 218 行 self.kafka_conn = KafkaClient(self.settings['KAFKA_HOSTS'])

TypeError: ____init___() 只需要 1 个参数（给定 2 个）

settings.py 包括：

KAFKA_HOSTS = 'localhost:9092'
KAFKA_INCOMING_TOPIC = 'demo.incoming'
KAFKA_GROUP = 'demo-group'
KAFKA_FEED_TIMEOUT = 5
KAFKA_CONN_TIMEOUT = 5

按照Quick Start 的指示，我也做了一个localsettings.py

# Here, 'scdev' is the host with Kafka, Redis, and Zookeeper
REDIS_HOST = 'scdev'
KAFKA_HOSTS = 'scdev:9092'
ZOOKEEPER_HOSTS = 'scdev:2181'

即使我进入 kafka_monitor.py 并更改第 218 行（从上面的初始错误回溯）

发件人：

self.kafka_conn = KafkaClient(self.settings['KAFKA_HOSTS'])

收件人：

self.kafka_conn = KafkaClient(self.settings['scdev:9092'])

我收到同样的错误。

Answer 1

成功了！

@Madison Bahmer 感谢您引导我朝着正确的方向前进。我想澄清一下我是如何让它运行的。

kafka_monitor.py 中实际代码的第一行（即第 3 行）为：

from kafka.client import KafkaClient

我改成：

from kafka import KafkaClient

这让它工作了！

这是一个短视的解决方案吗？如果没有：

from kafka.client import KafkaClient

我不知道，但如果出现特定于此的问题，我会在以后更新。

Answer 2

虽然我不确定您的环境是什么，但我强烈建议您遵循简介here 中的快速入门指南，该指南教您如何设置和测试您的 repo 克隆，甚至为您提供虚拟机运行并测试一切。我似乎无法复制您的问题，但这里有一些我针对 kafka-python（支持）和 pykafka（不支持）进行的测试。

这里，scdev 只是一个运行 Kafka 的虚拟机。使用 Python 2.7.10。

kafka-python 0.9.5（固定在 requirements.txt 中）：

>>> from kafka import KafkaClient
>>> k = KafkaClient()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: __init__() takes at least 2 arguments (1 given)
>>> k = KafkaClient('scdev:9092')
>>> k.ensure_topic_exists('test')
>>>

请注意，你不能在没有参数的情况下调用 KafkaClient() 构造函数，所以我实际上不知道你是如何让它工作的。

pykafka 2.2.1（pypi 上的最新版本）

>>> from pykafka import KafkaClient
>>> k = KafkaClient()
>>> k.ensure_topic_exists()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'KafkaClient' object has no attribute 'ensure_topic_exists'
>>> k2 = KafkaClient('scdev:9092')
>>> 

请注意，现在可以使用了，但是我们使用的是完全不同的库，并且出现了相同的属性错误。我不知道您是否更改了包导入，但我似乎在项目中找不到任何 reference 到 pykafka。

kafka-python 1.0.1（pypi 上的最新版本）

>>> from kafka import KafkaClient
>>> k = KafkaClient()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/madisonb/.local/share/virtualenvs/sc3/lib/python2.7/site-packages/kafka/__init__.py", line 41, in __init__
    super(KafkaClient, self).__init__(*args, **kwargs)
TypeError: __init__() takes at least 2 arguments (1 given)
>>> k = KafkaClient('scdev:9092')
>>> k.ensure_topic_exists('test')
>>>

只是三重检查它是否适用于最新的 pypi 包。

还需要注意的是，问题中引用的变量settings 只是设置文件中元素的字典。所以调用self.settings['scdev:9092'] 没有实际意义，因为该密钥不存在。真的，self.settings['KAFKA_HOSTS'] 只是一个字典查找，似乎是字符串'scdev:9092'

编辑：

我可以使用 kafka_monitor.py 提供的语法和 Kafka-Python 1.0.1 重现 __init__() 错误。

>>> from kafka.client import KafkaClient # new syntax for importing KafkaClient
>>> k = KafkaClient()
>>> k.ensure_topic_exists('test')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'KafkaClient' object has no attribute 'ensure_topic_exists'
>>> k2 = KafkaClient('scdev:9092')
Exception AttributeError: "'KafkaClient' object has no attribute '_wake_r'" in <bound method KafkaClient.__del__ of <kafka.client_async.KafkaClient object at 0x102c51d50>> ignored
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: __init__() takes exactly 1 argument (2 given)
>>>

您安装的版本似乎与 requirements.txt 文件不匹配。这只会给你带来更多问题，我建议pip install -r requirements.txt 看看是否能解决问题。