простой конвейер Apache Beam генерирует ошибку TypeError: требуется целое число

Question

У меня есть этот простой конвейер apache-beam, написанный на python.

import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
from apache_beam.runners.direct.direct_runner import DirectRunner
from apache_beam.options.pipeline_options import PipelineOptions

p = beam.Pipeline(InteractiveRunner(underlying_runner=DirectRunner()), options=PipelineOptions())

class Foo(beam.DoFn):
  def process(self, element, *args, **kwargs):
    k,v = element
    yield (k , {k:v})


a = [('a',1), ('a',1), ('b',1)]

x0 = p | "0" >> beam.Create(a) | "2" >> beam.ParDo(Foo())
r = p.run()
r.wait_until_finish()
print(r.get(x0))

Выполнение приведенного выше кода вызывает следующую ошибку

Traceback (most recent call last):
  File "apache_beam/runners/common.py", line 883, in apache_beam.runners.common.DoFnRunner.process
  File "apache_beam/runners/common.py", line 497, in apache_beam.runners.common.SimpleInvoker.invoke_process
  File "apache_beam/runners/common.py", line 1028, in apache_beam.runners.common._OutputProcessor.process_outputs
  File "apache_beam/runners/worker/operations.py", line 177, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive
  File "apache_beam/runners/worker/operations.py", line 155, in apache_beam.runners.worker.operations.ConsumerSet.update_counters_start
  File "apache_beam/runners/worker/opcounters.py", line 214, in apache_beam.runners.worker.opcounters.OperationCounters.update_from
  File "apache_beam/runners/worker/opcounters.py", line 233, in apache_beam.runners.worker.opcounters.OperationCounters.do_sample
  File "apache_beam/coders/coder_impl.py", line 1162, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
  File "apache_beam/coders/coder_impl.py", line 1172, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
  File "apache_beam/coders/coder_impl.py", line 768, in apache_beam.coders.coder_impl.AbstractComponentCoderImpl.get_estimated_size_and_observables
  File "apache_beam/coders/coder_impl.py", line 187, in apache_beam.coders.coder_impl.CoderImpl.get_estimated_size_and_observables
  File "apache_beam/coders/coder_impl.py", line 686, in apache_beam.coders.coder_impl.VarIntCoderImpl.estimate_size
  File "apache_beam/coders/stream.pyx", line 225, in apache_beam.coders.stream.get_varint_size
TypeError: an integer is required

Однако тот же код работает, если мы установим a в список списков вместо списка кортежей.

a = [('a',1), ('a',1), ('b',1)] -> a = [['a',1], ['a',1], ['b',1]]

Answer 1

Я подтвердил, что это ошибка. Подано https://issues.apache.org/jira/browse/BEAM-10833