Transform Your
MongoDB Business Intelligence

Use Case:

"Real-time analytics pipeline with change streams"

Solution:

                        
// Set up change stream pipeline
const pipeline = [
  {
    $match: {
      'operationType': { $in: ['insert', 'update', 'delete'] },
      'fullDocument.type': 'order'
    }
  },
  {
    $lookup: {
      from: 'products',
      localField: 'fullDocument.productId',
      foreignField: '_id',
      as: 'productDetails'
    }
  },
  {
    $lookup: {
      from: 'customers',
      localField: 'fullDocument.customerId',
      foreignField: '_id',
      as: 'customerDetails'
    }
  },
  {
    $project: {
      operationType: 1,
      orderDetails: '$fullDocument',
      product: { $arrayElemAt: ['$productDetails', 0] },
      customer: { $arrayElemAt: ['$customerDetails', 0] },
      timestamp: '$clusterTime'
    }
  }
];

// Real-time analytics aggregation
const analyticsAggregation = [
  {
    $group: {
      _id: {
        productId: '$orderDetails.productId',
        hour: {
          $hour: '$timestamp'
        }
      },
      totalOrders: { $sum: 1 },
      totalRevenue: {
        $sum: {
          $multiply: ['$orderDetails.quantity', '$product.price']
        }
      },
      uniqueCustomers: {
        $addToSet: '$orderDetails.customerId'
      }
    }
  },
  {
    $project: {
      _id: 0,
      productId: '$_id.productId',
      hour: '$_id.hour',
      totalOrders: 1,
      totalRevenue: 1,
      uniqueCustomers: { $size: '$uniqueCustomers' }
    }
  }
];

// Implement change stream
const changeStream = db.orders.watch(pipeline);
changeStream.on('change', async data => {
  // Process real-time metrics
  const hourlyMetrics = await db.orders
    .aggregate(analyticsAggregation)
    .toArray();

  // Store metrics
  await db.hourlyAnalytics.insertMany(hourlyMetrics);

  // Check for anomalies
  const anomalyDetection = await db.hourlyAnalytics.aggregate([
    {
      $group: {
        _id: '$productId',
        avgOrders: { $avg: '$totalOrders' },
        stdDevOrders: { $stdDevPop: '$totalOrders' }
      }
    },
    {
      $match: {
        $expr: {
          $gt: [
            { $abs: { $subtract: ['$avgOrders', hourlyMetrics.totalOrders] } },
            { $multiply: ['$stdDevOrders', 2] }
          ]
        }
      }
    }
  ]).toArray();

  // Handle anomalies
  if (anomalyDetection.length > 0) {
    await db.anomalyAlerts.insertOne({
      timestamp: new Date(),
      type: 'order_anomaly',
      details: anomalyDetection
    });
  }
});