Why does my DoFn restart the processing of an element indefinitely?

user_19815647 0
Viewed 51

Problem:

For long running (2-3 Hours) elements, a web scraping DoFn restarts execution indefinitely right when returning the new pCollection, without proceeding to the next step in the pipeline.

Error:

Main Error:

"Commit failed, will be retried at higher level but may not succeed"
computation = 'P51', sharding_key = '12c92e1638acbb09',
status = generic::internal: Windmill failed to commit the work item.
CommitStatus: VALIDATION_FAILED"

Full Error:

[
  {
    "insertId": "7054061089043139090:270578:0:287794",
    "jsonPayload": {
      "job": "2022-08-19_13_38_49-15282523611410283510",
      "work": "2",
      "step": "Scrape Products",
      "instruction": "process_bundle-2-1",
      "thread": "Thread-14",
      "logger": "REDACTED PATH/product_scraper.py:69",
      "worker": "df-hvm-beamapp-vincentye-0819203-08191338-dtgt-harness-l4gg",
      "portability_worker_id": "sdk-0-0",
      "message": "returning new_products_dataframe and REDACTED_dataframe"
    },
    "resource": {
      "type": "dataflow_step",
      "labels": {
        "job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
        "step_id": "Scrape Products",
        "project_id": "REDACTED",
        "job_id": "2022-08-19_13_38_49-15282523611410283510",
        "region": "us-west2"
      }
    },
    "timestamp": "2022-08-19T21:38:39.037309408Z",
    "severity": "WARNING",
    "labels": {
      "dataflow.googleapis.com/region": "us-west2",
      "dataflow.googleapis.com/job_id": "2022-08-19_13_38_49-15282523611410283510",
      "compute.googleapis.com/resource_name": "df-hvm-beamapp-vincentye-0819203-08191338-dtgt-harness-l4gg",
      "dataflow.googleapis.com/job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
      "dataflow.googleapis.com/service_option": "prime",
      "compute.googleapis.com/resource_id": "7054061089043139090",
      "compute.googleapis.com/resource_type": "instance"
    },
    "logName": "projects/REDACTED/logs/dataflow.googleapis.com%2Fworker",
    "receiveTimestamp": "2022-08-19T21:38:52.093172441Z"
  },
  {
    "insertId": "s=6f21f5e3f31c4dfca9b6c2dc7f2de30a;i=664;b=f6f95ab5a8234fdeac1c3d31e7f63933;m=d31e6263;t=5e69eebe139ed;x=77b6120c808dd2fb",
    "jsonPayload": {
      "line": "sampler.go:311",
      "message": "Successfully sampled resources"
    },
    "resource": {
      "type": "dataflow_step",
      "labels": {
        "job_id": "2022-08-19_13_38_49-15282523611410283510",
        "project_id": "REDACTED",
        "region": "us-west2",
        "step_id": "",
        "job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u"
      }
    },
    "timestamp": "2022-08-19T21:38:40.180661Z",
    "severity": "INFO",
    "labels": {
      "dataflow.googleapis.com/region": "us-west2",
      "dataflow.googleapis.com/log_type": "system",
      "dataflow.googleapis.com/job_id": "2022-08-19_13_38_49-15282523611410283510",
      "compute.googleapis.com/resource_id": "4422558927708899858",
      "dataflow.googleapis.com/job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
      "compute.googleapis.com/resource_name": "df-hvm-beamapp-vincentye-0819203-08191338-dtgt-harness-3jp2",
      "compute.googleapis.com/resource_type": "instance",
      "dataflow.googleapis.com/service_option": "prime"
    },
    "logName": "projects/REDACTED/logs/dataflow.googleapis.com%2Fresource",
    "receiveTimestamp": "2022-08-19T21:38:41.688521247Z"
  },
  {
    "insertId": "7054061089043139090:270576:0:17358",
    "jsonPayload": {
      "message": "Commit failed, will be retried at higher level but may not succeed. computation = \"P51\", sharding_key = \"62fa5cb5ab454560\", status = generic::internal: Windmill failed to commit the work item. CommitStatus: VALIDATION_FAILED",
      "thread": "119",
      "line": "streaming_worker_client.cc:514"
    },
    "resource": {
      "type": "dataflow_step",
      "labels": {
        "region": "us-west2",
        "project_id": "REDACTED",
        "job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
        "job_id": "2022-08-19_13_38_49-15282523611410283510",
        "step_id": ""
      }
    },
    "timestamp": "2022-08-19T21:38:40.686591Z",
    "severity": "ERROR",
    "labels": {
      "dataflow.googleapis.com/service_option": "prime",
      "dataflow.googleapis.com/job_id": "2022-08-19_13_38_49-15282523611410283510",
      "dataflow.googleapis.com/log_type": "system",
      "dataflow.googleapis.com/job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
      "dataflow.googleapis.com/region": "us-west2",
      "compute.googleapis.com/resource_name": "df-hvm-beamapp-vincentye-0819203-08191338-dtgt-harness-l4gg",
      "compute.googleapis.com/resource_type": "instance",
      "compute.googleapis.com/resource_id": "7054061089043139090"
    },
    "logName": "projects/REDACTED/logs/dataflow.googleapis.com%2Fharness",
    "receiveTimestamp": "2022-08-19T21:38:42.095203701Z"
  },
  {
    "insertId": "7054061089043139090:270575:0:13058",
    "jsonPayload": {
      "line": "exec.go:66",
      "message": "E0819 21:38:40.686591     119 streaming_worker_client.cc:514] Commit failed, will be retried at higher level but may not succeed. computation = \"P51\", sharding_key = \"62fa5cb5ab454560\", status = generic::internal: Windmill failed to commit the work item. CommitStatus: VALIDATION_FAILED"
    },
    "resource": {
      "type": "dataflow_step",
      "labels": {
        "job_id": "2022-08-19_13_38_49-15282523611410283510",
        "step_id": "",
        "job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
        "project_id": "REDACTED",
        "region": "us-west2"
      }
    },
    "timestamp": "2022-08-19T21:38:40.686846Z",
    "severity": "INFO",
    "labels": {
      "dataflow.googleapis.com/log_type": "system",
      "dataflow.googleapis.com/service_option": "prime",
      "compute.googleapis.com/resource_type": "instance",
      "dataflow.googleapis.com/region": "us-west2",
      "compute.googleapis.com/resource_name": "df-hvm-beamapp-vincentye-0819203-08191338-dtgt-harness-l4gg",
      "dataflow.googleapis.com/job_id": "2022-08-19_13_38_49-15282523611410283510",
      "compute.googleapis.com/resource_id": "7054061089043139090",
      "dataflow.googleapis.com/job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u"
    },
    "logName": "projects/REDACTED/logs/dataflow.googleapis.com%2Fharness-startup",
    "receiveTimestamp": "2022-08-19T21:38:52.093608994Z"
  },
  {
    "insertId": "7054061089043139090:270576:0:17641",
    "jsonPayload": {
      "line": "streaming_worker_client.cc:566",
      "thread": "119",
      "message": "Error while processing a work item: INTERNAL: Windmill failed to commit the work item. CommitStatus: VALIDATION_FAILED\n=== Source Location Trace: ===\ndist_proc/dax/workflow/worker/streaming/streaming_worker_client.cc:492"
    },
    "resource": {
      "type": "dataflow_step",
      "labels": {
        "step_id": "",
        "job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
        "job_id": "2022-08-19_13_38_49-15282523611410283510",
        "project_id": "REDACTED",
        "region": "us-west2"
      }
    },
    "timestamp": "2022-08-19T21:38:40.687126Z",
    "severity": "WARNING",
    "labels": {
      "compute.googleapis.com/resource_type": "instance",
      "dataflow.googleapis.com/region": "us-west2",
      "compute.googleapis.com/resource_name": "df-hvm-beamapp-vincentye-0819203-08191338-dtgt-harness-l4gg",
      "dataflow.googleapis.com/job_name": "beamapp-vincentye-0819203841-963772-8quxtp6u",
      "dataflow.googleapis.com/job_id": "2022-08-19_13_38_49-15282523611410283510",
      "compute.googleapis.com/resource_id": "7054061089043139090",
      "dataflow.googleapis.com/service_option": "prime",
      "dataflow.googleapis.com/log_type": "system"
    },
    "logName": "projects/REDACTED/logs/dataflow.googleapis.com%2Fharness",
    "receiveTimestamp": "2022-08-19T21:39:02.095105228Z"
  }
]

More Info:

  • The error only occurs with large elements where the DoFn is forced to run for more than 2 hours.

    • The resultant pCollection for long running elements is usually around 100 to 150 MB. (It’s a Pandas dataframe with many images in byte form)

  • The failing element is retried forever.

  • Issue does not occur when running pipeline on LocalRunner

Already Tried:

  • Turning off Dataflow Prime, setting machine_type to n2-highmem-2
  • Setting number_of_worker_harness_threads to 3
apache-beam google-cloud-dataflow python
2 Answers

I spoke with Google Support recently and found that the VALIDATION_FAILED Windmill error actually had to do with a 80MB max single element size in streaming engine.

Solution:

Make sure any single element in a materialized pCollection is less than 80MB.

Some Thoughts:

If someone from google reads this, maybe make the error message more descriptive instead of a vague VALIDATION_FAILED windmill error-- If I didn't have Google Support I don't think I would have found the solution nearly this fast.

I suggest Windmill: VALIDATION_FAILED, QUOTA_ERROR: Element over 80MB

References:

https://cloud.google.com/dataflow/quotas

user_198156470

My guess would be that some sort of an internal timeout causes Dataflow streaming runner to not be able to commit elements that are being processed for a very long time. Is it possible to update the pipeline so that DoFn outputs elements more frequently (or returns from the ProcessElement call by splitting work into multiple elements) ?

If not, I suggest contact Google Cloud support so that they can look at your failed job.

user_46960150
Related