MongoDB

The mongodb reader extracts documents from MongoDB databases using find queries or aggregate pipelines. Document fields are automatically serialized to JSON-compatible data structures, including BSON types like ObjectId, datetime, and Decimal128.

Connection configuration

Authenticate using a connection URI secret or discrete host/port parameters:

source.yaml

spec:
  config:
    mongo_database: { type: string, default: analytics }
  secrets:
    mongo_uri: { type: string }
auth:
  provider: mongodb
  uri: "{{ secrets.mongo_uri }}"
  database: "{{ config.mongo_database }}"
  server_selection_timeout_ms: 5000

Alternatively, use discrete host, port, username, and password fields:

source.yaml

auth:
  provider: mongodb
  host: localhost
  port: 27017
  database: analytics
  username: reader
  password: "{{ secrets.mongo_password }}"

Reading data with find

Run filtered queries against a collection:

streams/events.yaml

requests:
  - name: raw_events
    sdk: mongodb
    method: find
    arguments:
      collection: events
      filter:
        status: completed
      projection:
        _id: 1
        user_id: 1
        amount: 1
        created_at: 1
      sort: [["created_at", 1]]
      limit: 5000
transform:
  mode: page
  steps:
    - name: events
      select: |
        SELECT record->>'_id'                                  AS event_id,
               record->>'user_id'                              AS user_id,
               (record->>'amount')::DOUBLE                     AS amount,
               record->>'created_at'                           AS created_at
        FROM raw_events
export:
  events:
    step: events
    primary_key: [event_id]

Aggregation pipelines with aggregate

Run aggregation pipelines directly on the cluster before extracting results:

streams/user_summaries.yaml

requests:
  - name: summaries
    sdk: mongodb
    method: aggregate
    arguments:
      collection: events
      pipeline:
        - $match: { status: "completed" }
        - $group:
            _id: "$user_id"
            total_spent: { $sum: "$amount" }
            order_count: { $sum: 1 }