$outと$merge：パイプライン結果の書き込み

Writing Pipeline Results to Collections

The $out Stage

// Materialise monthly sales summary into its own collection
db.orders.aggregate([
  { $match: { status: 'completed' } },
  { $group: {
    _id: {
      year: { $year: '$createdAt' },
      month: { $month: '$createdAt' }
    },
    totalRevenue: { $sum: '$amount' },
    orderCount: { $sum: 1 }
  }},
  { $sort: { '_id.year': 1, '_id.month': 1 } },
  { $out: 'monthly_sales_summary' }  // last stage, writes to collection
]);

$out Atomicity Guarantee

// Readers of 'monthly_sales_summary' always see complete data
// Even while $out is running, they see the previous full snapshot
// Only after $out completes does the new snapshot become visible

// Scheduled nightly refresh pattern:
// 1. Run at midnight: aggregate([...stages..., { $out: 'sales_report' }])
// 2. During the day: app reads from 'sales_report' (fast, pre-computed)
// 3. Next midnight: repeat

db.orders.aggregate([
  ...stages,
  { $out: 'sales_report' }  // atomic swap
]);

$out to a Different Database

// Write to a collection in a different database
db.orders.aggregate([
  { $match: { status: 'completed' } },
  { $group: { _id: '$region', revenue: { $sum: '$amount' } } },
  {
    $out: {
      db: 'reporting',          // target database
      coll: 'regional_revenue'  // target collection
    }
  }
]);
// Result is in reporting.regional_revenue

The $merge Stage

// Upsert daily stats into a persistent stats collection
db.events.aggregate([
  { $group: {
    _id: {
      date: { $dateToString: { format: '%Y-%m-%d', date: '$timestamp' } },
      eventType: '$type'
    },
    count: { $sum: 1 }
  }},
  {
    $merge: {
      into: 'daily_event_stats',
      on: ['_id'],              // match key
      whenMatched: 'replace',   // update matching docs
      whenNotMatched: 'insert'  // insert new docs
    }
  }
]);

$merge whenMatched Options

// whenMatched: 'merge' - only update changed fields, keep others
db.orders.aggregate([
  { $project: { userId: 1, orderCount: { $literal: 1 } } },
  { $merge: {
    into: 'user_order_counts',
    on: 'userId',
    whenMatched: [{ $set: { orderCount: { $add: ['$orderCount', '$$new.orderCount'] } } }],
    whenNotMatched: 'insert'
  }}
]);
// Custom pipeline in whenMatched adds to existing count instead of replacing

$merge whenNotMatched Options

// Full upsert: replace existing, insert new
{ $merge: {
  into: 'product_stats',
  on: '_id',
  whenMatched: 'replace',
  whenNotMatched: 'insert'
}}

// Update only existing, silently skip new
{ $merge: {
  into: 'product_stats',
  on: '_id',
  whenMatched: 'replace',
  whenNotMatched: 'discard'  // only update existing products
}}

Incremental Materialised Views With $merge

// Incremental update: only process last hour of orders
const oneHourAgo = new Date(Date.now() - 3600000);

db.orders.aggregate([
  { $match: { createdAt: { $gte: oneHourAgo } } },  // only NEW data
  { $group: {
    _id: '$productId',
    recentRevenue: { $sum: '$amount' },
    recentOrders: { $sum: 1 }
  }},
  { $merge: {
    into: 'product_revenue',
    on: '_id',
    whenMatched: [
      { $set: {
        totalRevenue: { $add: ['$totalRevenue', '$$new.recentRevenue'] },
        totalOrders: { $add: ['$totalOrders', '$$new.recentOrders'] }
      }}
    ],
    whenNotMatched: 'insert'
  }}
]);

$out vs $merge: When to Use Each

// $out: nightly full replace
// Run at midnight: compute full summary, atomically replace target
{ $out: 'monthly_report' }

// $merge: hourly incremental update
// Run every hour: compute last hour's delta, merge into running total
{ $merge: {
  into: 'running_totals',
  on: '_id',
  whenMatched: 'merge',
  whenNotMatched: 'insert'
}}

Permissions and Indexes on $out/$merge Targets

// After $out, recreate indexes on the refreshed collection
db.orders.aggregate([...stages, { $out: 'order_summary' }]);
// Now recreate needed indexes:
db.order_summary.createIndex({ userId: 1 });
db.order_summary.createIndex({ createdAt: -1 });

// $merge preserves existing indexes automatically
// No index recreation needed after $merge

ETL Pipelines With $merge

// ETL: clean and transform raw events into a processed_events collection
db.raw_events.aggregate([
  // Extract: filter valid events
  { $match: { eventType: { $in: ['click', 'view', 'purchase'] }, userId: { $exists: true } } },
  // Transform: reshape and enrich
  { $addFields: {
    processedAt: '$$NOW',
    eventDate: { $dateToString: { format: '%Y-%m-%d', date: '$timestamp' } }
  }},
  { $project: { _id: 0, eventType: 1, userId: 1, eventDate: 1, processedAt: 1 } },
  // Load: upsert into destination
  { $merge: {
    into: 'processed_events',
    on: ['userId', 'eventDate', 'eventType'],
    whenMatched: 'keepExisting',  // don't reprocess
    whenNotMatched: 'insert'
  }}
]);

Quick Check

Lesson Recap