Cloud
Also in Self-Managed

mongodb

Inserts items into a MongoDB collection.

  • Common

  • Advanced

outputs:
  label: ""
  mongodb:
    url: "" # No default (required)
    database: "" # No default (required)
    username: ""
    password: ""
    collection: "" # No default (required)
    operation: update-one
    write_concern:
      w: majority
      j: false
      w_timeout: ""
    document_map: ""
    filter_map: ""
    hint_map: ""
    upsert: false
    max_in_flight: 64
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)
outputs:
  label: ""
  mongodb:
    url: "" # No default (required)
    database: "" # No default (required)
    username: ""
    password: ""
    app_name: benthos
    aws:
      enabled: false
      region: "" # No default (optional)
      session_duration: 1h
      id: "" # No default (optional)
      secret: "" # No default (optional)
      token: "" # No default (optional)
      role: "" # No default (optional)
      role_external_id: "" # No default (optional)
      roles: [] # No default (optional)
    collection: "" # No default (required)
    operation: update-one
    write_concern:
      w: majority
      j: false
      w_timeout: ""
    document_map: ""
    filter_map: ""
    hint_map: ""
    upsert: false
    max_in_flight: 64
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)

Performance

This output benefits from sending multiple messages in flight, in parallel, for improved performance. You can tune the maximum number of in flight messages (or message batches) using the max_in_flight field.

This output benefits from sending messages as a batch for improved performance. Batches can be formed at both the input and output level. For more information, see Message Batching.

Fields

app_name

The client application name.

Type: string

Default: benthos

aws

AWS IAM authentication using the MONGODB-AWS mechanism, for example against MongoDB Atlas. When enabled, IAM credentials are used instead of a static username and password. Role-derived session credentials are resolved when the component connects and are re-resolved whenever it reconnects. The mongodb processor and cache establish their client once at creation and cannot refresh expiring session credentials, so role, roles and session tokens are rejected for those components; use the ambient credential chain or long-lived access keys with them. For long-running pipelines, prefer the ambient credential chain (leave keys and roles unset), which the driver refreshes automatically.

Type: object

aws.enabled

Enable AWS IAM authentication using the driver-native MONGODB-AWS mechanism. The MongoDB Atlas database user must be created with the AWS IAM authentication type, and connections require TLS. When no static credentials or roles are configured, the ambient AWS credential chain (environment variables, EC2 instance profile, EKS pod role) is used and expiring credentials are refreshed automatically.

Type: bool

Default: false

aws.id

The ID of credentials to use.

Type: string

aws.region

The AWS region used when assuming roles (for STS calls). Only used when role or roles are configured; the ambient and static-key paths ignore it. If no region is specified then the environment default is used.

Type: string

aws.role

Optional AWS IAM role ARN to assume for authentication. Cannot be combined with roles; use the roles array instead when chaining multiple roles.

Type: string

aws.role_external_id

Optional external ID for the role assumption. Only used with the role field, which cannot be combined with roles.

Type: string

aws.roles[]

Optional array of AWS IAM roles to assume for authentication. Roles can be assumed in sequence, enabling chaining for purposes such as cross-account access. Each role can optionally specify an external ID. Cannot be combined with role.

Type: array<object>

aws.roles[].role

AWS IAM role ARN to assume.

Type: string

Default: ""

aws.roles[].role_external_id

Optional external ID for the role assumption.

Type: string

Default: ""

aws.secret

The secret for the credentials being used.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

aws.session_duration

The duration of the STS session requested when assuming roles. AWS requires at least 15 minutes and caps sessions created through role chaining at one hour. Only used when role or roles are configured. For long-running pipelines, prefer the ambient credential chain over a fixed session duration, since the driver refreshes ambient credentials automatically as they near expiry.

Type: string

Default: 1h

aws.token

The token for the credentials being used, required when using short term credentials.

Type: string

batching

Allows you to configure a batching policy.

Type: object

# Examples:
batching:
  byte_size: 5000
  count: 0
  period: 1s

# ---

batching:
  count: 10
  period: 1s

# ---

batching:
  check: this.contains("END BATCH")
  count: 0
  period: 1m

batching.byte_size

The number of bytes at which the batch is flushed. Set to 0 to disable size-based batching.

Type: int

Default: 0

batching.check

A Bloblang query that should return a boolean value indicating whether a message should end a batch.

Type: string

Default: ""

# Examples:
check: this.type == "end_of_transaction"

batching.count

The number of messages after which the batch is flushed. Set to 0 to disable count-based batching.

Type: int

Default: 0

batching.period

The period of time after which an incomplete batch is flushed regardless of its size. This field accepts Go duration format strings such as 100ms, 1s, or 5s.

Type: string

Default: ""

# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms

batching.processors[]

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. All resulting messages are flushed as a single batch, and therefore splitting the batch into smaller batches using these processors is a no-op.

Type: array<processor>

# Examples:
processors:
  - archive:
      format: concatenate


# ---

processors:
  - archive:
      format: lines


# ---

processors:
  - archive:
      format: json_array

collection

The name of the target collection. This field supports interpolation functions.

Type: string

database

The name of the target MongoDB database.

Type: string

document_map

A Bloblang map that represents a document to store in MongoDB, expressed as extended JSON in canonical form. The document_map parameter is required for the following database operations: insert-one, replace-one, and update-one.

Type: string

Default: ""

# Examples:
document_map: |-
  root.a = this.foo
  root.b = this.bar

filter_map

A Bloblang map that represents a filter for a MongoDB command, expressed as extended JSON in canonical form. The filter_map parameter is required for all database operations except insert-one.

This output uses filter_map to find documents for the specified operation. For example, for a delete-one operation, the filter map should include the fields required to locate the document for deletion.

Type: string

Default: ""

# Examples:
filter_map: |-
  root.a = this.foo
  root.b = this.bar

hint_map

A Bloblang map that represents a hint or index for a MongoDB command to use, expressed as extended JSON in canonical form. This map is optional, and is used with all operations except insert-one.

Define a hint_map to improve performance when finding documents in the MongoDB database.

Type: string

Default: ""

# Examples:
hint_map: |-
  root.a = this.foo
  root.b = this.bar

max_in_flight

The maximum number of messages to have in flight at a given time. Increase this number to improve throughput.

Type: int

Default: 64

operation

The MongoDB database operation to perform.

Type: string

Default: update-one

Options: insert-one, delete-one, delete-many, replace-one, update-one

password

The password to use for authentication. Used together with username for basic authentication or with encrypted private keys for secure access.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

upsert

The upsert parameter is optional, and only applies for update-one and replace-one operations. If the filter specified in filter_map matches an existing document, this operation updates or replaces the document, otherwise a new document is created.

Type: bool

Default: false

url

The URL of the target MongoDB server.

Type: string

# Examples:
url: mongodb://localhost:27017

username

The username required to connect to the database.

Type: string

Default: ""

write_concern

The write concern settings for the MongoDB connection.

Type: object

write_concern.j

The j requests acknowledgement from MongoDB, which is created when write operations are written to the journal.

Type: bool

Default: false

write_concern.w

The w requests acknowledgement, which write operations propagate to the specified number of MongoDB instances.

Type: string

Default: majority

write_concern.w_timeout

The write concern timeout.

Type: string

Default: ""