apache/beam · error · FileNotFoundError

Artifacts not found at location: %s when using read_artifact

Error message

Artifacts not found at location: %s when using read_artifact_location. Make sure you've run the pipeline with write_artifact_location using this artifact location before running with read_artifact_location set.

What it means

In read mode (read_artifact_location set), MLTransform's expand() must load the raw-data metadata schema written by a previous write run. If SCHEMA_FILE does not exist under RAW_DATA_METADATA_DIR inside the artifact location, a FileNotFoundError is raised because there is nothing to read from.

Source

Thrown at sdks/python/apache_beam/ml/transforms/handlers.py:392

                dict[str, Union[tuple(column_type_mapping.values())]]))  # type: ignore
        # AnalyzeAndTransformDataset raise type hint since this is
        # schema'd PCollection and the current output type would be a
        # custom type(NamedTuple) or a beam.Row type.
      else:
        column_type_mapping = self._map_column_names_to_types_from_transforms()
        # Add id so TFT can output id as output but as a no-op.
      raw_data_metadata = self.get_raw_data_metadata(
          input_types=column_type_mapping)
      # Write untransformed metadata to a file so that it can be re-used
      # during Transform step.
      metadata_io.write_metadata(
          metadata=raw_data_metadata,
          path=os.path.join(self.artifact_location, RAW_DATA_METADATA_DIR))
    else:
      # Read the metadata from the artifact_location.
      if not FileSystems.exists(os.path.join(
          self.artifact_location, RAW_DATA_METADATA_DIR, SCHEMA_FILE)):
        raise FileNotFoundError(
            "Artifacts not found at location: %s when using "
            "read_artifact_location. Make sure you've run the pipeline with "
            "write_artifact_location using this artifact location before "
            "running with read_artifact_location set." %
            os.path.join(self.artifact_location, RAW_DATA_METADATA_DIR))
      raw_data_metadata = metadata_io.read_metadata(
          os.path.join(self.artifact_location, RAW_DATA_METADATA_DIR))

      element_type = raw_data.element_type
      if (isinstance(element_type, RowTypeConstraint) or
          native_type_compatibility.match_is_named_tuple(element_type)):
        # convert Row or NamedTuple to Dict
        column_type_mapping = self._map_column_names_to_types(
            row_type=element_type)
        raw_data = (
            raw_data
            | _ConvertNamedTupleToDict().with_output_types(
                dict[str, Union[tuple(column_type_mapping.values())]]))  # type: ignore

View on GitHub (pinned to 12126d8942)

Solutions

Run the pipeline first with with_write_artifact_location pointing at the same path so the metadata schema is produced.
Verify the exact path exists and contains <artifact_location>/RAW_DATA_METADATA_DIR/SCHEMA_FILE (use FileSystems.exists or gsutil ls).
Check for typos or permission issues in the artifact_location path.
If a previous write run failed, re-run the write pipeline to completion before reading.

Example fix

# before
result = pcoll | MLTransform(...).with_read_artifact_location('gs://bucket/artifacts')

# after
# step 1: write
train = pcoll | MLTransform(...).with_write_artifact_location('gs://bucket/artifacts')
# step 2: read
result = eval_pcoll | MLTransform(...).with_read_artifact_location('gs://bucket/artifacts')

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.io.filesystems import FileSystems
import os
from apache_beam.ml.transforms.handlers import RAW_DATA_METADATA_DIR, SCHEMA_FILE

def artifacts_exist(artifact_location: str) -> bool:
    return FileSystems.exists(os.path.join(artifact_location, RAW_DATA_METADATA_DIR, SCHEMA_FILE))

assert artifacts_exist(loc), 'Run the write pipeline first'

Type guard

def artifacts_exist(artifact_location: str) -> bool:
    import os
    from apache_beam.io.filesystems import FileSystems
    from apache_beam.ml.transforms.handlers import RAW_DATA_METADATA_DIR, SCHEMA_FILE
    return FileSystems.exists(os.path.join(artifact_location, RAW_DATA_METADATA_DIR, SCHEMA_FILE))

Try / catch

try:
    result = eval_pcoll | MLTransform(...).with_read_artifact_location(loc)
except FileNotFoundError as e:
    # fall back to a fresh write pass to bootstrap artifacts
    train_pcoll | MLTransform(...).with_write_artifact_location(loc)
    result = eval_pcoll | MLTransform(...).with_read_artifact_location(loc)

Prevention

Always run the write-artifact pipeline before any read-artifact pipeline.
Centralize artifact_location in config to avoid path typos.
Check FileSystems.exists on the schema file before submitting read pipelines.
Do not clean the artifact directory between write and read runs.

When it happens

Trigger: Calling MLTransform(...).with_read_artifact_location(path) on a path where no prior pipeline run used with_write_artifact_location, or the path is wrong/empty, or the run that wrote artifacts failed partway.

Common situations: Pointing read at a GCS/local path with a typo, running read before ever running write, cleaning the artifact directory between runs, or a failed first pipeline run that never materialized the schema file.

Understand the failure class

Background: "File not found" and ENOENT errors: why libraries can't find a file that should exist — this error's family across 50 libraries.

Related errors

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