Aggregate

Run aggregation pipeline on a MongoDB collection.

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

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