René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Add custom retry and timeout options for partition reads and queries","articleBody":"I am building an application in Apache Beam and Python that runs in Google DataFlow. I am using the `ReadFromSpanner` method in `apache_beam.io.gcp.experimental.spannerio`. This works for most of my Spanner tables but the really large ones that are \u003e40m rows tend to fail due to the following error:\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.8/site-packages/dataflow_worker/batchworker.py\", line 649, in do_work\r\n    work_executor.execute()\r\n  File \"/usr/local/lib/python3.8/site-packages/dataflow_worker/executor.py\", line 179, in execute\r\n    op.start()\r\n  File \"dataflow_worker/shuffle_operations.py\", line 63, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start\r\n  File \"dataflow_worker/shuffle_operations.py\", line 64, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start\r\n  File \"dataflow_worker/shuffle_operations.py\", line 79, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start\r\n  File \"dataflow_worker/shuffle_operations.py\", line 80, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start\r\n  File \"dataflow_worker/shuffle_operations.py\", line 84, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start\r\n  File \"apache_beam/runners/worker/operations.py\", line 359, in apache_beam.runners.worker.operations.Operation.output\r\n  File \"apache_beam/runners/worker/operations.py\", line 221, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive\r\n  File \"dataflow_worker/shuffle_operations.py\", line 261, in dataflow_worker.shuffle_operations.BatchGroupAlsoByWindowsOperation.process\r\n  File \"dataflow_worker/shuffle_operations.py\", line 268, in dataflow_worker.shuffle_operations.BatchGroupAlsoByWindowsOperation.process\r\n  File \"apache_beam/runners/worker/operations.py\", line 359, in apache_beam.runners.worker.operations.Operation.output\r\n  File \"apache_beam/runners/worker/operations.py\", line 221, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive\r\n  File \"apache_beam/runners/worker/operations.py\", line 718, in apache_beam.runners.worker.operations.DoOperation.process\r\n  File \"apache_beam/runners/worker/operations.py\", line 719, in apache_beam.runners.worker.operations.DoOperation.process\r\n  File \"apache_beam/runners/common.py\", line 1241, in apache_beam.runners.common.DoFnRunner.process\r\n  File \"apache_beam/runners/common.py\", line 1306, in apache_beam.runners.common.DoFnRunner._reraise_augmented\r\n  File \"apache_beam/runners/common.py\", line 1239, in apache_beam.runners.common.DoFnRunner.process\r\n  File \"apache_beam/runners/common.py\", line 587, in apache_beam.runners.common.SimpleInvoker.invoke_process\r\n  File \"apache_beam/runners/common.py\", line 1401, in apache_beam.runners.common._OutputProcessor.process_outputs\r\n  File \"apache_beam/runners/worker/operations.py\", line 221, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive\r\n  File \"apache_beam/runners/worker/operations.py\", line 718, in apache_beam.runners.worker.operations.DoOperation.process\r\n  File \"apache_beam/runners/worker/operations.py\", line 719, in apache_beam.runners.worker.operations.DoOperation.process\r\n  File \"apache_beam/runners/common.py\", line 1241, in apache_beam.runners.common.DoFnRunner.process\r\n  File \"apache_beam/runners/common.py\", line 1306, in apache_beam.runners.common.DoFnRunner._reraise_augmented\r\n  File \"apache_beam/runners/common.py\", line 1239, in apache_beam.runners.common.DoFnRunner.process\r\n  File \"apache_beam/runners/common.py\", line 587, in apache_beam.runners.common.SimpleInvoker.invoke_process\r\n  File \"apache_beam/runners/common.py\", line 1401, in apache_beam.runners.common._OutputProcessor.process_outputs\r\n  File \"apache_beam/runners/worker/operations.py\", line 221, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive\r\n  File \"apache_beam/runners/worker/operations.py\", line 718, in apache_beam.runners.worker.operations.DoOperation.process\r\n  File \"apache_beam/runners/worker/operations.py\", line 719, in apache_beam.runners.worker.operations.DoOperation.process\r\n  File \"apache_beam/runners/common.py\", line 1241, in apache_beam.runners.common.DoFnRunner.process\r\n  File \"apache_beam/runners/common.py\", line 1321, in apache_beam.runners.common.DoFnRunner._reraise_augmented\r\n  File \"/usr/local/lib/python3.8/site-packages/future/utils/__init__.py\", line 446, in raise_with_traceback\r\n    raise exc.with_traceback(traceback)\r\n  File \"apache_beam/runners/common.py\", line 1239, in apache_beam.runners.common.DoFnRunner.process\r\n  File \"apache_beam/runners/common.py\", line 587, in apache_beam.runners.common.SimpleInvoker.invoke_process\r\n  File \"apache_beam/runners/common.py\", line 1374, in apache_beam.runners.common._OutputProcessor.process_outputs\r\n  File \"/usr/local/lib/python3.8/site-packages/apache_beam/io/gcp/experimental/spannerio.py\", line 550, in process\r\n    for row in read_action(element['partitions']):\r\n  File \"/usr/local/lib/python3.8/site-packages/google/cloud/spanner_v1/streamed.py\", line 143, in __iter__\r\n    self._consume_next()\r\n  File \"/usr/local/lib/python3.8/site-packages/google/cloud/spanner_v1/streamed.py\", line 116, in _consume_next\r\n    response = six.next(self._response_iterator)\r\n  File \"/usr/local/lib/python3.8/site-packages/google/cloud/spanner_v1/snapshot.py\", line 45, in _restart_on_unavailable\r\n    for item in iterator:\r\n  File \"/usr/local/lib/python3.8/site-packages/google/api_core/grpc_helpers.py\", line 116, in next\r\n    six.raise_from(exceptions.from_grpc_error(exc), exc)\r\n  File \"\u003cstring\u003e\", line 3, in raise_from\r\ngoogle.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded [while running 'Read from Spanner/Read From Partitions']\r\n```\r\nFrom my understanding this error comes from the `ReadFromSpanner` operation as it's workers have timed out.\r\n\r\nTo solve this I have tried the following:\r\n\r\n- Changed the num_workers and disk_size_gb and added the `--experiments=shuffle_mode=service` flag as suggested in [Google's Common Error Guidance](https://cloud.google.com/dataflow/docs/guides/common-errors#tsg-rpc-timeout)\r\n- Changed the Machine Type from `n1-standard-1` up to `n1-standard-32`\r\n- My latest code is attached below. I am including Transformation for simple data wrangling in the rows.\r\n\r\n```\r\n \"\"\"Set pipeline arguments.\"\"\"\r\n    options = PipelineOptions(\r\n        region=RUNNER_REGION,\r\n        project=RUNNER_PROJECT_ID,\r\n        runner=RUNNER,\r\n        temp_location=TEMP_LOCATION,\r\n        job_name=JOB_NAME,\r\n        service_account_email=SA_EMAIL,\r\n        setup_file=SETUP_FILE_PATH,\r\n        disk_size_gb=500,\r\n        num_workers=10,\r\n        machine_type=\"n1-standard-2\",\r\n        save_main_session=True)\r\n\r\n    \"\"\"Build and run the pipeline.\"\"\"\r\n        with beam.Pipeline(options=options) as p:\r\n            (p\r\n             | \"Read from Spanner\" \u003e\u003e ReadFromSpanner(SPANNER_PROJECT_ID, SPANNER_INSTANCE_ID, SPANNER_DB, sql=QUERY)\r\n             | \"Transform elements into dictionary\" \u003e\u003e beam.ParDo(Transformation)\r\n             | \"Write new records to BQ\" \u003e\u003e WriteToBigQuery(\r\n                 BIGQUERY_TABLE,\r\n                 schema=SCHEMA,\r\n                 write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,\r\n                 create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)\r\n                 ) \r\n```\r\nA potential solution is to edit the timeout control; I have seen this being available in Java but not in Python. How can I edit timeout control in Python or is there any other solution to this issue?\r\n","author":{"url":"https://github.com/sebastian-montero","@type":"Person","name":"sebastian-montero"},"datePublished":"2021-03-10T17:44:50.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":3},"url":"https://github.com/261/python-spanner/issues/261"}