Apache Beam DoFn Init: почему значения инициализации сбрасываются между входными потоками в DirectRunner? ⇐ Python

[Anonymous]

Я пытаюсь понять жизненный цикл DoFn более подробно.
Я добавил этот счетчик в init моего DoFn:

Код: Выделить всё

> self.counter = Metrics.counter(self.__class__, 'counts')

И соответственно увеличивать внутри соответствующего процесса()

Код: Выделить всё

> self.counter.inc()

Для простоты цель состоит в том, чтобы подсчитать общее количество элементов, прочитанных по всему потоку, т. е. сохранить метрику во всех вызовах DoFn. Я вижу, что при увеличении водяного знака/времени обработки значение счетчика сбрасывается. Журнал для справки:

Код: Выделить всё

> pardo start
> bundle started
> incremented counter.
> bundle started
> incremented counter.
> counter value: 1

Я ожидаю, что этот счетчик будет равен 2. Я могу увидеть правильное увеличение счетчика, если отправлю поток из нескольких элементов без искусственного разрыва во времени, увеличиваясь, как ожидалось. Почему происходит сброс при продвижении водяного знака? Разве метрики не должны использоваться для агрегированной статистики?
Поток отправляется через TestStream, при этом один набор элементов отправляется со значением 0, продвигается водяной знак и отправляется другой набор.
Кроме того, обратите внимание, что метрики правильно агрегируются по входным потокам при тестировании с использованием DataflowRunner, поэтому, похоже, это связано только с DirectRunner.
Спасибо, и, пожалуйста, дайте мне знать, если я могу уточнить где угодно.

Подробнее здесь: https://stackoverflow.com/questions/793 ... on-directr