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

[Anonymous]

Я пытаюсь понять этот пример общего кэша: https://beam.apache.org/documentation/p ... red-class/
Я добавил этот счетчик в инициализация:

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

> self.cache_load_counter = Metrics.counter(self.__class__, 'cache_loads')

И, соответственно, увеличивайте в пределах дорогих_remote_call_to_load_customers:

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

> self.cache_load_counter.inc()

Я пишу тест, чтобы убедиться, что кеш загружается должным образом, т. е. через max_stale секунд кеш должен перезагрузиться. Это правильно отражается в журнале, но кажется, что счетчик не сохраняется между пакетами. Журнал для справки:

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

> pardo
> bundle started
> Loading rule cache.
> Cache loaded with tag: 1734739200.0
> bundle started
> Loading rule cache.
> Cache loaded with tag: 1734742800.0
> Cache load counter: 1

Поток отправляется через TestStream с одним набором элементов, отправленным со значением 0, с продвижением водяного знака и отправкой другого набора. Течение времени имитируется с помощью ложного патча, чтобы заставить тег измениться.
Похоже, что счетчик сбрасывается между пакетами. Я ожидаю, что, поскольку в функции «Загрузка кэша правил» есть приращение счетчика, счетчик загрузки кэша должен быть равен 2. Почему этот счетчик возвращает 1?
Другая связанная проблема заключается в том, что при запуске пакета можно проверить, что каждый рабочий процесс загружает кеш (поэтому num_workers == кэш_загрузок), используя эти параметры

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

> options.\_all_options\['direct_num_workers'\] = num_workers
> options.\_all_options\['direct_running_mode'\] = 'multi_processing'

Это не работает для потока. Есть ли предпочтительный способ имитировать несколько рабочих для потока, который отличается от пакетного? Я попробовал приведенное ниже, и, похоже, он все еще использует только одного работника.

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

> options.view_as(beam.options.pipeline_options.WorkerOptions).num_workers = num_workers

Спасибо. Если я смогу что-то уточнить, дайте мне знать.

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