MongoDB Aggregate

MongoDB Aggregate

Certified

Run an aggregation pipeline

Executes a MongoDB aggregation pipeline on a collection. Pipeline stages are rendered from Flow variables as BSON documents. Allows disk use by default, caps server execution at 60s and batches 1000 docs. Results are fetched into the outputs by default. Use store: STORE for large results, which writes them to internal storage.

yaml
type: io.kestra.plugin.mongodb.Aggregate

Simple aggregation pipeline to group and sum data

yaml
id: mongodb_aggregate
namespace: company.team

tasks:
  - id: aggregate
    type: io.kestra.plugin.mongodb.Aggregate
    connection:
      uri: "mongodb://root:example@localhost:27017/?authSource=admin"
    database: "my_database"
    collection: "sales"
    pipeline:
      - $match:
          status: "active"
      - $group:
          _id: "$category"
          total:
            $sum: "$amount"
          count:
            $sum: 1
      - $sort:
          total: -1

Complex aggregation with lookup and data transformation

yaml
id: mongodb_complex_aggregate
namespace: company.team

tasks:
  - id: aggregate_with_lookup
    type: io.kestra.plugin.mongodb.Aggregate
    connection:
      uri: "mongodb://root:example@localhost:27017/?authSource=admin"
    database: "my_database"
    collection: "users"
    pipeline:
      - $lookup:
          from: "orders"
          localField: "_id"
          foreignField: "userId"
          as: "userOrders"
      - $addFields:
          totalOrders:
            $size: "$userOrders"
          totalSpent:
            $sum: "$userOrders.amount"
      - $project:
          name: 1
          email: 1
          totalOrders: 1
          totalSpent: 1
      - $match:
          totalOrders:
            $gt: 0
    allowDiskUse: true
    maxTimeMs: 30000
Properties

MongoDB collection

MongoDB connection properties

Definitions
uri*string

Connection string to MongoDB server

URL format like mongodb://mongodb0.example.com: 27017

MongoDB database

SubTypeobject

Aggregation pipeline

List of stages as a BSON string or a map list, rendered once before execution. Values that come from data (inputs, outputs, documents) are kept as literal text and are not evaluated again as expressions. A null inside a map stage is dropped when the flow is parsed. Pass the pipeline as a JSON string when a stage needs a null, for example [{"$match": {"k": null}}].

Defaulttrue

Allow disk use

Enables server-side temp files when a stage exceeds 100 MB; defaults to true.

Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.

Definitions
assetFailureBehaviorstring
Possible Values
IGNOREFAILWARN

Asset failure behavior

Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.

enableAutobooleanstring

Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.

inputsarray

The assets consumed as inputs.

id*string
Min length1
typestring
outputs

The assets produced as outputs.

id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*string
Min length1

Custom asset type

descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
Default1000

Cursor batch size

Documents returned per batch; defaults to 1000.

Default60000

Max execution time (ms)

Server-side limit for the pipeline; defaults to 60000.

DefaultHEX
Possible Values
HEXEXTENDED_JSON

ObjectId format

HEX returns ObjectIds as bare hex strings, as before. EXTENDED_JSON returns them as $oid wrappers, which Load writes back as ObjectIds, so an Aggregate with store: STORE followed by Load keeps the type. Under EXTENDED_JSON, rows[0]._id becomes a map with $oid, for example {"$oid": "60930c39a982931c20ef6cd6"}. Defaults to HEX.

DefaultFETCH
Possible Values
STOREFETCHFETCH_ONENONE

Result handling

FETCH returns all rows in rows, FETCH_ONE a single row in row, NONE only the size, STORE writes an Ion file to internal storage. Defaults to FETCH. FETCH holds every row in memory and saves it in the execution. Use STORE for large results: Kestra rejects task results above kestra.queue.message-protection.limit (1 MiB by default since 2.0), and the task logs a warning when the rows pass 1 MiB.

A single aggregation row

Present when result handling is FETCH_ONE.

Aggregation rows

Present when result handling is FETCH.

Documents returned

Formaturi

Stored result URI

Internal storage URI when result handling is STORE.

Unitcount

Number of documents returned by the aggregation pipeline

More Plugins in this Category

Firecrawl
Firecrawl

Tasks that scrape and crawl websites into clean, LLM-ready content with Firecrawl.

Data

2 tasks

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

8 blueprints