변경 피드 추정기 사용

아티클
02/12/2024

적용 대상: NoSQL

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

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

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

변경 피드 프로세서 배포는 CPU, 메모리, 네트워크 등과 같은 사용 가능한 리소스에 따라 특정 속도로 변경 내용을 처리할 수 있습니다.

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

이 시나리오를 식별하면 변경 피드 프로세서 배포의 크기를 조정해야 하는지 이해하는 데 도움이 됩니다.

자동 알림에 대한 푸시 모델

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

추가 리소스

다음 단계

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

변경 피드 프로세서 개요