kafka-python consumer used in streamparse spout not work, and throw timeout #447
Comments
|
you can use from streamparse import Spout
from pykafka import KafkaClient
class WordSpout(Spout):
outputs = ['word']
def initialize(self, stormconf, context):
client = KafkaClient(hosts="c1:9092,c1_1:9092,c1_2:9092")
topic = client.topics['test'.encode('utf-8')]
self.balanced_consumer = topic.get_balanced_consumer(
consumer_group=b"test_group",
auto_commit_enable=True,
zookeeper_connect="c1:2181,c1_1:2181,c1_2:2181"
)
self.reg = self.regex()
def next_tuple(self):
message = self.balanced_consumer.consume()
info = message.value.decode('utf-8')
self.logger.info('==================={}'.format(info))
word = next(self.words)
self.emit([message]) |
|
i have found that it is just right to throw the exceptino when the kafka producer dost not push data intervally. it works without exception when kafka producer push data every second. |
how do you solve it? |
1 similar comment
how do you solve it? |
|
@jhhnjhhn how do you solve it ? |
|
This code will be fix the issue: from streamparse import Spout
from kafka import KafkaConsumer
class PySpout(Spout):
outputs = ['word']
def initialize(self, stormconf, context):
self.consumer = KafkaConsumer('test', group_id='my-group', bootstrap_servers=['localhost:9092'], consumer_timeout_ms=10)
def next_tuple(self):
try:
message = self.consumer.next_v1()
self.emit([message.value])
except:
pass |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
pyspout.py:
The text was updated successfully, but these errors were encountered: