MongoDB

Reads from or writes to a MongoDB collection. As a source it can copy a collection, follow its changes (change data capture), or work a collection as a job queue. database and collection are parameters of their own; the database is not read from the URL path.

URL format

mongodb://[user:password@]host[:port]/?database=<db>&collection=<name>

Rules that the table follows:

• database is required and is always a parameter. mongodb://localhost/shop?collection=orders does not start.
• Driver options stay on the connection string. A parameter that is not a field of this connector, such as replicaSet, authSource or tls=true, is passed to the MongoDB driver.
• A host list goes in url=, with _ as a placeholder host, because it is not a URL the command line can parse. In a config file the connection string is used as it is.

Config (YAML / library)

Every URL parameter is a field of the same name under mongodb:. url takes any MongoDB connection string.

output:
  mongodb:
    url: "mongodb+srv://cluster0.ab12c.mongodb.net"
    username: "app"
    password: "${MONGO_PASSWORD}"
    database: "shop"
    collection: "orders"
    format: raw

username and password take precedence over credentials in the url. For TLS with a private CA or a client certificate, set tls:

output:
  mongodb:
    url: "mongodb://db.example.com:27017"
    database: "shop"
    collection: "orders"
    tls: { required: true, ca_file: "/etc/ssl/mongo-ca.pem" }

Writing

format decides what a stored document looks like. For a collection that other programs query, use format: raw. The default keeps the message intact for another mq-bridge route to read, and stores the payload as binary.

For the message {"order_id": "A-1", "total": 20}:

With raw, a payload that is not a JSON object is stored like normal.

Load a JSONL file into a collection, with the order number as _id:

load_orders:
  input:
    file: { path: "/data/orders.jsonl", format: raw }
  output:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "shop"
      collection: "orders"
      format: raw
      id_field: "order_id"

• id_field makes the write repeatable. Its value becomes _id, with its JSON type kept. A document whose _id exists is left as it is and counts as written, so running the load twice does not duplicate anything. It does not update the stored document. A message without the field fails. More in Upserts & insert-if-absent.
• Without id_field the _id is the payload’s own _id (raw), or one that is generated.
• ttl_seconds creates a TTL index, so documents are removed that long after they were written. capped_size_bytes creates the collection as a capped collection.

Reading

consume selects one of four ways to read:

A single-node replica set is enough for the two capture modes; without one they refuse to start. The comparison in detail is under MongoDB consume modes.

Follow a collection and publish every change to Kafka:

orders_cdc:
  input:
    mongodb:
      url: "mongodb://localhost:27017/?replicaSet=rs0"
      database: "shop"
      collection: "orders"
      consume: capture_all
      cursor_id: "orders-to-kafka"
  output:
    kafka: { url: "localhost:9092", topic: "shop.orders" }

• The payload is the document as JSON. Types that JSON does not have are written the MongoDB way: an ObjectId as {"$oid": "…"}, a date as {"$date": …}.
• mongodb.operation says what happened: insert, update, replace or delete. An update carries the whole document as it is after the change. A delete carries only the key, {"_id": …}. mongodb.document_id has the _id in every case.
• cursor_id names the saved position, so a restart continues where it stopped. By default the position is kept in the source database, in the collection mqb_cursors_<collection>; checkpoint_store puts it elsewhere, for a source you may not write to. See Checkpoints.

Copy a collection of a standalone mongod into a file, once:

mqb copy --drain \
  --from 'mongodb://localhost/?database=shop&collection=customers&consume=snapshot' \
  --to 'file:///data/customers.jsonl?format=raw'

snapshot delivers what exists when it runs. It cannot resume and refuses a cursor_id.

A collection as a job queue

With consume: delete several instances share the work: each document goes to one of them and is deleted when it was processed. The documents must have been written by an mq-bridge output with the format normal, json or text, which is the shape the queue expects; do not point this mode at a collection of your own data.

enqueue_jobs:
  input:
    http: { url: "0.0.0.0:8080", path: "/jobs" }
  output:
    mongodb: { url: "mongodb://localhost:27017", database: "work", collection: "jobs", format: json }

work_jobs:
  input:
    mongodb: { url: "mongodb://localhost:27017", database: "work", collection: "jobs", consume: delete }
  output:
    http: { url: "http://worker.internal:9000/run" }

The queue is read in the order the documents were written. The publisher stores that position in the field insert_order, apart from the _id, so a message_id from another system does not change the order. It is a UUIDv7 that always rises within one process; two writers are ordered by their clocks. A reader that needs an exact order uses capture_all or capture_new.

Documents without insert_order are read first, in _id order: those written with format: raw, by another program, or by an earlier version.

Examples

Load a CSV file into a collection, one-shot:

mqb copy --drain \
  --from 'file:///data/customers.csv?format=csv' \
  --to 'mongodb://localhost/?database=app&collection=customers&format=raw'

Copy a collection into PostgreSQL (replica set required):

mqb copy --drain \
  --from 'mongodb://localhost/?database=app&collection=customers' \
  --to 'postgres://user:pass@localhost/app?table=customers&auto_create_table=true'

Watch for changes only, continuous:

mqb copy \
  --from 'mongodb://localhost/?database=app&collection=orders&consume=capture_new' \
  --to 'kafka://kafka.local:9092?topic=orders'

Key options

Full field list: reference/mongodb.md.