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

TypeError: Expected bytes, got list in apache_beam.coders.coder_impl.BytesCoderImpl.encode_to_stream

Details

    • Bug
    • Status: Resolved
    • P3
    • Resolution: Won't Fix
    • None
    • Missing
    • None
    • Flink 1.10.1
      Beam worker pool: apache/beam_python3.6_sdk:2.22.0
      SDK: apache beam 2.22.0
      Python 3.6

    Description

      I am trying to run https://github.com/tensorflow/tfx/tree/master/tfx/examples/chicago_taxi_pipeline using Flink Runner on Kubernetes. 

      I resolved a few issues and finally make beam-worker-pool to pick up the tasks. Then, I get follow error messages

       

      // code placeholder
      
      Traceback (most recent call last):
        File "/usr/local/lib/python3.6/threading.py", line 916, in _bootstrap_inner
          self.run()
        File "/usr/local/lib/python3.6/threading.py", line 864, in run
          self._target(*self._args, **self._kwargs)
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/data_plane.py", line 545, in <lambda>
          target=lambda: self._read_inputs(elements_iterator),
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/data_plane.py", line 528, in _read_inputs
          for elements in elements_iterator:
        File "/usr/local/lib/python3.6/site-packages/grpc/_channel.py", line 416, in __next__
          return self._next()
        File "/usr/local/lib/python3.6/site-packages/grpc/_channel.py", line 689, in _next
          raise self
      grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
              status = StatusCode.UNAVAILABLE
              details = "DNS resolution failed for service: "
              debug_error_string = "{"created":"@1598390105.406081334","description":"Resolver transient failure","file":"src/core/ext/filters/client_channel/resolving_lb_policy.cc","file_line":213,"referenced_errors":[{"created":"@1598390105.406079944","description":"DNS resolution failed for service: ","file":"src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc","file_line":377,"grpc_status":14,"referenced_errors":[{"created":"@1598390105.406076335","description":"unparseable host:port","file":"src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc","file_line":410,"target_address":""}]}]}"
      >Traceback (most recent call last):
        File "apache_beam/runners/common.py", line 961, in apache_beam.runners.common.DoFnRunner.process
        File "apache_beam/runners/common.py", line 553, in apache_beam.runners.common.SimpleInvoker.invoke_process
        File "apache_beam/runners/common.py", line 1122, in apache_beam.runners.common._OutputProcessor.process_outputs
        File "apache_beam/runners/worker/operations.py", line 194, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive
        File "apache_beam/runners/worker/operations.py", line 164, in apache_beam.runners.worker.operations.ConsumerSet.update_counters_start
        File "apache_beam/runners/worker/opcounters.py", line 210, in apache_beam.runners.worker.opcounters.OperationCounters.update_from
        File "apache_beam/runners/worker/opcounters.py", line 230, in apache_beam.runners.worker.opcounters.OperationCounters.do_sample
        File "apache_beam/coders/coder_impl.py", line 1187, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 1198, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 1000, in apache_beam.coders.coder_impl.SequenceCoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 192, in apache_beam.coders.coder_impl.CoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 165, in apache_beam.coders.coder_impl.CoderImpl.estimate_size
        File "apache_beam/coders/coder_impl.py", line 488, in apache_beam.coders.coder_impl.BytesCoderImpl.encode_to_stream
      TypeError: Expected bytes, got listDuring handling of the above exception, another exception occurred:Traceback (most recent call last):
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", line 245, in _execute
          response = task()
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", line 302, in <lambda>
          lambda: self.create_worker().do_instruction(request), request)
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", line 471, in do_instruction
          getattr(request, request_type), request.instruction_id)
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", line 506, in process_bundle
          bundle_processor.process_bundle(instruction_id))
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", line 972, in process_bundle
          element.data)
        File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", line 218, in process_encoded
          self.output(decoded_value)
        File "apache_beam/runners/worker/operations.py", line 330, in apache_beam.runners.worker.operations.Operation.output
        File "apache_beam/runners/worker/operations.py", line 332, in apache_beam.runners.worker.operations.Operation.output
        File "apache_beam/runners/worker/operations.py", line 195, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive
        File "apache_beam/runners/worker/operations.py", line 755, in apache_beam.runners.worker.operations.SdfProcessSizedElements.process
        File "apache_beam/runners/worker/operations.py", line 764, in apache_beam.runners.worker.operations.SdfProcessSizedElements.process
        File "apache_beam/runners/common.py", line 971, in apache_beam.runners.common.DoFnRunner.process_with_sized_restriction
        File "apache_beam/runners/common.py", line 711, in apache_beam.runners.common.PerWindowInvoker.invoke_process
        File "apache_beam/runners/common.py", line 807, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
        File "apache_beam/runners/common.py", line 1122, in apache_beam.runners.common._OutputProcessor.process_outputs
        File "apache_beam/runners/worker/operations.py", line 195, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive
        File "apache_beam/runners/worker/operations.py", line 670, in apache_beam.runners.worker.operations.DoOperation.process
        File "apache_beam/runners/worker/operations.py", line 671, in apache_beam.runners.worker.operations.DoOperation.process
        File "apache_beam/runners/common.py", line 963, in apache_beam.runners.common.DoFnRunner.process
        File "apache_beam/runners/common.py", line 1045, in apache_beam.runners.common.DoFnRunner._reraise_augmented
        File "/usr/local/lib/python3.6/site-packages/future/utils/__init__.py", line 446, in raise_with_traceback
          raise exc.with_traceback(traceback)
        File "apache_beam/runners/common.py", line 961, in apache_beam.runners.common.DoFnRunner.process
        File "apache_beam/runners/common.py", line 553, in apache_beam.runners.common.SimpleInvoker.invoke_process
        File "apache_beam/runners/common.py", line 1122, in apache_beam.runners.common._OutputProcessor.process_outputs
        File "apache_beam/runners/worker/operations.py", line 194, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive
        File "apache_beam/runners/worker/operations.py", line 164, in apache_beam.runners.worker.operations.ConsumerSet.update_counters_start
        File "apache_beam/runners/worker/opcounters.py", line 210, in apache_beam.runners.worker.opcounters.OperationCounters.update_from
        File "apache_beam/runners/worker/opcounters.py", line 230, in apache_beam.runners.worker.opcounters.OperationCounters.do_sample
        File "apache_beam/coders/coder_impl.py", line 1187, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 1198, in apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 1000, in apache_beam.coders.coder_impl.SequenceCoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 192, in apache_beam.coders.coder_impl.CoderImpl.get_estimated_size_and_observables
        File "apache_beam/coders/coder_impl.py", line 165, in apache_beam.coders.coder_impl.CoderImpl.estimate_size
        File "apache_beam/coders/coder_impl.py", line 488, in apache_beam.coders.coder_impl.BytesCoderImpl.encode_to_stream
      TypeError: Expected bytes, got list [while running 'InputToSerializedExample/InputSourceToExample/ParseCSVLine']2020/08/25 21:15:10 Python exited: <nil>
      
      

      This code can run using DirectRunner without any problems..Not sure why it doesn't work on Flink.  TFX Job Information

       

       - --pipeline_name      - parameterized_tfx_oss      - --pipeline_root      - '{{inputs.parameters.pipeline-root}}'      - --kubeflow_metadata_config      - |-        {          "grpc_config": {            "grpc_service_host": {              "environment_variable": "METADATA_GRPC_SERVICE_SERVICE_HOST"            },            "grpc_service_port": {              "environment_variable": "METADATA_GRPC_SERVICE_SERVICE_PORT"            }          }        }      - --beam_pipeline_args      - '["--runner=FlinkRunner", "--flink_master=beam-flink-cluster-jobmanager:8081",        "--flink_submit_uber_jar", "--environment_type=EXTERNAL", "--environment_config=localhost:50000"]'      - --additional_pipeline_args      - '{}'      - --component_launcher_class_path      - tfx.orchestration.launcher.in_process_component_launcher.InProcessComponentLauncher      - --serialized_component      - '{"__class__": "NodeWrapper", "__module__": "tfx.orchestration.kubeflow.node_wrapper",        "__tfx_object_type__": "jsonable", "_exec_properties": {"custom_config": null,        "input_config": "{\n  \"splits\": [\n    {\n      \"name\": \"single_split\",\n      \"pattern\":        \"*\"\n    }\n  ]\n}", "output_config": "{\n  \"split_config\": {\n    \"splits\":        [\n      {\n        \"hash_buckets\": 2,\n        \"name\": \"train\"\n      },\n      {\n        \"hash_buckets\":        1,\n        \"name\": \"eval\"\n      }\n    ]\n  }\n}"}, "_id": "CsvExampleGen",        "_inputs": {"__class__": "_PropertyDictWrapper", "__module__": "tfx.types.node_common",        "__tfx_object_type__": "jsonable", "_compat_aliases": {"input_base": "input"},        "_data": {"input": {"__class__": "Channel", "__module__": "tfx.types.channel",        "__tfx_object_type__": "jsonable", "artifacts": [{"__artifact_class_module__":        "tfx.types.standard_artifacts", "__artifact_class_name__": "ExternalArtifact",        "artifact": {"uri": "{{inputs.parameters.data-root}}"}, "artifact_type": {"name":        "ExternalArtifact"}}], "output_key": null, "producer_component_id": null,        "type": {"name": "ExternalArtifact"}}}}, "_outputs": {"__class__": "_PropertyDictWrapper",        "__module__": "tfx.types.node_common", "__tfx_object_type__": "jsonable",        "_compat_aliases": {}, "_data": {"examples": {"__class__": "Channel", "__module__":        "tfx.types.channel", "__tfx_object_type__": "jsonable", "artifacts": [{"__artifact_class_module__":        "tfx.types.standard_artifacts", "__artifact_class_name__": "Examples", "artifact":        {"custom_properties": {"name": {"string_value": "examples"}, "pipeline_name":        {"string_value": "parameterized_tfx_oss"}, "producer_component": {"string_value":        "CsvExampleGen"}}, "properties": {"split_names": {"string_value": "[\"train\",        \"eval\"]"}}}, "artifact_type": {"name": "Examples", "properties": {"span":        "INT", "split_names": "STRING"}}}], "output_key": "examples", "producer_component_id":        "CsvExampleGen", "type": {"name": "Examples", "properties": {"span": "INT",        "split_names": "STRING"}}}}}, "_type": "tfx.components.example_gen.csv_example_gen.component.CsvExampleGen",        "driver_class": {"__class__": "Driver", "__module__": "tfx.components.example_gen.driver",        "__tfx_object_type__": "class"}, "executor_spec": {"__class__": "ExecutorClassSpec",        "__module__": "tfx.components.base.executor_spec", "__tfx_object_type__":        "jsonable", "executor_class": {"__class__": "Executor", "__module__": "tfx.components.example_gen.csv_example_gen.executor",        "__tfx_object_type__": "class"}}}'      - --component_config      - 'null'      - --enable_cache      command: [python, /tfx-src/tfx/orchestration/kubeflow/container_entrypoint.py]
      

       

       

      Attachments

        Activity

          People

            Unassigned Unassigned
            seedjeffwan Jiaxin Shan
            Votes:
            0 Vote for this issue
            Watchers:
            3 Start watching this issue

            Dates

              Created:
              Updated:
              Resolved: