apache/beam · error · RuntimeError

Failed to create Dataflow client. Pipeline options are…

Error message

Failed to create Dataflow client. Pipeline options are required to save the attributes.in the artifact location %s

What it means

When saving attributes to a GCS path, the manager may use a Dataflow-based GCS upload helper that needs pipeline options to construct its client. If an exception occurs in that helper and no pipeline options were supplied, the original error is re-raised as a RuntimeError explaining that pipeline options are required to save attributes to the artifact location.

Solutions

Pass PipelineOptions to MLTransform (e.g. PipelineOptions(['--project=my-project'])) when artifact_location is a GCS path.
Run with proper Google Cloud credentials (GOOGLE_APPLICATION_CREDENTIALS or ADC) so the client can be constructed.
Save artifacts to a local path instead if you do not need GCS.
Fix the underlying chained exception (inspect the 'from exc' cause) such as network or permission errors.

Example fix

// before
MLTransform(artifact_location='gs://bucket/artifacts')
// after
from apache_beam.options.pipeline_options import PipelineOptions
MLTransform(artifact_location='gs://bucket/artifacts',
            options=PipelineOptions(['--project=my-project']))

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.options.pipeline_options import PipelineOptions
if artifact_location.startswith('gs://') and options is None:
    options = PipelineOptions(['--project=my-project'])

Try / catch

try:
    save_attributes(gcs_path)
except RuntimeError as e:
    if 'Pipeline options are required' in str(e):
        save_attributes(gcs_path, options=PipelineOptions(['--project=my-project']))
    else:
        raise

Prevention

Always pass PipelineOptions when artifact_location is a GCS path.
Ensure GOOGLE_APPLICATION_CREDENTIALS / ADC is configured in all environments.
Provide --project (and temp_location) in worker environment options.
Inspect the chained exception (from exc) to catch credential issues early.

When it happens

Trigger: Saving MLTransform attributes to a gs:// artifact_location without passing PipelineOptions (e.g. options=None), so the GCS/Dataflow client cannot be created (no project/credentials context).

Common situations: Running MLTransform in a standalone script or unit test writing to GCS without providing --project/--temp_location options; missing default credentials so the Dataflow/GCS client construction fails.

Understand the failure class

Background: "X is required", "must be set", "cannot be empty": the missing-required-config error family, from Vertex AI project/location to WeChat keys — this error's family across 18 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/ac5f36b00b513c15. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/ml/transforms/base.py:577

    if _JsonPickleTransformAttributeManager._is_remote_path(artifact_location):
      temp_dir = tempfile.mkdtemp()
      temp_json_file = os.path.join(temp_dir, _ATTRIBUTE_FILE_NAME)
      with open(temp_json_file, 'w+') as f:
        f.write(jsonpickle.encode(ptransform_list))
      with open(temp_json_file, 'rb') as f:
        from apache_beam.runners.dataflow.internal import apiclient
        _LOGGER.info('Creating artifact location: %s', artifact_location)
        # pipeline options required to for the client to configure project.
        options = kwargs.get('options')
        try:
          apiclient.DataflowApplicationClient(options=options).stage_file(
              gcs_or_local_path=artifact_location,
              file_name=_ATTRIBUTE_FILE_NAME,
              stream=f,
              mime_type='application/json')
        except Exception as exc:
          if not options:
            raise RuntimeError(
                "Failed to create Dataflow client. "
                "Pipeline options are required to save the attributes."
                "in the artifact location %s" % artifact_location) from exc
          raise
    else:
      if not FileSystems.exists(artifact_location):
        FileSystems.mkdirs(artifact_location)
      # FileSystems.open() fails if the file does not exist.
      with open(os.path.join(artifact_location, _ATTRIBUTE_FILE_NAME),
                'w+') as f:
        f.write(jsonpickle.encode(ptransform_list))

  @staticmethod
  def load_attributes(artifact_location):
    with FileSystems.open(os.path.join(artifact_location, _ATTRIBUTE_FILE_NAME),
                          'rb') as f:
      return jsonpickle.decode(f.read())

View on GitHub (pinned to 12126d8942)