We have a default VPC. Tried to run dataflow job. Initial step (Read file) manages to process 1/2 steps. Get JOB_MESSAGE_ERROR: SDK harness sdk-0-0 disconnected
error message, but nothing else in the logs. Have tried setup roles and vpc firewall rules.
I want to run a Dataflow job using the Geobeam image (Apache Beam Python 3.9 SDK 2.41.0). I have defined the job as follows:
def run(pipeline_args, known_args):
import apache_beam as beam
from apache_beam.io.gcp.internal.clients import storage
from apache_beam.options.pipeline_options import PipelineOptions
from geobeam.io import GeoJSONSource, filebasedsource
from geobeam.fn import format_record, make_valid, filter_invalid
pipeline_options = PipelineOptions([
] + pipeline_args)
with beam.Pipeline(options=pipeline_options) as p:
(p
| beam.io.Read(GeoJSONSource(known_args.gcs_url, encoding='utf-8'))
| 'FilterCords' >> beam.Filter(lambda x: len(x[-1]["coordinates"]) > 1)
| 'MakeValid' >> beam.Map(make_valid)
| 'FilterInvalid' >> beam.Filter(filter_invalid)
| 'FormatRecords' >> beam.Map(format_record)
| beam.io.WriteToText(known_args.gcs_write_url)
)
if __name__ == '__main__':
import logging
import argparse
logging.getLogger().setLevel(logging.INFO)
parser = argparse.ArgumentParser()
parser.add_argument('--gcs_url')
parser.add_argument('--gcs_write_url')
known_args, pipeline_args = parser.parse_known_args()
run(pipeline_args, known_args)
I run the job with the following command:
python -m main --runner DataflowRunner --project [[project_id]] \
--temp_location gs://[[temp_bucket_name]]/tmp \
--gcs_url gs://[[inputbucket_name]]/[[filename]].geojson \
--region europe-north1 --sdk_container_image gcr.io/dataflow-geobeam/example \
--gcs_write_url gs://gs://[[outputbucket_name]]/[[filename]]_processed.geojson \
--subnetwork [[full_link_to_subnet]]
We have setup a custom default VPC, and I added the recommended ranges for ingress/egress firewall rules for compute vm resources in GCP. I also gave the default service account used for the dataflow job the following roles:
I have also given my user roles on the service account:
It says the job was stopped, but that's because the job just wouldn't progress. I get the following log output
INFO:apache_beam.runners.dataflow.dataflow_runner:Job 2022-10-18_05_33_31-17288646308046950877 is in state JOB_STATE_PENDING
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:31.708Z: JOB_MESSAGE_BASIC: Dataflow Runner V2 auto-enabled. Use --experiments=disable_runner_v2 to opt out.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:32.780Z: JOB_MESSAGE_DETAILED: Autoscaling is enabled for job 2022-10-18_05_33_31-17288646308046950877. The number of workers will be between 1 and 1000.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:32.803Z: JOB_MESSAGE_DETAILED: Autoscaling was automatically enabled for job 2022-10-18_05_33_31-17288646308046950877.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:34.374Z: JOB_MESSAGE_BASIC: Worker configuration: n1-standard-1 in europe-north1-b.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.092Z: JOB_MESSAGE_DETAILED: Expanding SplittableParDo operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.109Z: JOB_MESSAGE_DETAILED: Expanding CollectionToSingleton operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.141Z: JOB_MESSAGE_DETAILED: Expanding CoGroupByKey operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.160Z: JOB_MESSAGE_DEBUG: Combiner lifting skipped for step WriteToText/Write/WriteImpl/GroupByKey: GroupByKey not followed by a combiner.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.184Z: JOB_MESSAGE_DETAILED: Expanding GroupByKey operations into optimizable parts.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.200Z: JOB_MESSAGE_DEBUG: Annotating graph with Autotuner information.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.226Z: JOB_MESSAGE_DETAILED: Fusing adjacent ParDo, Read, Write, and Flatten operations
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.243Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/InitializeWrite into WriteToText/Write/WriteImpl/DoOnce/Map(decode)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.262Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/DoOnce/FlatMap(<lambda at core.py:3481>) into WriteToText/Write/WriteImpl/DoOnce/Impulse
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.278Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/DoOnce/Map(decode) into WriteToText/Write/WriteImpl/DoOnce/FlatMap(<lambda at core.py:3481>)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.294Z: JOB_MESSAGE_DETAILED: Fusing consumer Read/Map(<lambda at iobase.py:908>) into Read/Impulse
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.310Z: JOB_MESSAGE_DETAILED: Fusing consumer ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction into Read/Map(<lambda at iobase.py:908>)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.325Z: JOB_MESSAGE_DETAILED: Fusing consumer ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/SplitWithSizing into ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.340Z: JOB_MESSAGE_DETAILED: Fusing consumer FilterCords into ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/ProcessElementAndRestrictionWithSizing
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.356Z: JOB_MESSAGE_DETAILED: Fusing consumer MakeValid into FilterCords
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.372Z: JOB_MESSAGE_DETAILED: Fusing consumer FilterInvalid into MakeValid
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.387Z: JOB_MESSAGE_DETAILED: Fusing consumer FormatRecords into FilterInvalid
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.402Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/WindowInto(WindowIntoFn) into FormatRecords
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.417Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/WriteBundles into WriteToText/Write/WriteImpl/WindowInto(WindowIntoFn)
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.432Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/Pair into WriteToText/Write/WriteImpl/WriteBundles
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.447Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/GroupByKey/Write into WriteToText/Write/WriteImpl/Pair
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.464Z: JOB_MESSAGE_DETAILED: Fusing consumer WriteToText/Write/WriteImpl/Extract into WriteToText/Write/WriteImpl/GroupByKey/Read
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.489Z: JOB_MESSAGE_DEBUG: Workflow config is missing a default resource spec.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.504Z: JOB_MESSAGE_DEBUG: Adding StepResource setup and teardown to workflow graph.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.519Z: JOB_MESSAGE_DEBUG: Adding workflow start and stop steps.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.535Z: JOB_MESSAGE_DEBUG: Assigning stage ids.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.624Z: JOB_MESSAGE_DEBUG: Executing wait step start19
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.655Z: JOB_MESSAGE_BASIC: Executing operation Read/Impulse+Read/Map(<lambda at iobase.py:908>)+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/SplitWithSizing
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.668Z: JOB_MESSAGE_BASIC: Executing operation WriteToText/Write/WriteImpl/DoOnce/Impulse+WriteToText/Write/WriteImpl/DoOnce/FlatMap(<lambda at core.py:3481>)+WriteToText/Write/WriteImpl/DoOnce/Map(decode)+WriteToText/Write/WriteImpl/InitializeWrite
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.682Z: JOB_MESSAGE_DEBUG: Starting worker pool setup.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:33:35.696Z: JOB_MESSAGE_BASIC: Starting 1 workers in europe-north1-b...
INFO:apache_beam.runners.dataflow.dataflow_runner:Job 2022-10-18_05_33_31-17288646308046950877 is in state JOB_STATE_RUNNING
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:34:21.585Z: JOB_MESSAGE_DETAILED: Autoscaling: Raised the number of workers to 1 based on the rate of progress in the currently running stage(s).
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:37:30.456Z: JOB_MESSAGE_DETAILED: Workers have started successfully.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:42:40.315Z: JOB_MESSAGE_BASIC: Finished operation Read/Impulse+Read/Map(<lambda at iobase.py:908>)+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/PairWithRestriction+ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6/SplitWithSizing
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:42:40.354Z: JOB_MESSAGE_DEBUG: Value "ref_AppliedPTransform_Read-SDFBoundedSourceReader-ParDo-SDFBoundedSourceDoFn-_6-split-with-sizing-out3" materialized.
INFO:apache_beam.runners.dataflow.dataflow_runner:2022-10-18T12:42:42.422Z: JOB_MESSAGE_ERROR: SDK harness sdk-0-0 disconnected.
And then it tries to raise the numbers of workers again to 1, then it instantly gets JOB_MESSAGE_ERROR: SDK harness sdk-0-0 disconnected.
over and over again. As a side note - it also takes about 10 minutes before the pipeline actually starts.
I managed to get it working with the DirectRunner
option. I don't know where to look? Could it be related to the VPC?
I tried to run the word count example both on the native/default image and geobeam image and it works on the native/default but not for the geobeam image.
Why could that be?
After some trial and error, I found that the python version of the geobeam base image has to match the local python version on your machine, or else it won't work. At time of answering, this is python 3.8.