AMQP, MQTT, NATS, etc.

[amotl]

About

While omniload already supports Kafka and Kinesis, it doesn't support message broker protocols like AMQP, MQTT, or NATS yet. We were thinking of using LorryStream/streamz, but now discovered mq-bridge written in Rust by @marcomq, which also provides Python bindings per mq-bridge-py package. 💯

Details

mq-bridge supports a number of backends: Kafka, NATS, AMQP (RabbitMQ), MQTT, MongoDB, SQL Databases (PostgreSQL, MySQL, SQLite via sqlx), HTTP, WebSocket, ZeroMQ, Files, AWS (SQS/SNS), IBM MQ, and in-memory channels.

References

• Considering alternatives daq-tools/lorrystream#231

[marcomq]

Hi! Thx for mentioning mq-bridge. I mostly wrote it for myself and it mostly was focussed on rust and batch performance. The python wrapper is kind of new.

There is already python "publish" and I recently exposed "pull". So most would probably already be there for streams and queues:
https://github.com/marcomq/mq-bridge/tree/main/python/mq-bridge-py#pull-based-consumer

It's an interesting approach here. Let me know if I can help adding code.

[amotl]

Hi Marco,

thanks for your swift reply. We are just adding one more source+target nozzle with GH-109, which you could use as a blueprint for orientation purposes. We certainly love any contributions, omniload is still in its infancy. Please also note this reference ticket, which was written when ingestr v0 was still using Python:

• Subscribe to streaming sources (continuous ingest) bruin-data/ingestr#282

In this spirit, while ingestr now provides support for streaming transfers, omniload does not. It would be nice to fill that gap with your nice mq-bridge-py package. ¹

With kind regards,
Andreas.

Footnotes

1. Note: While dlt provides pipeline components for Kafka and Kinesis, for example, they are not continuous: They just read the whole stream once and that's it. In this spirit, we haven't evaluated at all whether dlt itself is suitable for that job to process streaming data well. What we do know from the patch Connectors: Add source and target adapters for Delta Lake #109, is that it is super easy to interface with many data types, like the newly added Delta Lake source which just yields Polars data frames, that's it. ↩

[amotl]

Notes

We haven't evaluated at all whether dlt itself is suitable for that job to process streaming data well.

Both Pandas and Arrow allow streaming records in batches.

Why should it not work?

References:

• sources/scraping/queue.py
• arrow-pandas.md

[amotl]

Idea

In the spirit of LorryStream, we can imagine omniload to provide a similar slim interface like outlined on LorryStream's MQTT and AMQP documentation pages.

lorry relay \
    "mqtt://localhost/testdrive/%23" \
    "crate://localhost/?table=testdrive"

Trivia

I don't know if the --source-table and --dest-table options of omniload are applicable in such a scenario, especially when relaying data from MQTT to AMQP, for example. --dest-table is already defined optional, maybe we need to do the same with --source-table.

Edit: I've submitted a quick patch that makes--source-table optional, so any table option no longer gets in the way if it's not needed.

• API: Option --source-table becomes optional for streaming sources #119

[marcomq]

I just added some draft PR:

• Add mq-bridge #125

It would add NATS/AMQP/MQTT/ZeroMQ/AWS SQS and an alternative Kafka support that commits the source after target commit.
I added an +mqb to all urls. Not sure if this is necessary (except for kafka) - I used the delta lake example as orientation and it also had this in the urls. It all still needs a cleanup, maybe other code modifications and I some deeper manual test, but I already added the draft PR, just for some early feedback. The kafka test already seems to run fine also needs some fixes.

I didn't add mongodb or sqlx yet, as the current implementation has much wider and flexible data format support. Mq-bridge mostly only supports json and some internal format there.

Also, I found some documentation about amqp support in docs/supported-sources/amqp.md, which was probably a leftover of ingestr. Can/should this be replaced?

[amotl]

That was fast, thank you; very much appreciated. I think the implementation looks very good already, and I acknowledge all your decisions. Maybe we can also humbly ask @hampsterx to have a second look to discover anything that needs improvement?

What I can't find is any amqp.md in ingestr v0.x nor ingestr head, so yes, please replace ;].

[amotl]

Hi @marcomq,

thank you so much for swiftly submitting a patch that brings mq-bridge to the project. 💯

• Add mq-bridge #125

We've released your improvements with omniload v0.5.0. Now, with Amazon SQS, IBM MQ, MQTT, NATS, and ZeroMQ excellently documented, the documentation only lacks a dedicated page about AMQP, right?

With kind regards,
Andreas.

[amotl]

I see. I am only now seeing that AMQP is shed under the RabbitMQ page. Note to self: Let's improve the situation with GH-58 latest.

[marcomq]

Yes - I added the amqp in the rabbitmq page.
Thx for mentioning mq-bridge 😄. I hope it is useful here. Let me know if you have issues or ideas for improvement.

[amotl]

Hi again. I am currently renaming s/rabbitmq/amqp/, to provide better jargon. While I found corresponding documentation about an eventual "native" AMQP connector, there isn't any implementation at all. In this spirit, let's make the amqp+mqb handler the new canonical / default?

• See Documentation: s/RabbitMQ/AMQP/ #154.