집계 파이프라인으로 이벤트 필터링하기
학습자는 watch()에 파이프라인을 전달해 애플리케이션에 필요한 이벤트만 수신합니다.
집계 파이프라인으로 이벤트 필터링하기은(는) CoddyKit의 무료 MongoDB Academy 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 MongoDB Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. MongoDB Academy 강의에는 총 4개의 강의가 포함되어 있습니다.
이 강의의 일부는 아직 번역되지 않았으며 영어로 표시됩니다.
Why Filter Change Stream Events?
Without filtering, a change stream delivers every change event on a collection. In a busy production collection, this can mean thousands of events per second, most of which your application does not care about. Filtering at the server using an aggregation pipeline reduces network traffic, lowers CPU usage in your application, and ensures your event handler only processes relevant events. Filtering happens before events leave MongoDB—only matching events are transmitted to your client.
Passing a Pipeline to watch()
The first argument to watch() is an aggregation pipeline array. MongoDB applies this pipeline to each change event document before deciding whether to deliver it to your application. Not all aggregation stages are permitted in change stream pipelines—only a specific subset is allowed, primarily $match, $project, $addFields, $replaceRoot, and $redact. The $group and $lookup stages are not allowed.
// Only receive insert events — filter everything else
const changeStream = db.collection('orders').watch([
{
$match: {
operationType: 'insert'
}
}
]);
for await (const change of changeStream) {
// Only insert events arrive here
console.log('New order:', change.fullDocument._id);
}Filtering by Operation Type
Filtering by operationType is the most common pipeline filter. You can use $match with a single operation type string or with $in to match multiple types. This is useful when your application cares about inserts and updates but not deletes, or when different microservices subscribe to different operation types on the same collection.
// React only to new orders and status updates
const stream = db.collection('orders').watch([
{
$match: {
operationType: { $in: ['insert', 'update'] }
}
}
]);
// Or match a single type:
const deletedStream = db.collection('orders').watch([
{ $match: { operationType: 'delete' } }
]);Filtering Update Events by Changed Fields
You can filter update events based on which fields were modified by querying the updateDescription.updatedFields object in the $match stage. This lets you subscribe only to specific field changes—for example, only when a document's status field transitions to a particular value. This is more efficient than receiving all updates and filtering in application code.
// Only receive updates where status changed to 'shipped'
const shippedStream = db.collection('orders').watch([
{
$match: {
operationType: 'update',
'updateDescription.updatedFields.status': 'shipped'
}
}
], { fullDocument: 'updateLookup' });
for await (const change of shippedStream) {
const order = change.fullDocument;
await sendShippingEmail(order.customerId, order.trackingNumber);
}Filtering by Document Field Values
For insert events, you can filter based on fields in the fullDocument sub-document. For example, receive only inserts where fullDocument.priority is 'high' or fullDocument.region equals 'US-WEST'. This server-side filtering is especially powerful in multi-tenant architectures where different application instances need events for different subsets of data.
// Only receive inserts for high-priority orders in the US-WEST region
const priorityStream = db.collection('orders').watch([
{
$match: {
operationType: 'insert',
'fullDocument.priority': 'high',
'fullDocument.region': 'US-WEST'
}
}
]);
for await (const change of priorityStream) {
await escalateOrder(change.fullDocument);
}Using $project to Reshape Events
The $project stage in a change stream pipeline reshapes the event document before it is delivered to your application. You can include only the fields your handler needs, rename fields, or compute derived fields. This reduces the payload size transmitted over the network and simplifies your event handler code by presenting only the data it needs.
// Project only the fields the handler needs
const stream = db.collection('users').watch([
{ $match: { operationType: { $in: ['insert', 'update'] } } },
{
$project: {
operationType: 1,
'documentKey._id': 1,
'updateDescription.updatedFields.email': 1,
'fullDocument.email': 1,
'fullDocument.name': 1
}
}
]);
// Handler receives trimmed events with only email and nameUsing $addFields to Enrich Events
The $addFields stage lets you add computed fields to the change event document. You can add a timestamp when the event was processed, derive a category from the operation type, or compute a routing key. These enriched fields are included in the event your application receives, allowing downstream code to use pre-computed values without recalculating them.
const stream = db.collection('payments').watch([
{
$addFields: {
processedAt: '$$NOW', // current timestamp as event enrichment
eventCategory: {
$switch: {
branches: [
{ case: { $eq: ['$operationType', 'insert'] }, then: 'NEW_PAYMENT' },
{ case: { $eq: ['$operationType', 'update'] }, then: 'PAYMENT_UPDATE' }
],
default: 'OTHER'
}
}
}
}
]);Chaining Multiple Stages
You can chain multiple pipeline stages in a change stream pipeline for powerful composition. A common pattern is: $match to filter events → $addFields to enrich → $project to trim. Each stage processes the output of the previous one. Remember that the order of stages matters—apply the most selective $match first to minimize the documents processed by later stages.
const stream = db.collection('inventory').watch([
// Stage 1: filter to updates only
{ $match: { operationType: 'update' } },
// Stage 2: add computed field
{
$addFields: {
isLowStock: {
$lt: ['$updateDescription.updatedFields.quantity', 10]
}
}
},
// Stage 3: only pass through low-stock events
{ $match: { isLowStock: true } },
// Stage 4: trim to essential fields
{ $project: { 'documentKey._id': 1, operationType: 1 } }
]);Performance Impact of Server-Side Filtering
Server-side pipeline filtering in change streams is significantly more efficient than receiving all events and filtering in application code. Without server-side filtering, every event must be serialized and transmitted over the network. With a $match stage, MongoDB evaluates the filter internally and transmits only matching events. For high-traffic collections, this can reduce network usage and application CPU by orders of magnitude.
// Inefficient: receive all events, filter in JS
for await (const change of db.collection('orders').watch()) {
if (change.operationType === 'insert' && change.fullDocument.total > 1000) {
// Most events are discarded here — wasted network I/O
}
}
// Efficient: filter server-side
for await (const change of db.collection('orders').watch([
{ $match: { operationType: 'insert', 'fullDocument.total': { $gt: 1000 } } }
])) {
// Only matching events arrive here
}Permitted vs Forbidden Stages
MongoDB restricts which aggregation stages can be used in change stream pipelines. Permitted: $match, $project, $addFields, $replaceRoot, $replaceWith, $redact. Forbidden: $group, $lookup, $unwind, $geoNear, $out, $merge, and several others. Attempting to use a forbidden stage causes an error when opening the stream. If you need complex transformations, do them in application code after receiving the (pre-filtered) events.
// WRONG — $group is not allowed in change stream pipelines
db.collection('orders').watch([
{ $group: { _id: '$fullDocument.region', count: { $sum: 1 } } } // Error!
]);
// RIGHT — use only permitted stages in the pipeline
db.collection('orders').watch([
{ $match: { operationType: 'insert' } },
{ $project: { 'fullDocument.region': 1, 'fullDocument.total': 1 } }
]);Multi-Tenant Filtering Pattern
In multi-tenant applications, multiple tenants share one collection with a tenantId field. Rather than running one change stream per tenant (expensive), run one stream per service instance with a $match filter on fullDocument.tenantId scoped to the tenants that instance serves. This scales to hundreds of tenants with far fewer open cursors on the MongoDB server.
// Service instance handles tenants T1 and T2 only
const myTenants = ['T1', 'T2'];
const stream = db.collection('events').watch([
{
$match: {
$or: [
{ 'fullDocument.tenantId': { $in: myTenants } }, // for inserts
{ 'updateDescription.updatedFields.tenantId': { $in: myTenants } } // for updates
]
}
}
], { fullDocument: 'updateLookup' });Quick Check
Test your understanding of MongoDB & NoSQL Databases concepts from this lesson.
Lesson Recap
In this lesson you learned: pass an aggregation pipeline as the first argument to watch() to filter events server-side, permitted stages include $match, $project, $addFields, $replaceRoot, and $redact — but not $group or $lookup, and server-side filtering dramatically reduces network traffic and application CPU compared to application-side filtering. Next up we explore resuming change streams after an interruption using resume tokens.
자주 묻는 질문
“집계 파이프라인으로 이벤트 필터링하기” 강의는 무료인가요?
네 — “집계 파이프라인으로 이벤트 필터링하기” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 MongoDB Academy 강의 전체를 잠금 해제할 수 있습니다. MongoDB Academy 강의에는 총 4개의 강의가 포함되어 있습니다.
“집계 파이프라인으로 이벤트 필터링하기”에서 뭘 배우나요?
학습자는 watch()에 파이프라인을 전달해 애플리케이션에 필요한 이벤트만 수신합니다. 브라우저에서 직접 실행하는 실습 코드로 MongoDB Academy을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
MongoDB Academy을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 MongoDB Academy은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.
“집계 파이프라인으로 이벤트 필터링하기” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 MongoDB Academy 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 MongoDB Academy 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 컬렉션에서 변경 스트림 열기
- 변경 이벤트 문서 구조
- 집계 파이프라인으로 이벤트 필터링하기
- 중단 후 변경 스트림 재개하기