Azure Cosmos DB의 API for MongoDB에서 변경 스트림

적용 대상: MongoDB

Azure Cosmos DB의 API for MongoDB의 변경 피드 지원은 변경 스트림 API를 사용하여 제공됩니다. 변경 스트림 API를 사용하여 애플리케이션에서 컬렉션 또는 단일 분할된 데이터베이스의 항목에 대한 변경 내용을 가져올 수 있습니다. 나중에 결과에 따라 추가 작업을 수행할 수 있습니다. 컬렉션의 항목에 대한 변경 내용은 수정 시간 순서대로 캡처되고 분할된 데이터베이스 키당 정렬 순서가 보장됩니다.

참고 항목

변경 스트림을 사용하려면 버전 3.6 이상의 Azure Cosmos DB 계정을 만듭니다. 이전 버전에 대해 변경 스트림 예제를 실행하는 경우 인식할 수 없는 파이프라인 단계 이름: $changeStream오류가 표시될 수 있습니다.

예제

다음 예에서는 컬렉션의 모든 항목에 대한 변경 스트림을 가져오는 방법을 보여 줍니다. 이 예에서는 항목이 삽입, 업데이트 또는 교체될 때 항목을 감시하는 커서를 만듭니다. 변경 스트림을 가져오려면 $match 단계, $project 단계 및 fullDocument 옵션이 필요합니다. 변경 스트림을 사용한 삭제 작업에 대한 감시는 현재 지원되지 않습니다. 이 문제를 해결하려면 삭제할 항목에 소프트 마커를 추가하면 됩니다. 예를 들어 "deleted"라는 항목에 특성을 추가할 수 있습니다. 항목을 삭제하려면 "삭제"를 true로 설정하고 항목에 TTL을 설정할 수 있습니다. “삭제됨”을 true로 업데이트하는 것은 업데이트 작업이므로 변경 스트림에 이 변경 내용이 표시됩니다.

JavaScript

var cursor = db.coll.watch(
    [
        { $match: { "operationType": { $in: ["insert", "update", "replace"] } } },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1 } }
    ],
    { fullDocument: "updateLookup" });

while (!cursor.isExhausted()) {
    if (cursor.hasNext()) {
        printjson(cursor.next());
    }
}

C#

var collection = new MongoClient("<connection-string>")
    .GetDatabase("<database-name>")
    .GetCollection<BsonDocument>("<collection-name>");

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change => 
        change.OperationType == ChangeStreamOperationType.Insert || 
        change.OperationType == ChangeStreamOperationType.Update || 
        change.OperationType == ChangeStreamOperationType.Replace
    )
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
        @"{ 
            $project: { 
                '_id': 1, 
                'fullDocument': 1, 
                'ns': 1, 
                'documentKey': 1 
            }
        }"
    );

ChangeStreamOptions options = new ()
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup
};

using IChangeStreamCursor<BsonDocument> enumerator = collection.Watch(
    pipeline, 
    options
);

Console.WriteLine("Waiting for changes...");
while (enumerator.MoveNext())
{
    IEnumerable<BsonDocument> changes = enumerator.Current;
    foreach(BsonDocument change in changes)
    {
        Console.WriteLine(change);
    }  
}

Java

다음 예제에서는 Java에서 스트림 변경 기능을 사용하는 방법을 보여줍니다. 전체 예제는 이 GitHub 리포지토리를 참조하세요. 또한 이 예제에서는 resumeAfter 메서드를 사용하여 마지막 읽기부터 모든 변경 내용을 검색하는 방법을 보여줍니다.

Bson match = Aggregates.match(Filters.in("operationType", asList("update", "replace", "insert")));

// Pick the field you are most interested in
Bson project = Aggregates.project(fields(include("_id", "ns", "documentKey", "fullDocument")));

// This variable is for second example
BsonDocument resumeToken = null;

// Now time to build the pipeline
List<Bson> pipeline = Arrays.asList(match, project);

//#1 Simple example to seek changes

// Create cursor with update_lookup
MongoChangeStreamCursor<ChangeStreamDocument<org.bson.Document>> cursor = collection.watch(pipeline)
        .fullDocument(FullDocument.UPDATE_LOOKUP).cursor();

Document document = new Document("name", "doc-in-step-1-" + Math.random());
collection.insertOne(document);

while (cursor.hasNext()) {
    // There you go, we got the change document.
    ChangeStreamDocument<Document> csDoc = cursor.next();

    // Let is pick the token which will help us resuming
    // You can save this token in any persistent storage and retrieve it later
    resumeToken = csDoc.getResumeToken();
    //Printing the token
    System.out.println(resumeToken);
    
    //Printing the document.
    System.out.println(csDoc.getFullDocument());
    //This break is intentional but in real project feel free to remove it.
    break;
}

cursor.close();

단일 분할된 데이터베이스가 포함된 변경 내용

다음 예에서는 단일 분할된 데이터베이스 내의 항목에 대한 변경 내용을 가져오는 방법을 보여 줍니다. 이 예에서는 분할 키가 “a”와 같고 분할된 데이터베이스 키 값은 “1”과 같은 항목의 변경 내용을 가져옵니다. 다른 클라이언트에서 다른 분할된 데이터베이스의 변경 내용을 병렬로 읽을 수 있습니다.

var cursor = db.coll.watch(
    [
        { 
            $match: { 
                $and: [
                    { "fullDocument.a": 1 }, 
                    { "operationType": { $in: ["insert", "update", "replace"] } }
                ]
            }
        },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1} }
    ],
    { fullDocument: "updateLookup" });

변경 스트리밍 크기 조정

대규모 변경 스트리밍을 사용할 때는 부하를 고르게 분산하는 것이 가장 좋습니다. GetChangeStreamTokens 사용자 지정 명령을 사용하여 실제 분할/파티션에 부하를 분산합니다.

현재 제한 사항

변경 스트림을 사용하는 경우 다음과 같은 제한이 적용될 수 있습니다.

• operationType 및 updateDescription 속성은 출력 문서에서 아직 지원되지 않습니다.
• 현재 insert, update 및 replace 작업 유형이 지원됩니다. 그러나 삭제 작업 또는 기타 이벤트는 아직 지원되지 않습니다.

해당 제한 사항으로 인해 이전 예제에 나와 있는 것처럼 $match 단계, $project 단계 및 fullDocument 옵션이 필요합니다.

Azure Cosmos DB의 NoSQL용 API에 있는 변경 피드와 달리, 변경 스트림을 사용할 별도의 변경 피드 프로세서 라이브러리 또는 임대 컨테이너를 사용할 필요성이 없습니다. 현재 변경 스트림을 처리하기 위한 Azure Functions 트리거는 지원되지 않습니다.

오류 처리

변경 스트림을 사용할 때 지원되는 오류 코드와 메시지는 다음과 같습니다.

• HTTP 오류 코드 16500 - 변경 스트림이 제한되면 빈 페이지를 반환합니다.

• NamespaceNotFound(OperationType 무효화) - 존재하지 않는 컬렉션에서 변경 스트림을 실행하거나 컬렉션을 삭제하면 NamespaceNotFound 오류가 반환됩니다. operationType 속성은 operationType Invalidate 오류 대신 출력 문서에서 반환될 수 없으므로 NamespaceNotFound 오류가 반환됩니다.

다음 단계

• Azure Cosmos DB의 API for MongoDB에서 TTL(Time to live)을 사용하여 자동으로 데이터 만료
• Azure Cosmos DB의 API for MongoDB에서 인덱싱