Madhuvishy has uploaded a new change for review.
https://gerrit.wikimedia.org/r/225480
Change subject: [WIP] Parallelize eventlogging processor
......................................................................
[WIP] Parallelize eventlogging processor
- Using PyKafka BalancedConsumer
Bug: T104226
Change-Id: If3470ca80740e820c89432c390354ab2494fd2bf
---
M server/bin/eventlogging-processor
M server/eventlogging/handlers.py
M server/setup.py
3 files changed, 27 insertions(+), 14 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/mediawiki/extensions/EventLogging
refs/changes/80/225480/1
diff --git a/server/bin/eventlogging-processor
b/server/bin/eventlogging-processor
index 5e87fd6..823071d 100755
--- a/server/bin/eventlogging-processor
+++ b/server/bin/eventlogging-processor
@@ -125,4 +125,4 @@
else:
for w in writers_list:
- w.send(event)
+ w.send(event)
\ No newline at end of file
diff --git a/server/eventlogging/handlers.py b/server/eventlogging/handlers.py
index 28d5680..4879b8a 100644
--- a/server/eventlogging/handlers.py
+++ b/server/eventlogging/handlers.py
@@ -24,6 +24,7 @@
from kafka import SimpleProducer
from kafka.producer.base import Producer
from kafka.common import KafkaTimeoutError
+from pykafka import KafkaClient as PyKafkaClient
import logging
import logging.handlers
@@ -384,6 +385,7 @@
def kafka_reader(
path,
topic='eventlogging',
+ consumer_group='eventlogging-group',
identity='eventlogging',
raw=False,
**kafka_consumer_args
@@ -392,11 +394,14 @@
Reads events from Kafka.
Kafka URIs look like:
- kafka:///b1:9092,b2:9092?topic=topic_name&auto_commit_enable=True&...
+ kafka:///b1:9092,b2:9092?topic=topic_name&consumer_group=group_name&
+ auto_commit_enable=True&...
- This reader uses the kafka-python KafkaConsumer. You may pass
- any configs that KafkaConsumer takes as keyword arguments via
+ This reader uses the pykafka BalancedConsumer. You may pass
+ any configs that BalancedConsumer takes as keyword arguments via
the kafka URI query params.
+
+ FIXME: This stuff should be worked out for the BalancedConsumer:
If auto_commit_enable is True, then messages will be
marked as done as soon as they are read. This has the
@@ -408,29 +413,36 @@
will have to fix this problem somehow. Perhaps a callback?
"""
# Brokers should be in the uri path
- brokers = path.strip('/')
+ brokers = path.strip('/').encode('ascii', 'ignore')
+ # FIXME: Dont think this will work for Balanced Consumer
# remove non KafkaConsumer args from kafka_consumer_args
kafka_consumer_args = {
k: v for k, v in items(kafka_consumer_args)
- if k in kafka.DEFAULT_CONSUMER_CONFIG
+ if k in inspect.getargspec(PyKafkaClient.__init__).args
}
- consumer = KafkaConsumer(
- topic,
- group_id=identity,
- bootstrap_servers=brokers,
- **kafka_consumer_args
- )
+ kafka_client = PyKafkaClient(hosts=brokers)
+ kafka_topic = kafka_client.topics[topic]
+
+ consumer = kafka_topic.get_balanced_consumer(
+ consumer_group=consumer_group,
+ **kafka_consumer_args)
+
+ def get_message_stream(consumer):
+ while True:
+ yield consumer.consume()
# No need to bother calling task_done() if we aren't going to commit.
+ """
if consumer._config['auto_commit_enable']:
return stream(
(_ack_kafka_message(consumer, message) for message in consumer),
raw
)
else:
- return stream((message.value for message in consumer), raw)
+ """
+ return stream((message.value for message in get_message_stream(consumer)),
raw)
def _ack_kafka_message(consumer, message):
diff --git a/server/setup.py b/server/setup.py
index 8b1514b..d37903e 100644
--- a/server/setup.py
+++ b/server/setup.py
@@ -58,6 +58,7 @@
"pygments>=1.5",
"pyzmq>=2.1",
"sqlalchemy>=0.7",
- "kafka-python>=0.9.3"
+ "kafka-python>=0.9.3",
+ "pykafka>=1.0.3"
)
)
--
To view, visit https://gerrit.wikimedia.org/r/225480
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: If3470ca80740e820c89432c390354ab2494fd2bf
Gerrit-PatchSet: 1
Gerrit-Project: mediawiki/extensions/EventLogging
Gerrit-Branch: master
Gerrit-Owner: Madhuvishy <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
