Uploaded image for project: 'Beam'
  1. Beam
  2. BEAM-7981

ParDo function wrapper doesn't support Iterable output types

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Resolved
    • 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

          Activity

            People

            • Assignee:
              udim Udi Meiri
              Reporter:
              udim Udi Meiri
            • Votes:
              0 Vote for this issue
              Watchers:
              2 Start watching this issue

              Dates

              • Created:
                Updated:
                Resolved:

                Time Tracking

                Estimated:
                Original Estimate - Not Specified
                Not Specified
                Remaining:
                Remaining Estimate - 0h
                0h
                Logged:
                Time Spent - 2h 50m
                2h 50m