Użycie narzędzia do szacowania strumienia zmian

W tym artykule opisano, jak można monitorować postęp wystąpień procesora zestawienia zmian podczas odczytywania zestawienia zmian.

Dlaczego postęp monitorowania jest ważny?

Procesor zestawienia zmian działa jako wskaźnik, który porusza się do przodu w kanale zmian i przekazuje zmiany do implementacji delegata.

Twoje wdrożenie procesora strumienia zmian może przetwarzać zmiany z określoną szybkością w oparciu o dostępne zasoby, takie jak procesor CPU, pamięć, sieć itd.

Jeśli ta szybkość jest niższa niż szybkość, w jakiej zmiany występują w kontenerze usługi Azure Cosmos DB, procesor zacznie się opóźniać.

Zidentyfikowanie tego scenariusza pomaga zrozumieć, czy musimy skalować wdrożenie procesora strumienia zmian.

Implementacja estymatora zmian w strumieniu danych

.NET

Jako tryb push dla powiadomień automatycznych

Narzędzie do szacowania zestawienia zmian może działać jako model oparty na wypychaniu, podobnie jak procesor zestawienia zmian. Narzędzie do szacowania mierzy różnicę między ostatnim przetworzonym elementem (zdefiniowanym przez stan kontenera dzierżaw) a najnowszą zmianą w kontenerze oraz przekazuje tę wartość do delegata. Interwał, w którym pomiar jest wykonywany, można również dostosować z wartością domyślną wynoszącą 5 sekund.

Przykładowo, jeśli procesor strumienia zmian używa trybu najnowszej wersji i jest zdefiniowany w następujący sposób:

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();

Właściwym sposobem na zainicjowanie estymatora do pomiaru tego procesora jest użycie GetChangeFeedEstimatorBuilder w następujący sposób:

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

Gdzie procesor i narzędzie do szacowania mają taką samą leaseContainer i taką samą nazwę.

Pozostałe dwa parametry to delegat, który otrzymuje liczbę, która reprezentuje liczbę oczekujących zmian do odczytania przez procesor, oraz interwał czasu, w którym ma zostać podjęta ta miara.

Przykładem delegata, który otrzymuje oszacowanie, jest:

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);
}

Możesz wysłać to oszacowanie do rozwiązania do monitorowania i użyć go, aby zrozumieć, jak postęp działa w czasie.

Jako szczegółowe szacowanie na żądanie

W przeciwieństwie do modelu wypychania istnieje model alternatywny umożliwiający uzyskanie oszacowania na żądanie. Ten model zawiera również bardziej szczegółowe informacje:

• Szacowane opóźnienie na dzierżawę.
• Każda instancja posiada i przetwarza każdą dzierżawę, dzięki czemu możesz zidentyfikować, czy występuje problem w instancji.

Jeśli procesor przesyłu zmian jest zdefiniowany w następujący sposób:

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();

Estymator można utworzyć przy użyciu tej samej konfiguracji dzierżawy.

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

I za każdym razem, gdy chcesz, z wymaganą częstotliwością, możesz uzyskać szczegółowe oszacowanie:

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.");
    }
}

Każdy ChangeFeedProcessorState zawiera informacje o dzierżawie i opóźnieniu, a także o tym, która bieżąca instancja jest jego właścicielem.

Wdrażanie modułu szacującego

Narzędzie do szacowania zestawienia zmian nie musi być wdrażane jako część procesora zestawienia zmian ani być częścią tego samego projektu. Zalecamy wdrożenie narzędzia do szacowania na niezależnym wystąpieniu z procesorów. Pojedyncze wystąpienie estimatora może śledzić postęp dla dzierżaw i wystąpień we wdrożeniu procesora przepływu zmian.

Każde oszacowanie zużywa jednostki żądań z monitorowanych kontenerów i dzierżaw. Częstotliwość co 1 minutę jest dobrym punktem wyjścia, im niższa częstotliwość, tym większe zużycie jednostek żądania.

Java

W tym przykładzie użyto procesora zestawienia zmian w trybie najnowszej wersji z oszacowaniem opóźnienia w przetwarzaniu zdarzeń zestawienia zmian. W aplikacji dokumenty są wstawiane do jednego kontenera (kontener zmian), a tymczasem inny wątek roboczy lub aplikacja robocza pobiera wstawione dokumenty z kanału zmian kontenera i przetwarza je w określony sposób.

Procesor zestawienia zmian został skompilowany i uruchomiony w następujący sposób:

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;
}

Sprawdzanie opóźnienia procesora strumienia zmian można przeprowadzić w oddzielnej aplikacji (takiej jak monitor stanu), pod warunkiem, że są używane te same kontenery wejściowe (feedContainer i leaseContainer) oraz dokładnie ten sam prefiks dzierżawy (CONTAINER_NAME + "-lease"). Kod narzędzia do szacowania wymaga, aby procesor strumienia zmian miał możliwość pełnego zainicjowania dokumentów leaseContainer.

Narzędzie do szacowania oblicza skumulowane opóźnienie, pobierając bieżący stan procesora zestawienia zmian i sumując szacowane wartości opóźnień dla każdego przetwarzanego zdarzenia:

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());

Łączne opóźnienie początkowo powinno wynosić zero, a ostatecznie powinno być większe lub równe liczbie utworzonych dokumentów. Łączną wartość opóźnienia można rejestrować lub wykorzystywać do dalszej analizy, co umożliwia monitorowanie wydajności przetwarzania strumienia zmian oraz identyfikowanie potencjalnych wąskich gardeł lub opóźnień w systemie.

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());

Przykładem delegata, który odbiera zmiany i obsługuje je z opóźnieniem, jest:

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()");
    };
}

Obsługiwane tryby zestawienia zmian

Narzędzie do szacowania przepływu zmian może być używane zarówno w trybie najnowszej wersji, jak i w trybie wszystkich wersji i usunięć. W obu trybach podane oszacowanie nie gwarantuje dokładnej liczby zaległych zmian w procesie.

Dodatkowe zasoby

• Azure Cosmos DB SDK
• Przykłady użycia w witrynie GitHub (najnowsza wersja platformy .NET)
• Przykłady użycia na GitHubie (.NET - wszystkie wersje i przykłady usunięć)
• Przykłady użycia w witrynie GitHub (Java)
• Dodatkowe przykłady w witrynie GitHub

Dalsze kroki

Teraz możesz dowiedzieć się więcej o procesorze śledzenia zmian w poniższym artykule.

• Omówienie procesora strumienia zmian