apache/beam · error · RuntimeError

"Please run python -m apache_beam.yaml.generate_yaml_docs…

Error message

"Please run python -m apache_beam.yaml.generate_yaml_docs --schema_file='{transform_schemas_path}' to run with transform-specific validation."

What it means

Raised by the cached pipeline_schema helper when strictness is 'per_transform' but the generated transforms.schema.yaml file does not exist next to pipeline.schema.yaml. That file is produced by a code-generation step, not shipped; the error tells the user to run the generate_yaml_docs module with --schema_file pointing at the located path. The input at fault is the environment/checkout state, not user pipeline code.

Solutions

1. Run: python -m apache_beam.yaml.generate_yaml_docs --schema_file=<the path printed in the error> to generate the missing schema
2. Use a different strictness level (e.g. 'basic') that does not require the per-transform schema

When it happens

Trigger: Thrown at sdks/python/apache_beam/yaml/yaml_transform.py:67 when the library encounters an invalid state.

Common situations: See trigger scenarios.

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

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_transform.py:67

_LOGGER = logging.getLogger(__name__)
yaml_provider.fix_pycallable()

try:
  import jsonschema
except ImportError:
  jsonschema = None


@functools.lru_cache
def pipeline_schema(strictness):
  with open(yaml_utils.locate_data_file('pipeline.schema.yaml')) as yaml_file:
    pipeline_schema = yaml.safe_load(yaml_file)
  if strictness == 'per_transform':
    transform_schemas_path = yaml_utils.locate_data_file(
        'transforms.schema.yaml')
    if not os.path.exists(transform_schemas_path):
      raise RuntimeError(
          "Please run "
          "python -m apache_beam.yaml.generate_yaml_docs "
          f"--schema_file='{transform_schemas_path}' "
          "to run with transform-specific validation.")
    with open(transform_schemas_path) as fin:
      pipeline_schema['$defs']['transform']['allOf'].extend(yaml.safe_load(fin))
  return pipeline_schema


def _closest_line(o, path):
  best_line = SafeLineLoader.get_line(o)
  for step in path:
    o = o[step]
    maybe_line = SafeLineLoader.get_line(o)
    if maybe_line != 'unknown':
      best_line = maybe_line
  return best_line

View on GitHub (pinned to 12126d8942)