[BEAM-7981] ParDo function wrapper doesn't support Iterable output types - ASF JIRA

XML

Word

Printable

JSON

Details

Type: Bug
Status: Triage Needed
Priority: P2
Resolution: Fixed
Affects Version/s: None
Fix Version/s: 2.18.0
Component/s: sdk-py-core
Labels:
None

Description

I believe the bug is in CallableWrapperDoFn.default_type_hints, which converts Iterable[str] to str.

This test will be included (commented out) in https://github.com/apache/beam/pull/9283

  def test_typed_callable_iterable_output(self):
    @typehints.with_input_types(int)
    @typehints.with_output_types(typehints.Iterable[str])
    def do_fn(element):
      return [[str(element)] * 2]

    result = [1, 2] | beam.ParDo(do_fn)
    self.assertEqual([['1', '1'], ['2', '2']], sorted(result))

Result:

======================================================================
ERROR: test_typed_callable_iterable_output (apache_beam.typehints.typed_pipeline_test.MainInputTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/typehints/typed_pipeline_test.py", line 104, in test_typed_callable_iterable_output
    result = [1, 2] | beam.ParDo(do_fn)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/transforms/ptransform.py", line 519, in __ror__
    p.run().wait_until_finish()
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/pipeline.py", line 406, in run
    self._options).run(False)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/pipeline.py", line 419, in run
    return self.runner.run_pipeline(self, self._options)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/direct/direct_runner.py", line 129, in run_pipeline
    return runner.run_pipeline(pipeline, options)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 366, in run_pipeline
    default_environment=self._default_environment))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 373, in run_via_runner_api
    return self.run_stages(stage_context, stages)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 455, in run_stages
    stage_context.safe_coders)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 733, in _run_stage
    result, splits = bundle_manager.process_bundle(data_input, data_output)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1663, in process_bundle
    part, expected_outputs), part_inputs):
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 586, in result_iterator
    yield fs.pop().result()
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 432, in result
    return self.__get_result()
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/usr/lib/python3.7/concurrent/futures/thread.py", line 57, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1663, in <lambda>
    part, expected_outputs), part_inputs):
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1601, in process_bundle
    result_future = self._worker_handler.control_conn.push(process_bundle_req)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1080, in push
    response = self.worker.do_instruction(request)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/sdk_worker.py", line 343, in do_instruction
    request.instruction_id)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/sdk_worker.py", line 369, in process_bundle
    bundle_processor.process_bundle(instruction_id))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/bundle_processor.py", line 593, in process_bundle
    data.ptransform_id].process_encoded(data.data)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/bundle_processor.py", line 143, in process_encoded
    self.output(decoded_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 256, in output
    cython.cast(Receiver, self.receivers[output_index]).receive(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 143, in receive
    self.consumer.process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 435, in process
    self.output(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 256, in output
    cython.cast(Receiver, self.receivers[output_index]).receive(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 143, in receive
    self.consumer.process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 594, in process
    delayed_application = self.dofn_receiver.receive(o)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 776, in receive
    self.process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 782, in process
    self._reraise_augmented(exn)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 849, in _reraise_augmented
    raise_with_traceback(new_exn)
  File "/usr/local/google/home/ehudm/virtualenvs/beam-py37/lib/python3.7/site-packages/future/utils/__init__.py", line 419, in raise_with_traceback
    raise exc.with_traceback(traceback)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 780, in process
    return self.do_fn_invoker.invoke_process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 441, in invoke_process
    windowed_value, self.process_method(windowed_value.value))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 919, in process_outputs
    self.main_receivers.receive(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 142, in receive
    self.update_counters_start(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 122, in update_counters_start
    self.opcounter.update_from(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/opcounters.py", line 196, in update_from
    self.do_sample(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/opcounters.py", line 214, in do_sample
    self.coder_impl.get_estimated_size_and_observables(windowed_value))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coder_impl.py", line 1024, in get_estimated_size_and_observables
    value.value, nested=nested))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coder_impl.py", line 220, in get_estimated_size_and_observables
    return self.estimate_size(value, nested), []
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coder_impl.py", line 212, in estimate_size
    return self._get_nested_size(self._size_estimator(value), nested)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coders.py", line 135, in estimate_size
    return len(self.encode(value))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coders.py", line 326, in encode
    return value.encode('utf-8')
AttributeError: 'list' object has no attribute 'encode' [while running 'ParDo(CallableWrapperDoFn)']
-------------------- >> begin captured logging << --------------------
root: INFO: Generating grammar tables from /usr/lib/python3.7/lib2to3/Grammar.txt
root: INFO: Generating grammar tables from /usr/lib/python3.7/lib2to3/PatternGrammar.txt
avro.schema: Level 5: Register new name for 'org.apache.avro.file.Header'
avro.schema: Level 5: Register new name for 'org.apache.avro.file.magic'
avro.schema: Level 5: Register new name for 'org.apache.avro.file.sync'
root: INFO: ==================== <function annotate_downstream_side_inputs at 0x7f3a39918158> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function fix_side_input_pcoll_coders at 0x7f3a39918268> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function lift_combiners at 0x7f3a399182f0> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function expand_sdf at 0x7f3a39918378> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function expand_gbk at 0x7f3a39918400> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function sink_flattens at 0x7f3a39918510> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function greedily_fuse at 0x7f3a39918598> ====================
root: DEBUG: 1 [3]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  CreatePInput0/Read:beam:transform:read:v1\nParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function read_to_impulse at 0x7f3a39918620> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read/Impulse:beam:transform:impulse:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function impulse_to_input at 0x7f3a399186a8> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function inject_timer_pcollections at 0x7f3a39918840> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function sort_stages at 0x7f3a399188c8> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function window_pcollection_coders at 0x7f3a39918950> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: Running ((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)
root: DEBUG: start <DoOperation _MaterializeValues0 output_tags=['out'], receivers=[ConsumerSet[_MaterializeValues0.out0, coder=WindowedValueCoder[FastPrimitivesCoder], len(consumers)=0]]>
root: DEBUG: start <DoOperation ParDo(CallableWrapperDoFn) output_tags=['out'], receivers=[SingletonConsumerSet[ParDo(CallableWrapperDoFn).out0, coder=WindowedValueCoder[StrUtf8Coder], len(consumers)=1]]>
root: DEBUG: start <ImpulseReadOperation receivers=[SingletonConsumerSet[CreatePInput0/Read.out0, coder=WindowedValueCoder[VarIntCoder], len(consumers)=1]]>
root: DEBUG: start <DataInputOperation receivers=[SingletonConsumerSet[CreatePInput0/Read/Impulse.out0, coder=WindowedValueCoder[BytesCoder], len(consumers)=1]]>
--------------------- >> end captured logging << ---------------------

----------------------------------------------------------------------
Ran 1 test in 0.068s

FAILED (errors=1)

Error
Traceback (most recent call last):
  File "/usr/lib/python3.7/unittest/case.py", line 59, in testPartExecutor
    yield
  File "/usr/lib/python3.7/unittest/case.py", line 615, in run
    testMethod()
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/typehints/typed_pipeline_test.py", line 104, in test_typed_callable_iterable_output
    result = [1, 2] | beam.ParDo(do_fn)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/transforms/ptransform.py", line 519, in __ror__
    p.run().wait_until_finish()
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/pipeline.py", line 406, in run
    self._options).run(False)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/pipeline.py", line 419, in run
    return self.runner.run_pipeline(self, self._options)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/direct/direct_runner.py", line 129, in run_pipeline
    return runner.run_pipeline(pipeline, options)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 366, in run_pipeline
    default_environment=self._default_environment))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 373, in run_via_runner_api
    return self.run_stages(stage_context, stages)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 455, in run_stages
    stage_context.safe_coders)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 733, in _run_stage
    result, splits = bundle_manager.process_bundle(data_input, data_output)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1663, in process_bundle
    part, expected_outputs), part_inputs):
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 586, in result_iterator
    yield fs.pop().result()
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 432, in result
    return self.__get_result()
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/usr/lib/python3.7/concurrent/futures/thread.py", line 57, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1663, in <lambda>
    part, expected_outputs), part_inputs):
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1601, in process_bundle
    result_future = self._worker_handler.control_conn.push(process_bundle_req)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/portability/fn_api_runner.py", line 1080, in push
    response = self.worker.do_instruction(request)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/sdk_worker.py", line 343, in do_instruction
    request.instruction_id)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/sdk_worker.py", line 369, in process_bundle
    bundle_processor.process_bundle(instruction_id))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/bundle_processor.py", line 593, in process_bundle
    data.ptransform_id].process_encoded(data.data)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/bundle_processor.py", line 143, in process_encoded
    self.output(decoded_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 256, in output
    cython.cast(Receiver, self.receivers[output_index]).receive(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 143, in receive
    self.consumer.process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 435, in process
    self.output(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 256, in output
    cython.cast(Receiver, self.receivers[output_index]).receive(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 143, in receive
    self.consumer.process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 594, in process
    delayed_application = self.dofn_receiver.receive(o)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 776, in receive
    self.process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 782, in process
    self._reraise_augmented(exn)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 849, in _reraise_augmented
    raise_with_traceback(new_exn)
  File "/usr/local/google/home/ehudm/virtualenvs/beam-py37/lib/python3.7/site-packages/future/utils/__init__.py", line 419, in raise_with_traceback
    raise exc.with_traceback(traceback)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 780, in process
    return self.do_fn_invoker.invoke_process(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 441, in invoke_process
    windowed_value, self.process_method(windowed_value.value))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/common.py", line 919, in process_outputs
    self.main_receivers.receive(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 142, in receive
    self.update_counters_start(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/operations.py", line 122, in update_counters_start
    self.opcounter.update_from(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/opcounters.py", line 196, in update_from
    self.do_sample(windowed_value)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/runners/worker/opcounters.py", line 214, in do_sample
    self.coder_impl.get_estimated_size_and_observables(windowed_value))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coder_impl.py", line 1024, in get_estimated_size_and_observables
    value.value, nested=nested))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coder_impl.py", line 220, in get_estimated_size_and_observables
    return self.estimate_size(value, nested), []
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coder_impl.py", line 212, in estimate_size
    return self._get_nested_size(self._size_estimator(value), nested)
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coders.py", line 135, in estimate_size
    return len(self.encode(value))
  File "/usr/local/google/home/ehudm/src/beam/sdks/python/apache_beam/coders/coders.py", line 326, in encode
    return value.encode('utf-8')
Exception: 'list' object has no attribute 'encode' [while running 'ParDo(CallableWrapperDoFn)']
-------------------- >> begin captured logging << --------------------
root: INFO: Generating grammar tables from /usr/lib/python3.7/lib2to3/Grammar.txt
root: INFO: Generating grammar tables from /usr/lib/python3.7/lib2to3/PatternGrammar.txt
avro.schema: Level 5: Register new name for 'org.apache.avro.file.Header'
avro.schema: Level 5: Register new name for 'org.apache.avro.file.magic'
avro.schema: Level 5: Register new name for 'org.apache.avro.file.sync'
root: INFO: ==================== <function annotate_downstream_side_inputs at 0x7f3a39918158> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function fix_side_input_pcoll_coders at 0x7f3a39918268> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function lift_combiners at 0x7f3a399182f0> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function expand_sdf at 0x7f3a39918378> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function expand_gbk at 0x7f3a39918400> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function sink_flattens at 0x7f3a39918510> ====================
root: DEBUG: 3 [1, 1, 1]
root: DEBUG: Stages: ['ref_AppliedPTransform_CreatePInput0/Read_3\n  CreatePInput0/Read:beam:transform:read:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ', 'ref_AppliedPTransform__MaterializeValues0_5\n  _MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function greedily_fuse at 0x7f3a39918598> ====================
root: DEBUG: 1 [3]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  CreatePInput0/Read:beam:transform:read:v1\nParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function read_to_impulse at 0x7f3a39918620> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read/Impulse:beam:transform:impulse:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function impulse_to_input at 0x7f3a399186a8> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function inject_timer_pcollections at 0x7f3a39918840> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function sort_stages at 0x7f3a399188c8> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: ==================== <function window_pcollection_coders at 0x7f3a39918950> ====================
root: DEBUG: 1 [4]
root: DEBUG: Stages: ['((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)\n  ParDo(CallableWrapperDoFn):beam:transform:pardo:v1\n_MaterializeValues0:beam:transform:pardo:v1\nCreatePInput0/Read:beam:transform:read_from_impulse_python:v1\nCreatePInput0/Read/Impulse:beam:source:runner:0.1\n  must follow: \n  downstream_side_inputs: ']
root: INFO: Running ((ref_AppliedPTransform_CreatePInput0/Read_3)+(ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_4))+(ref_AppliedPTransform__MaterializeValues0_5)
root: DEBUG: start <DoOperation _MaterializeValues0 output_tags=['out'], receivers=[ConsumerSet[_MaterializeValues0.out0, coder=WindowedValueCoder[FastPrimitivesCoder], len(consumers)=0]]>
root: DEBUG: start <DoOperation ParDo(CallableWrapperDoFn) output_tags=['out'], receivers=[SingletonConsumerSet[ParDo(CallableWrapperDoFn).out0, coder=WindowedValueCoder[StrUtf8Coder], len(consumers)=1]]>
root: DEBUG: start <ImpulseReadOperation receivers=[SingletonConsumerSet[CreatePInput0/Read.out0, coder=WindowedValueCoder[VarIntCoder], len(consumers)=1]]>
root: DEBUG: start <DataInputOperation receivers=[SingletonConsumerSet[CreatePInput0/Read/Impulse.out0, coder=WindowedValueCoder[BytesCoder], len(consumers)=1]]>
--------------------- >> end captured logging << ---------------------

Attachments

Issue Links

links to

GitHub Pull Request #9708

GitHub Pull Request #11446

Activity

People

Assignee:: Udi Meiri

Reporter:: Udi Meiri

Votes:: 0 Vote for this issue

Watchers:: 2 Start watching this issue

Dates

Created:: 14/Aug/19 23:17

Updated:: 13/Apr/23 10:58

Resolved:: 22/Apr/20 17:09

Time Tracking

Estimated:

Not Specified

Remaining:

Logged:

2h 50m