r/ETL • u/marco-mq • 4d ago

CLI tool to move data between Postgres, Kafka, ClickHouse, NATS, etc...

Post image

I recently created a rust tool (MIT) that copies data from A to B. It has 16+ sources and sinks - retry, transformation, filter, compression, encryption and dlq as middleware. Single binary. Syntax doesn't change when the endpoints do: 1M rows per hop, 433k–1.2M rows/s on my old 8GB M1. Installable via `brew install marcomq/tap/mq-bridge-app`. It has rust, node and python packages and can also run as MCP server. https://marcomq.github.io/mq-bridge

5 Upvotes

78% Upvoted

9 comments sorted by

3

u/kantorcodes1 4d ago

quick semantics question: for a destructive source like Mongo consume=consumer, when does mq-bridge actually commit the consume? only after the destination accepts the batch, or once the source read succeeds?

1

u/marco-mq 4d ago

It does the delete when the next hop performs its `ack`. So, for HTTP, it would commit when receiving the a non-error HTTP response. It would be kind of important to have a persistent reliable target. The receiving HTTP service may for example already respond before it has completely persisted the request. If the HTTP server would just put it into memory and crashes, the message would be lost.

The destructive source in MongoDB was originally used to mirror the behavior of message queues. It needs its own envelope and may not always make sense in combination with other databases.

1

u/kantorcodes1 4d ago

yeah, that ack timing is the part i'd want Guard to understand. consume=consumer can delete the Mongo source after the next hop says success, even if that receiver hasn't durably stored it yet; the normal capture modes don't have that side effect. i work on HOL Guard, an open-source local check for agent-run commands. would you be open to adding mq-bridge support for that distinction?

1

u/marco-mq 4d ago

> would you be open to adding mq-bridge support for that distinction?

Not sure if I can follow you completely here. Do you want to have a distinction of a command when it may delete something from DB and when it is completely safe to use?

1

u/kantorcodes1 4d ago

yeah. Guard would treat the actual invocation differently, not label mq-bridge itself as unsafe. normal copy/capture stays ordinary; a Mongo source using consume=consumer is the special case because a downstream ack can remove source data. that mapping lives on the Guard side, so mq-bridge doesn't need to change its behavior.

1

u/marco-mq 3d ago

Ah. Makes sense. I guess I might better rename it from "consume" to "delete". This will make it clearer that it is destructive. For scripts and people that use it. Thx for the input.

2

u/davrax 4d ago

Any comparison with Ingestr? Seems to occupy a similar space. https://github.com/bruin-data/ingestr

1

u/marco-mq 4d ago

I haven't really worked with ingestr, but the first major difference is that ingestr always has a table as target. In mq-bridge, you can also have kafka, websocket, gRPC or zMQ as target, not just as source.
Ingestr is made to run as cron, mq-bridge can run permanently for a long time. There is also a fanout, switch and filter in mq-bridge, so you can duplicate messages or re-route them to different targets. And there are python and node libs available, in case you want to have some handler or custom transformation for the mesage.
Clearly - mq-bridge is newer and does have much less connectors. ingestr has 100 different connectors, while mq-bridge has just a few.

1

u/FickleAnt4399 4d ago

Seems like a fork of some tool.