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, batches 1000 docs, and can fetch results or store 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*Requiredstring

Connection string to MongoDB server

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

MongoDB database

SubTypeobject

Aggregation pipeline

List of stages as BSON string/array or map list rendered before execution.

Defaulttrue

Allow disk use

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

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.

Reference (ref) of the pluginDefaults to apply to this task.

DefaultFETCH
Possible Values
STOREFETCHFETCH_ONENONE

Result handling

Fetch returns rows in output; STORE writes an Ion file to internal storage. Defaults to FETCH.

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

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

PostgreSQL
PostgreSQL

Tasks that connect to PostgreSQL via JDBC to run queries and loads.

Data

6 tasks

25 blueprints