Hello! I’m new to Pulsar and I’d like to build a M...
# generalx
Xianle Wang
11/25/2025, 11:07 PMHello! I’m new to Pulsar and I’d like to build a Map-Reduce like pipeline using Pulsar Functions in Python:
producer -> (submit raw messages without key) -> Topic A -> (Shared subscription) -> “Key Function” to assign key for each record -> (submit messages with keys) -> Topic B -> (key_shared subscription) -> “Compute Function” to process records
The key requirements are:
1. The records of the same key should be processed in sequence and in micro batch
2. Different keys should be processed in parallel
3. Low end2end latency: around 100ms
4. High throughput: 1 million messages / second
Questions:
1. Can “Compute Function” process a (micro) batch of records at a time. It seems like Pulsar Function doesn’t support batch input?
2. If batch input is not supported, is it a common/good practice to buffer the inputs inside function instance? What's the latency like to persist every record (1KB) to the BookKeeper?
3. Is it possible to create multiple threads within a function instance to handle different keys? Is it anti-pattern?
Thanks!
👀 1
d
David K
11/26/2025, 5:12 PMPulsar functions written in Java support windowing, which can be used to gather a pre-defined number of messages before the function logic is triggered. A simple guide is available in the documentation. If you must use Python, consider creating a feature request to add this capability to Python-based functions as well.
x
Xianle Wang
11/26/2025, 5:17 PMThanks for your reply. Yeah I have to use Python. If I maintain a buffer inside the function instance, do you know what's the latency like to persist every record (1KB) to the BookKeeper?
d
David K
11/26/2025, 5:23 PMI wouldn’t advise using BookKeeper as a low-latency database. In my experience, it is not well-suited for frequent writes.
x
Xianle Wang
11/26/2025, 5:24 PMGot it, thanks!
Xianle Wang
12/08/2025, 6:10 PMHi @David K
I read from StreamNative doc
State storage is only available for Java functions for now.Does it mean I can't enable state storage for Python functions?
d
David K
12/08/2025, 7:01 PMThe docs might be wrong, as this PR indicates that support for state was added to Python functions in 2018.
David K
12/08/2025, 7:10 PM Copy code
from pulsar import Function
class StatefulFunction(Function):
def process(self, input, context):
# Increment a counter
context.incr_counter("message_count", 1)
# Get the current counter value
count = context.get_counter("message_count")
context.get_logger().info(f"Processed {count} messages")
# Store arbitrary state (value must be bytes)
context.put_state("last_message", input.encode('utf-8'))
# Retrieve state
last_msg = context.get_state("last_message")
# Delete state if needed
# context.del_counter("some_key")
return inputx
Xianle Wang
12/08/2025, 7:18 PMI was trying to use
context.put_state(key, value)context.get_state(key)context.get_state(key)"enableStateStore": false,d
David K
12/08/2025, 7:37 PMYes, we don’t have it enabled on our serverless product. State storage is available in private preview. To test it out, you will need to use our BYOC (Bring Your Own Cloud) deployment model. Please let me know, and we can provide you with some complimentary credits to try it out.
x
Xianle Wang
12/09/2025, 12:49 AMBefore setting everything up, I'd like to estimate the performance in this design. Do you know what's the avg latency for
put_stateget_state🤔 1
d
David K
12/09/2025, 2:15 AMBased on my research, I found benchmark data specifically for Apache Pulsar Functions’ state storage, which uses the Apache BookKeeper Table Service (State Store) as its backend. From benchmarks presented at Pulsar Summit NA 2021, the avg latency is ~1.7 ms for put_state, and ~8.8 ms for get_state. Writes are inherently slower because they require journaling and replication to multiple bookies for durability.
gratitude thank you 1
David K
12/09/2025, 2:16 AMScreenshot 2025-12-08 at 18.16.13.png