apache/beam · error · ValueError

{format} format requires valid {format} schema to be passed

Error message

{format} format requires valid {format} schema to be passed to schema parameter.

What it means

In yaml_io._create_parser (used by read_from_pubsub), formats like JSON, AVRO, and PROTO need an explicit schema string to decode messages. The nested _validate_schema helper raises ValueError if schema is falsy (None or empty) for such a format, naming the format in the message.

Source

Thrown at sdks/python/apache_beam/yaml/yaml_io.py:240

          _ = write_result.failed_rows_with_errors | beam.Map(raise_exception)
          return {
              'post_write': write_result.failed_rows_with_errors
              | beam.FlatMap(lambda x: None)
          }

  return WriteToBigQueryHandlingErrors()


def _create_parser(
    format,
    schema: Any) -> tuple[schema_pb2.Schema, Callable[[bytes], beam.Row]]:

  format = format.upper()

  def _validate_schema():
    if not schema:
      raise ValueError(
          f'{format} format requires valid {format} schema to be passed to '
          f'schema parameter.')

  if format == 'RAW':
    if schema:
      raise ValueError('RAW format does not take a schema')
    return (
        schema_pb2.Schema(fields=[schemas.schema_field('payload', bytes)]),
        lambda payload: beam.Row(payload=payload))
  if format == 'STRING':
    if schema:
      raise ValueError('STRING format does not take a schema')
    return (
        schema_pb2.Schema(fields=[schemas.schema_field('payload', str)]),
        lambda payload: beam.Row(payload=payload.decode('utf-8')))
  elif format == 'JSON':
    _validate_schema()
    beam_schema = json_utils.json_schema_to_beam_schema(schema)

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass a valid schema: for JSON a JSON-schema string/dict, for AVRO an Avro schema, for PROTO a descriptor.
2. If messages are plain bytes, use format 'RAW' (no schema needed).
3. If messages are plain text, use format 'STRING'.

Example fix

// before
- type: ReadFromPubSub
  format: JSON
// after
- type: ReadFromPubSub
  format: JSON
  schema: '{"type": "object", "properties": {"id": {"type": "integer"}}}'

if fmt.upper() in {'JSON', 'AVRO', 'PROTO'} and not schema:
    raise ValueError(f"{fmt.upper()} format requires a schema")

def schema_required(fmt: str, schema) -> bool:
    return fmt.upper() in {'JSON', 'AVRO', 'PROTO'} and not schema

try:
    read_from_pubsub(format=fmt, schema=schema)
except ValueError as e:
    if 'schema to be passed' in str(e):
        schema = load_schema_from_config_store(topic)

When it happens

Trigger: Calling read_from_pubsub with format 'JSON' (or AVRO/PROTO) but schema=None or schema='' — _create_parser invokes _validate_schema which raises.

Common situations: Reading structured Pub/Sub messages without supplying the JSON schema; assuming the schema can be auto-detected from messages; copying a RAW-format config and only changing the format field.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• When using 'query' in ReadFromBigQuery YAML transform, 'sche
• RAW format does not take a schema
• STRING format does not take a schema
• Unknown format: {format}
• You have to supply one of 'by' and 'level'

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