변경 피드 추정기 사용

적용 대상: NoSQL

이 문서에서는 변경 피드 프로세서 인스턴스가 변경 피드를 읽을 때 진행률을 모니터링하는 방법을 설명합니다.

진행률 모니터링이 중요한 이유는 무엇인가요?

변경 피드 프로세서는 변경 피드를 통해 앞으로 이동하여 변경 내용을 대리자 구현에 전달하는 포인터 역할을 합니다.

변경 피드 프로세서 배포는 CPU, 메모리, 네트워크 등의 사용 가능한 리소스를 기반으로 특정 속도로 변경 내용을 처리할 수 있습니다.

이 속도가 Azure Cosmos DB 컨테이너에서 변경이 발생하는 속도보다 느리면 프로세서가 뒤처지기 시작합니다.

이 시나리오를 파악하면 변경 피드 프로세서 배포를 확장해야 하는지 여부를 파악할 수 있습니다.

변경 피드 추정기 구현

.NET

자동 알림을 위한 푸시 모델

변경 피드 프로세서와 마찬가지로, 변경 피드 추정기는 푸시 모델로 작동할 수 있습니다. 추정기는 마지막으로 처리된 항목(임대 컨테이너의 상태를 기준으로 정의)과 컨테이너의 마지막 변경 내용의 차이를 측정하고, 이 값을 대리자에게 푸시합니다. 측정 간격을 사용자 지정할 수 있으며, 기본값은 5초입니다.

예를 들어, 변경 피드 프로세서가 최신 버전 모드를 사용하고 다음과 같이 정의된 경우:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

추정기를 초기화하여 해당 프로세서를 측정하는 올바른 방법은 다음과 같이 GetChangeFeedEstimatorBuilder를 사용하는 것입니다.

ChangeFeedProcessor changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
    .WithLeaseContainer(leaseContainer)
    .Build();

프로세서와 추정기는 동일한 leaseContainer 및 동일한 이름을 공유합니다.

다른 두 매개 변수는 대리자이며, 프로세서에서 읽을 보류 중인 변경 수 및 원하는 시간 간격을 나타내는 숫자를 수신합니다.

예측을 수신하는 대리자의 예는 다음과 같습니다.

static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
{
    if (estimation > 0)
    {
        Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
    }

    await Task.Delay(0);
}

이 예측을 모니터링 솔루션으로 전송하여 시간이 지남에 따라 진행률이 어떻게 되는지 이해할 수 있습니다.

주문형 세부 예측

푸시 모델과 달리 수요에 따라 예측을 얻을 수 있는 대안이 있습니다. 또한 이 모델은 보다 자세한 정보를 제공합니다.

• 임대당 예상 지연 시간입니다.
• 인스턴스가 각 임대를 소유하고 처리하므로 인스턴스에 문제가 있는지 확인할 수 있습니다.

변경 피드 프로세서가 다음과 같이 정의된 경우:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

동일한 임대 구성으로 추정기를 만들 수 있습니다.

ChangeFeedEstimator changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);

원할 때마다 필요한 빈도로 자세한 추정치를 얻을 수 있습니다.

Console.WriteLine("Checking estimation...");
using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
while (estimatorIterator.HasMoreResults)
{
    FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
    foreach (ChangeFeedProcessorState leaseState in states)
    {
        string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
        Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
    }
}

각 ChangeFeedProcessorState에는 임대 및 지연 정보와 해당 정보를 소유하는 현재 인스턴스가 포함됩니다.

예측 도구 배포

변경 피드 추정기를 변경 피드 프로세서의 일부로 배포하거나 동일한 프로젝트에 포함할 필요가 없습니다. 프로세서와 독립된 인스턴스에 예측 도구를 배포하는 것이 좋습니다. 단일 예측 도구 인스턴스는 변경 피드 프로세서 배포의 모든 임대 및 인스턴스에 대한 진행률을 추적할 수 있습니다.

각 예측은 모니터링 및 임대 컨테이너에서 요청 단위가 사용됩니다. 1분 사이의 빈도는 좋은 시작점이며, 빈도가 낮을수록 소비된 요청 단위가 높습니다.

Java

이 예에서는 최신 버전 모드에서 변경 피드 프로세서를 사용하여 변경 피드 이벤트 처리 지연을 예측합니다. 애플리케이션에서 문서는 하나의 컨테이너("피드 컨테이너")에 삽입되고 다른 작업자 스레드 또는 작업자 애플리케이션은 피드 컨테이너의 변경 피드에서 삽입된 문서를 끌어와 어떤 식으로든 작업합니다.

변경 피드 프로세서는 다음과 같이 빌드되고 시작됩니다.

ChangeFeedProcessor changeFeedProcessorMainInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChangesWithLag())
    .buildChangeFeedProcessor();

try {
    changeFeedProcessorMainInstance
        .start()
        .subscribeOn(Schedulers.boundedElastic())
        .timeout(Duration.ofSeconds(10))
        .then(Mono.just(changeFeedProcessorMainInstance)
                  .delayElement(Duration.ofSeconds(10))
                  .flatMap(value -> changeFeedProcessorMainInstance
                      .stop()
                      .subscribeOn(Schedulers.boundedElastic())
                      .timeout(Duration.ofSeconds(10))))
        .subscribe();
} catch (Exception ex) {
    logger.error("Change feed processor did not start and stopped in the expected time", ex);
    throw ex;
}

변경 피드 프로세서 지연 검사는 동일한 입력 컨테이너(feedContainer 및 leaseContainer)와 정확히 동일한 임대 접두사(CONTAINER_NAME + "-lease")가 사용되는 한 별도의 애플리케이션(예: 상태 모니터)에서 수행할 수 있습니다. 예측 도구 코드를 사용하려면 변경 피드 프로세서에서 leaseContainer의 문서를 완전히 초기화할 수 있어야 합니다.

예측 도구는 변경 피드 프로세서의 현재 상태를 검색하고 처리 중인 각 이벤트에 대한 예측 지연 값을 합산하여 누적 지연을 계산합니다.

Mono<List<ChangeFeedProcessorState>> currentState = changeFeedProcessorMainInstance.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

totalLag.set(0);
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

처음에는 총 지연이 0이어야 하며 마지막에는 만든 문서 수보다 크거나 같아야 합니다. 총 지연 값을 기록하거나 추가 분석에 사용할 수 있으므로 변경 피드 처리의 성능을 모니터링하고 시스템의 잠재적인 병목 상태 또는 지연을 식별할 수 있습니다.

totalLag.set(0);
changeFeedProcessorSideCart.start().block();
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).block();

// Finally, totalLag should be greater or equal to the number of documents created
logger.info("Finally total lag is : {}", totalLag.get());

변경 내용을 수신하고 지연을 통해 처리하는 대리자의 예는 다음과 같습니다.

private static Consumer<List<JsonNode>> handleChangesWithLag() {
    return (List<JsonNode> docs) -> {
        logger.info("Start handleChangesWithLag()");

        try {
            Thread.sleep(Duration.ofSeconds(5).toMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        for (JsonNode document : docs) {
            try {
                //Change Feed hands the document to you in the form of a JsonNode
                //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                //especially if you do not have a single uniform data model for all documents.
                logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                                                 .writeValueAsString(document));

                //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                //as shown below. Then you can operate on the POJO.
                CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                logger.info("id: " + pojo_doc.getId());

            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        logger.info("End handleChangesWithLag()");
    };
}

지원되는 변경 피드 모드

변경 피드 예측 도구는 최신 버전 모드와 모든 버전 및 삭제 모드 모두에 사용할 수 있습니다. 두 가지 모드 모두에서 제공되는 예상 비용은 프로세스에 대한 미처리 변경 내용의 정확한 수를 보장하지 않습니다.

추가 리소스

• Azure Cosmos DB SDK
• GitHub의 사용 샘플(.NET 최신 버전)
• GitHub의 사용 샘플(.NET 모든 버전 및 삭제)
• GitHub의 사용 샘플(Java)
• GitHub의 추가 샘플

다음 단계

이제 다음 문서에서 변경 피드 프로세서에 대해 자세히 알아볼 수 있습니다.

• 변경 피드 프로세서 개요