You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2023/05/09 01:34:10 UTC
[GitHub] [pulsar] Samreay created a discussion: [Python Client] How to subscribe to two schema's topics?
GitHub user Samreay created a discussion: [Python Client] How to subscribe to two schema's topics?
Hi all, I'm trying to write up a simple proof-of-concept application, wherein all it does it subscribe to two separate topics, each with their own different AvroSchema. So far, this doesn't seem possible to do easily with the (synchronous) python client, because each `Consumer` only supports one type of schema, and a `consumer.receive()` blocks.
Does anyone know a nice solution for this?
GitHub link: https://github.com/apache/pulsar/discussions/20263
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] yaalsn added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Posted by GitBox <gi...@apache.org>.
GitHub user yaalsn added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
If pydantic, how about using `JSON` instead of `Avro`, you can convert the models to `JSON` easily
https://docs.pydantic.dev/latest/usage/exporting_models/
GitHub link: https://github.com/apache/pulsar/discussions/20263#discussioncomment-5880223
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] Samreay edited a comment on the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Posted by GitBox <gi...@apache.org>.
GitHub user Samreay edited a comment on the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
What I've come up with doesn't seem optimal, but it works by hiding away multiple blocking synchronous code blocks into threads and then putting the messages received into an janus.Queue (asyncio.Queues are not thread safe, so we swapped to janus). This all works under FastAPI to support the existing REST endpoints we serve.
```python
class Consumer(Thread):
def __init__(self, queue: janus.Queue, topic: str, schema: pulsar.schema.AvroSchema):
super().__init__()
self.queue = queue
self.pulsar_client = pulsar.Client("pulsar://pulsar:6650")
self.topic = topic
self.schema = schema
def run(self):
reader = self.pulsar_client.create_reader(
self.topic, start_message_id=pulsar.MessageId.latest, schema=self.schema
)
while True:
msg = reader.read_next()
topic = msg.topic_name()
if "/" in topic:
topic = topic.split("/")[-1]
message = from_avro(msg.value())
self.queue.sync_q.put_nowait(message)
```
Then when the FastAPI app starts you can kick off the threads, and also launch an `async` method to get from the `asyncio.Queue` passed into the Consumer threads.
```python
app = FastAPI()
app.queue = asyncio.Queue()
@app.on_event("startup")
async def startup_event():
consumer_preds = Consumer(app.queue, "predictions", Prediction.schema())
consumer_preds.start()
consumer_trades = Consumer(app.queue, "trades", Trade.schema())
consumer_trades.start()
asyncio.create_task(receive_message())
```
`Prediction.schema` and what not are just some wrappers we have to convert pydantic to avro schemas.
GitHub link: https://github.com/apache/pulsar/discussions/20263#discussioncomment-5845219
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] Samreay edited a discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Posted by GitBox <gi...@apache.org>.
GitHub user Samreay edited a discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Hi all, I'm trying to write up a simple proof-of-concept application, wherein all it does it subscribe to two separate topics, each with their own different AvroSchema. So far, this doesn't seem possible to do easily with the (synchronous) python client, because each `Consumer` only supports one type of schema, and a `consumer.receive()` blocks.
Does anyone know a nice solution for this?
GitHub link: https://github.com/apache/pulsar/discussions/20263
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] Samreay added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Posted by GitBox <gi...@apache.org>.
GitHub user Samreay added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
What I've come up with doesn't seem optimal, but it works by hiding away multiple blocking synchronous code blocks into threads and then putting the messages received into an asynchio.Queue. This all works under FastAPI to support the existing REST endpoints we serve.
```python
class Consumer(Thread):
def __init__(self, queue: asyncio.Queue, topic: str, schema: pulsar.schema.AvroSchema):
super().__init__()
self.queue = queue
self.pulsar_client = pulsar.Client("pulsar://pulsar:6650")
self.topic = topic
self.schema = schema
def run(self):
queue: asyncio.Queue = self.queue
reader = self.pulsar_client.create_reader(
self.topic, start_message_id=pulsar.MessageId.latest, schema=self.schema
)
while True:
msg = reader.read_next()
topic = msg.topic_name()
if "/" in topic:
topic = topic.split("/")[-1]
message = from_avro(msg.value())
queue.put_nowait(message)
```
Then when the FastAPI app starts you can kick off the threads, and also launch an `async` method to get from the `asyncio.Queue` passed into the Consumer threads.
```python
app = FastAPI()
app.queue = asyncio.Queue()
@app.on_event("startup")
async def startup_event():
consumer_preds = Consumer(app.queue, "predictions", Prediction.schema())
consumer_preds.start()
consumer_trades = Consumer(app.queue, "trades", Trade.schema())
consumer_trades.start()
asyncio.create_task(receive_message())
```
`Prediction.schema` and what not are just some wrappers we have to convert pydantic to avro schemas.
GitHub link: https://github.com/apache/pulsar/discussions/20263#discussioncomment-5845219
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] yaalsn added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Posted by GitBox <gi...@apache.org>.
GitHub user yaalsn added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Not JSON schema, I mean sending JSON string, just bytes. I think this can solve your problem quickly.
GitHub link: https://github.com/apache/pulsar/discussions/20263#discussioncomment-5880284
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] Samreay added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Posted by GitBox <gi...@apache.org>.
GitHub user Samreay added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Doesn't this have the same issue as Avro, which is if I want to enforce schema on the topics with a well defined producer and consumer, it looks like the synchronous single-schema Consumer in the API means we'd need to turn to multithreading or multiprocessing? Does the JSON schema allow us some way around this?
GitHub link: https://github.com/apache/pulsar/discussions/20263#discussioncomment-5880245
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
[GitHub] [pulsar] Samreay added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Posted by GitBox <gi...@apache.org>.
GitHub user Samreay added a comment to the discussion: [Python Client] How to subscribe to two topics - both with different schemas?
Oh yeah, if we just accept json and do the schema validation ourself, no issues. I was hoping to make use of Pulsars schema registry features and how it enforces things to guarantee a consumer with an incompatible version of the schema can't connect. Figured if the pulsar devs have put in the hard work to implement all of that, I should make use of it! (There's also no rush on our side, the nasty threading approach works, but its very nice.
GitHub link: https://github.com/apache/pulsar/discussions/20263#discussioncomment-5880316
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org