Change Stream으로 데이터 모니터링

개요

이 가이드에서는 변경 스트림 을 사용하여 데이터베이스의 실시간 변경 사항을 모니터링하는 방법을 배울 수 있습니다. 변경 스트림은 애플리케이션이 컬렉션, 데이터베이스 또는 배포의 데이터 변경 사항을 구독할 수 있도록 하는 MongoDB Server 기능입니다.

변경 스트림 열기

변경 스트림을 열려면 watch() 메서드를 호출합니다. watch() 메서드를 호출하는 인스턴스에 따라 변경 스트림이 수신 대기하는 이벤트 범위가 결정됩니다. 다음 클래스에서 watch() 메서드를 호출할 수 있습니다.

• MongoClient: MongoDB deployment의 모든 변경 사항을 모니터링합니다.

• Database: 데이터베이스에 있는 모든 컬렉션의 변경 사항을 모니터링합니다.

• Collection: 컬렉션의 변경 사항을 모니터링합니다.

다음 예시 에서는 restaurants 컬렉션 에서 변경 스트림 열고 변경 사항이 발생할 때 출력합니다. Synchronous 또는 Asynchronous 탭 선택하여 해당 코드를 확인합니다.

database = client["sample_restaurants"]
collection = database["restaurants"]
with collection.watch() as stream:
    for change in stream:
        print(change)

database = client["sample_restaurants"]
collection = database["restaurants"]
async with await collection.watch() as stream:
    async for change in stream:
        print(change)

변경 사항을 확인하려면 애플리케이션 실행 . 그런 다음 별도의 애플리케이션 또는 셸 에서 restaurants 컬렉션 수정합니다. 다음 예시 name 필드 값이 Blarney Castle인 문서 업데이트합니다. Synchronous 또는 Asynchronous 탭 선택하여 해당 코드를 확인합니다.

database = client["sample_restaurants"]
collection = database["restaurants"]
query_filter = { "name": "Blarney Castle" }
update_operation = { '$set' : 
    { "cuisine": "Irish" }
}
result = collection.update_one(query_filter, update_operation)

database = client["sample_restaurants"]
collection = database["restaurants"]
query_filter = { "name": "Blarney Castle" }
update_operation = { '$set' : 
    { "cuisine": "Irish" }
}
result = await collection.update_one(query_filter, update_operation)

컬렉션을 업데이트하면 변경 스트림 애플리케이션은 변경 사항이 발생하는 즉시 출력합니다. 인쇄된 변경 이벤트는 다음과 유사합니다.

{'_id': {'_data': '...'}, 'operationType': 'update', 'clusterTime': Timestamp(...), 'wallTime': datetime.datetime(...),
'ns': {'db': 'sample_restaurants', 'coll': 'restaurants'}, 'documentKey': {'_id': ObjectId('...')},
'updateDescription': {'updatedFields': {'cuisine': 'Irish'}, 'removedFields': [], 'truncatedArrays': []}}

변경 스트림 출력 수정

pipeline 매개변수를 watch() 메서드에 전달하여 변경 스트림 출력을 수정할 수 있습니다. 이 매개변수를 사용하면 지정된 변경 이벤트만 감시할 수 있습니다. 매개변수의 형식을 각각 애그리게이션 단계를 나타내는 객체 목록으로 지정합니다.

pipeline 매개변수에 다음 단계를 지정할 수 있습니다.

• $addFields

• $match

• $project

• $replaceRoot

• $replaceWith

• $redact

• $set

• $unset

다음 예시 pipeline 매개 변수를 사용하여 업데이트 작업만 기록하는 변경 스트림 엽니다. Synchronous 또는 Asynchronous 탭 선택하여 해당 코드를 확인합니다.

change_pipeline = { "$match": { "operationType": "update" }},
with collection.watch(pipeline=change_pipeline) as stream:
    for change in stream:
        print(change)

change_pipeline = { "$match": { "operationType": "update" }},
async with await collection.watch(pipeline=change_pipeline) as stream:
    async for change in stream:
        print(change)

변경 스트림 출력 수정에 대해 자세히 알아보려면 MongoDB Server 매뉴얼의 변경 스트림 출력 수정 섹션을 참조하세요.

watch() 동작 수정

watch() 메서드는 작업을 구성하는 데 사용할 수 있는 옵션을 나타내는 선택적 매개변수를 허용합니다. 옵션을 지정하지 않으면 드라이버는 작업을 사용자 지정하지 않습니다.

다음 표에서는 watch() 의 동작을 사용자 지정하기 위해 설정할 수 있는 옵션에 대해 설명합니다.

database = client["sample_restaurants"]
collection = database["restaurants"]
with collection.watch(full_document='updateLookup') as stream:
    for change in stream:
        print(change)

database = client["sample_restaurants"]
collection = database["restaurants"]
async with await collection.watch(full_document='updateLookup') as stream:
    async for change in stream:
        print(change)

{'_id': {'_data': '...'}, 'operationType': 'update', 'clusterTime': Timestamp(...), 'wallTime': datetime.datetime(...),
 'fullDocument': {'_id': ObjectId('...'), 'address': {...}, 'borough': 'Queens',
 'cuisine': 'Irish', 'grades': [...], 'name': 'Blarney Castle', 'restaurant_id': '40366356'},
 'ns': {'db': 'sample_restaurants', 'coll': 'restaurants'}, 'documentKey': {'_id': ObjectId('...')},
 'updateDescription': {'updatedFields': {'cuisine': 'Irish'}, 'removedFields': [], 'truncatedArrays': []}}

추가 정보

변경 스트림에 대해 자세히 알아보려면 Change Streams 매뉴얼의 MongoDB Server 을 참조하세요.