Details
-
Bug
-
Status: Resolved
-
P3
-
Resolution: Won't Fix
-
None
-
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]