MongoDB Aggregation 파이프라인 최적화

Introduction

MongoDB의 Aggregation Framework는 복잡한 데이터 분석과 변환을 수행할 수 있는 강력한 도구입니다. 하지만 파이프라인이 복잡해지고 데이터량이 증가하면 성능 문제가 발생할 수 있습니다. 이 글에서는 aggregation 파이프라인 최적화 기법과 실제 프로덕션 환경에서의 활용 방법을 다루겠습니다.

Environment

// MongoDB 연결 및 테스트 데이터 생성
const { MongoClient } = require('mongodb');

const uri = 'mongodb://localhost:27017';
const client = new MongoClient(uri);

async function createTestData() {
  const db = client.db('analytics');
  const collection = db.collection('events');
  
  // 대용량 테스트 데이터 생성
  const bulkOps = [];
  for (let i = 0; i < 1000000; i++) {
    bulkOps.push({
      insertOne: {
        document: {
          eventType: ['click', 'view', 'purchase'][Math.floor(Math.random() * 3)],
          userId: Math.floor(Math.random() * 10000),
          amount: Math.random() * 1000,
          timestamp: new Date(Date.now() - Math.random() * 30 * 24 * 60 * 60 * 1000),
          metadata: {
            platform: ['web', 'mobile', 'tablet'][Math.floor(Math.random() * 3)],
            country: ['US', 'KR', 'JP', 'EU'][Math.floor(Math.random() * 4)]
          }
        }
      }
    });
  }
  
  await collection.bulkWrite(bulkOps);
}

Problem

비최적화된 aggregation 파이프라인의 성능 문제:

// 느린 aggregation 쿼리
const slowPipeline = [
  {
    $match: {
      timestamp: {
        $gte: new Date('2025-01-01'),
        $lt: new Date('2025-02-01')
      }
    }
  },
  {
    $group: {
      _id: {
        userId: '$userId',
        eventType: '$eventType'
      },
      totalAmount: { $sum: '$amount' },
      count: { $sum: 1 }
    }
  },
  {
    $group: {
      _id: '$_id.userId',
      eventStats: {
        $push: {
          eventType: '$_id.eventType',
          totalAmount: '$totalAmount',
          count: '$count'
        }
      }
    }
  },
  {
    $sort: { '_id': 1 }
  }
];

// 실행 시간 측정
console.time('slow-aggregation');
const result = await collection.aggregate(slowPipeline).toArray();
console.timeEnd('slow-aggregation');
// slow-aggregation: 45234ms

Analysis

성능 문제의 원인을 분석했습니다:

// 실행 계획 확인
const explainResult = await collection.aggregate(slowPipeline, {
  explain: true
});

console.log(JSON.stringify(explainResult, null, 2));

// 주요 문제점:
// 1. $match 단계에서 인덱스 미사용 (COLLSCAN)
// 2. $group 단계에서 대량의 중간 결과 집합 생성
// 3. 메모리 제한 초과 (기본 100MB)

// 인덱스 상태 확인
const indexes = await collection.indexes();
console.log('현재 인덱스:', indexes);

// 기존 인덱스:
// { _id: 1 }  // 기본 인덱스만 존재

// 쿼리 패턴 분석
const cursor = collection.aggregate([
  { $match: { timestamp: { $gte: new Date('2025-01-01') } } },
  { $group: { _id: '$userId', count: { $sum: 1 } } }
]);

Solution

1단계: 인덱스 최적화

// 필요한 인덱스 생성
await collection.createIndex(
  { timestamp: -1, userId: 1, eventType: 1 },
  { name: 'idx_timestamp_user_event' }
);

// 인덱스 힌트 사용
const optimizedPipeline = [
  {
    $match: {
      timestamp: {
        $gte: new Date('2025-01-01'),
        $lt: new Date('2025-02-01')
      }
    }
  },
  {
    $match: {  // 인덱스 활용을 위한 추가 필터링
      userId: { $exists: true }
    }
  },
  {
    $group: {
      _id: {
        userId: '$userId',
        eventType: '$eventType'
      },
      totalAmount: { $sum: '$amount' },
      count: { $sum: 1 }
    }
  }
];

console.time('optimized-aggregation');
const optimizedResult = await collection.aggregate(optimizedPipeline, {
  hint: 'idx_timestamp_user_event'
}).toArray();
console.timeEnd('optimized-aggregation');
// optimized-aggregation: 2345ms (20배 향상)

2단계: 파이프라인 단계 최적화

// $project로 조기 필드 축소
const optimizedPipelineV2 = [
  {
    $match: {
      timestamp: {
        $gte: new Date('2025-01-01'),
        $lt: new Date('2025-02-01')
      },
      eventType: { $in: ['click', 'view', 'purchase'] }
    }
  },
  {
    $project: {
      userId: 1,
      eventType: 1,
      amount: 1
      // 불필요한 필드 제외
    }
  },
  {
    $group: {
      _id: {
        userId: '$userId',
        eventType: '$eventType'
      },
      totalAmount: { $sum: '$amount' },
      count: { $sum: 1 }
    }
  },
  {
    $setWindowFields: {
      partitionBy: '$_id.userId',
      sortBy: { totalAmount: -1 },
      output: {
        rank: {
          $rank: {}
        }
      }
    }
  },
  {
    $match: { rank: { $lte: 3 } }  // 상위 3개만 유지
  }
];

3단계: allowDiskUse 및 메모리 최적화

// 대용량 데이터 처리를 위한 allowDiskUse
const largeDataPipeline = [
  {
    $match: {
      timestamp: {
        $gte: new Date('2025-01-01'),
        $lt: new Date('2025-06-01')
      }
    }
  },
  {
    $group: {
      _id: '$userId',
      totalAmount: { $sum: '$amount' },
      avgAmount: { $avg: '$amount' },
      eventCount: { $sum: 1 },
      firstEvent: { $min: '$timestamp' },
      lastEvent: { $max: '$timestamp' }
    }
  },
  {
    $addFields: {
      activityDays: {
        $divide: [
          { $subtract: ['$lastEvent', '$firstEvent'] },
          86400000
        ]
      }
    }
  },
  {
    $sort: { totalAmount: -1 }
  },
  {
    $limit: 100
  }
];

// allowDiskUse 활성화
const result = await collection.aggregate(largeDataPipeline, {
  allowDiskUse: true,
  maxTimeMS: 60000
}).toArray();

Lessons Learned

인덱스 설계: $match 단계에 사용되는 필드에 반드시 인덱스를 생성해야 합니다
조기 필터링: 가능한 한 빠른 단계에서 데이터를 필터링하여 처리량을 줄여야 합니다
$project 활용: 불필요한 필드를 조기에 제외하면 I/O와 메모리 사용량이 감소합니다
allowDiskUse: 대용량 데이터 처리 시 필수적이며, 메모리 제한을 초과하는 임시 데이터를 디스크에 쓸 수 있게 합니다
배치 처리: 매우 큰 결과 집합은 배치로 나누어 처리하면 메모리 효율이 향상됩니다

이 블로그는 외부 스폰서십, 제휴 마케팅 또는 광고 수익을 받지 않습니다.

MongoDB Aggregation 파이프라인 최적화: 대용량 데이터 처리