我有一個用 python 編寫的簡單的 apache-beam 管道。import apache_beam as beamfrom apache_beam.runners.interactive.interactive_runner import InteractiveRunnerfrom apache_beam.runners.direct.direct_runner import DirectRunnerfrom apache_beam.options.pipeline_options import PipelineOptionsp = 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_observablesa但是，如果我們設置為列表列表而不是元組列表，則相同的代碼可以工作。a = [('a',1), ('a',1), ('b',1)]-->a = [['a',1], ['a',1], ['b',1]]