MongoDB Aggregate
CertifiedRun an aggregation pipeline
MongoDB Aggregate
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.
type: io.kestra.plugin.mongodb.AggregateExamples
Simple aggregation pipeline to group and sum data
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
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
collection *string
MongoDB collection
connection *
MongoDB connection properties
io.kestra.plugin.mongodb.MongoDbConnection
Connection string to MongoDB server
URL format like mongodb://mongodb0.example.com: 27017
database *string
MongoDB database
pipeline *array
Aggregation pipeline
List of stages as BSON string/array or map list rendered before execution.
allowDiskUse booleanstring
trueAllow disk use
Enables server-side temp files when a stage exceeds 100 MB; defaults to true.
assets
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.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset 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.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150batchSize integerstring
1000Cursor batch size
Documents returned per batch; defaults to 1000.
maxTimeMs integerstring
60000Max execution time (ms)
Server-side limit for the pipeline; defaults to 60000.
store string
FETCHSTOREFETCHFETCH_ONENONEResult handling
Fetch returns rows in output; STORE writes an Ion file to internal storage. Defaults to FETCH.
Outputs
rows array
Aggregation rows
Present when result handling is FETCH.
size integer
Documents returned
uri string
uriStored result URI
Internal storage URI when result handling is STORE.
Metrics
records counter
countNumber of documents returned by the aggregation pipeline
