Change Stream으로 데이터 모니터링

샘플 데이터

이 가이드의 예시에서는 Atlas 샘플 데이터 세트의 sample_restaurants 데이터베이스에 있는 restaurants 컬렉션을 사용합니다. 무료 MongoDB 배포서버를 생성하고 샘플 데이터 세트를 로드하는 방법을 알아보려면 MongoDB 시작하기 가이드를 참조하세요.

다음 코틀린 (Kotlin) 데이터 클래스는 이 컬렉션 의 문서를 모델링합니다.

data class Restaurant(
    val name: String,
    val cuisine: String,
)

변경 스트림 예시 열기

다음 예시 에서는 restaurants 컬렉션 에서 변경 스트림 을 열고 변경 사항이 발생할 때 출력합니다.

collection.watch().forEach { change ->
    println(change)
}

변경 사항을 확인하려면 애플리케이션 을 실행 합니다. 그런 다음 별도의 애플리케이션 또는 shell 에서 restaurants 컬렉션 에 대한 쓰기 (write) 작업을 수행합니다. 다음 예시 에서는 name 값이 "Blarney Castle" 인 문서 를 업데이트합니다.

val filter = Filters.eq(Restaurant::name.name, "Blarney Castle")
val update = Updates.set(Restaurant::cuisine.name, "Irish")
val result = collection.updateOne(filter, update)

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

{
  "_id": { ... },
  "operationType": "update",
  "clusterTime": { ... },
  "ns": {
    "db": "sample_restaurants",
    "coll": "restaurants"
  },
  "updateDescription": {
    "updatedFields": {
      "cuisine": "Irish"
    },
    "removedFields": [],
    "truncatedArrays": []
  }
  ...
}

특정 이벤트 일치 예시

다음 예시 에서는 $match 가 포함된 pipeline 매개 변수를 사용하여 업데이트 작업만 기록하는 변경 스트림 을 엽니다.

val pipeline = listOf(
    Aggregates.match(Filters.eq("operationType", "update"))
)
collection.watch(pipeline).forEach { change ->
    println(change)
}

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

사전 이미지 및 사후 이미지 포함하기

기본값 으로 컬렉션 에서 작업을 수행할 때 해당 변경 이벤트 에는 해당 작업에 의해 수정된 필드의 델타만 포함됩니다. 변경 전후의 전체 문서 를 보려면 fullDocumentBeforeChange() 또는 fullDocument() 메서드를 watch() 메서드에 연결합니다.

사전 이미지 는 변경 전의 문서 전체 버전입니다. 변경 스트림 이벤트 에 사전 이미지를 포함하려면 다음 옵션 중 하나를 fullDocumentBeforeChange() 메서드에 전달합니다.

• FullDocumentBeforeChange.WHEN_AVAILABLE: 변경 이벤트에는 사전 이미지를 사용할 수 있는 경우에만 변경 이벤트에 대해 수정된 문서의 사전 이미지가 포함됩니다.

• FullDocumentBeforeChange.REQUIRED: 변경 이벤트에는 변경 이벤트에 대한 수정된 문서의 사전 이미지가 포함됩니다. 사전 이미지를 사용할 수 없는 경우 드라이버에서 오류가 발생합니다.

사후 이미지 는 변경 후 문서 의 전체 버전입니다. 변경 스트림 이벤트 에 사후 이미지를 포함하려면 다음 옵션 중 하나를 fullDocument() 메서드에 전달합니다.

• FullDocument.UPDATE_LOOKUP: 변경 이벤트에는 변경 후 일정 시간 이후의 변경된 문서 전체의 복사본이 포함됩니다.

• FullDocument.WHEN_AVAILABLE: 변경 이벤트에는 사후 이미지를 사용할 수 있는 경우에만 변경 이벤트에 대해 수정된 문서의 사후 이미지가 포함됩니다.

• FullDocument.REQUIRED: 변경 이벤트에는 변경 이벤트에 대한 수정된 문서의 사후 이미지가 포함됩니다. 사후 이미지를 사용할 수 없는 경우 드라이버에서 오류가 발생합니다.

다음 예시 에서는 컬렉션 에서 watch() 메서드를 호출하고 fullDocument 매개변수를 지정하여 업데이트된 문서의 사후 이미지를 결과에 포함합니다.

collection.watch().fullDocument(FullDocument.UPDATE_LOOKUP).forEach { change ->
    println("Received a change: $change")
}

변경 스트림 애플리케이션이 실행 중인 상태에서 앞의 업데이트 예시 를 사용하여 restaurants 컬렉션의 문서를 업데이트하면 다음과 유사한 변경 이벤트가 출력됩니다.

ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null, fullDocument=Restaurant(name=Blarney Castle, cuisine=Irish),
fullDocumentBeforeChange=null, documentKey={"_id": {"$oid": "..."}},
clusterTime=Timestamp{value=..., seconds=..., inc=...},
updateDescription=UpdateDescription{removedFields=[], updatedFields={"cuisine": "Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null, splitEvent=null,
wallTime=BsonDateTime{value=...}}

사전 이미지 및 사후 이미지에 대해 자세히 알아보려면 Change Streams 매뉴얼에서 문서 사전 및 사후 이미지로 MongoDB Server 을 참조하세요.

API 문서

이 가이드에서 사용되는 메서드 또는 유형에 대해 자세히 알아보려면 다음 API 설명서를 참조하세요.

• MongoClient.watch()

• MongoDatabase.watch()

• MongoCollection.watch()

• ChangeStreamIterable