Batch consumer ?

[gwthm-in]

Is there a way where I can get batch of messages instead of a single message.

What I'm trying to achieve is to get batch of messages and convert them to Python dataframe using from_json, and then do some operations in the dataframe and store the parquet files to s3.

The problem here is I can't just wait make the dataframe from only few messages. It has to be atleast 2L messages.

For this I can't iterate over 2L times, and poll the consume to get one by one message and store it and then process further.

Instead if there is a way where we can tell the consumer poll to return next 100 messages, it will be good. So it should return 100 messages if 100 messages are available or the n number of messages that are available.

I think this is general use case where most of the time we don't process message by message but a batch of messages.

[edenhill]

Feature request noted.

In the meantime you could make your own batching wrapper for poll().

[fsck-mount]

@edenhill when can we expect this? Or can you guide me with some resources to implement that myself in C?
I need this approach for one of my production optimisation. Current way is taking too much time like 2000s, with this one, I can bought this time under 20s - 50s.

[mrocklin]

Does anyone have a sense for what the performance would be for a batching solution? Do we know what the current bottleneck is within confluent-kafka-python or librdkafka?

This blogpost gives performance numbers for confluent-kafka-python at around 200k msgs/sec or 25MB/s.

This is certainly better than other Python Kafka libraries, but also not yet saturating typical ethernet bandwidth within datacenters.

[mrocklin]

@Gowtham-Sai JSON parsing speeds are only slightly better than this rate (~50MB/s?). It's not clear to me how much benefit you're likely to get from speeding up kafka bandwidth here.

[fsck-mount]

@mrocklin

This is certainly better than other Python Kafka libraries, but also not yet saturating typical ethernet bandwidth within datacenters.

Yep, you are correct. Confluent Kafka speed is better than any other python library now.

Do we know what the current bottleneck is within confluent-kafka-python or librdkafka?

As per my knowledge, poll is implemented in librdkafka, if this is the case, calling poll every time is worst when you want to fetch say 2,00,000 msgs. This means, we need to iterate over 2,00,000 like below..

from confluent_kafka import Consumer, KafkaError

c = Consumer({'bootstrap.servers': 'mybroker', 'group.id': 'mygroup',
              'default.topic.config': {'auto.offset.reset': 'smallest'}})
c.subscribe(['mytopic'])
running = True

def process_and_convert_parquet(batch_msgs):
	df = pd.DataFrame(batch_msgs)
	fastparquet.write(df, 'parqetcon.parquet');

msgs = []
while running:
    msg = c.poll()
    if msg:
	    if not msg.error():
	        msgs.append(msg.value().decode('utf8'))
	    elif msg.error().code() != KafkaError._PARTITION_EOF:
	        print(msg.error())
	        running = False
	if len(msgs) > (10 ^ 5):
		process_and_convert_parquet(msgs)
		msgs = []
c.close()

The above code is obviously not efficient. In case if you have batch consumer like below. It is more efficient as already messages are fetched and we are not creating the overhead of iteration for single message.

from confluent_kafka import Consumer, KafkaError

c = Consumer({'bootstrap.servers': 'mybroker', 'group.id': 'mygroup',
              'default.topic.config': {'auto.offset.reset': 'smallest'}})
c.subscribe(['mytopic'])
running = True

def process_and_convert_parquet(batch_msgs):
	df = pd.DataFrame(batch_msgs)
	fastparquet.write(df, 'parqetcon.parquet');


while running:
    msgs = c.poll(msg_count=100000, timeout=10) # Assuming msgs contains n (say 5k) msgs only as it can fetch max of n before it times out. 
    process_and_convert_parquet(msgs) 
	
c.close()

[edenhill]

As a side note: librdkafka 0.11.0 has a severe performance degradation with the Consumer and certain workloads (small batches), confluentinc/librdkafka#1384

A fix is available on librdkafka master branch and we're working on a new release.

I suggest you try out the fixed version to get some proper throughput numbers.

[mrocklin]

I'm confused, does poll have a msg_count= keyword? I'm not seeing this in either latest release or git-master.

[mrocklin]

@edenhill I think I'm still pinned against 0.9.4

mrocklin@carbon:~$ conda list | grep librdkafka
librdkafka                0.9.4                         0    conda-forge

https://github.com/conda-forge/librdkafka-feedstock/tree/master/recipe

[fsck-mount]

@mrocklin

I'm confused, does poll have a msg_count= keyword? I'm not seeing this in either latest release or git-master.

Sorry for the confusion. poll doesn't have msg_count. I've written that assuming if such implementation is available (Batch consumer). Or may be similar new api implementation - rd_kafka_consume_batch_queue.

from confluent_kafka import Consumer, KafkaError

c = Consumer({'bootstrap.servers': 'mybroker', 'group.id': 'mygroup',
              'default.topic.config': {'auto.offset.reset': 'smallest'}})
c.subscribe(['mytopic'])
running = True

def process_and_convert_parquet(batch_msgs):
	df = pd.DataFrame(batch_msgs)
	fastparquet.write(df, 'parqetcon.parquet');


while running:
    msgs = c.consume_batch(msg_count=100000, timeout=10) # Assuming msgs contains n (say 5k) msgs only as it can fetch max of n before it times out. 
    process_and_convert_parquet(msgs) 
	
c.close()

[gwthm-in]

Awesome 👍

[ikucan]

https://github.com/ikucan/pykafarr