{"record":{"id":"004eef1e3e414cb2","repo":"apache/beam","slug":"format-format-requires-valid-format-schema-to-be-passed-to","errorCode":null,"errorMessage":"{format} format requires valid {format} schema to be passed to schema parameter.","messagePattern":"(.+?) format requires valid (.+?) schema to be passed to schema parameter\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/yaml/yaml_io.py","lineNumber":240,"sourceCode":"\n          _ = write_result.failed_rows_with_errors | beam.Map(raise_exception)\n          return {\n              'post_write': write_result.failed_rows_with_errors\n              | beam.FlatMap(lambda x: None)\n          }\n\n  return WriteToBigQueryHandlingErrors()\n\n\ndef _create_parser(\n    format,\n    schema: Any) -> tuple[schema_pb2.Schema, Callable[[bytes], beam.Row]]:\n\n  format = format.upper()\n\n  def _validate_schema():\n    if not schema:\n      raise ValueError(\n          f'{format} format requires valid {format} schema to be passed to '\n          f'schema parameter.')\n\n  if format == 'RAW':\n    if schema:\n      raise ValueError('RAW format does not take a schema')\n    return (\n        schema_pb2.Schema(fields=[schemas.schema_field('payload', bytes)]),\n        lambda payload: beam.Row(payload=payload))\n  if format == 'STRING':\n    if schema:\n      raise ValueError('STRING format does not take a schema')\n    return (\n        schema_pb2.Schema(fields=[schemas.schema_field('payload', str)]),\n        lambda payload: beam.Row(payload=payload.decode('utf-8')))\n  elif format == 'JSON':\n    _validate_schema()\n    beam_schema = json_utils.json_schema_to_beam_schema(schema)","sourceCodeStart":222,"sourceCodeEnd":258,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/yaml/yaml_io.py#L222-L258","documentation":"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.","triggerScenarios":"Calling read_from_pubsub with format 'JSON' (or AVRO/PROTO) but schema=None or schema='' — _create_parser invokes _validate_schema which raises.","commonSituations":"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.","solutions":["Pass a valid schema: for JSON a JSON-schema string/dict, for AVRO an Avro schema, for PROTO a descriptor.","If messages are plain bytes, use format 'RAW' (no schema needed).","If messages are plain text, use format 'STRING'."],"exampleFix":"// before\n- type: ReadFromPubSub\n  format: JSON\n// after\n- type: ReadFromPubSub\n  format: JSON\n  schema: '{\"type\": \"object\", \"properties\": {\"id\": {\"type\": \"integer\"}}}'","handlingStrategy":"validation","validationCode":"if fmt.upper() in {'JSON', 'AVRO', 'PROTO'} and not schema:\n    raise ValueError(f\"{fmt.upper()} format requires a schema\")","typeGuard":"def schema_required(fmt: str, schema) -> bool:\n    return fmt.upper() in {'JSON', 'AVRO', 'PROTO'} and not schema","tryCatchPattern":"try:\n    read_from_pubsub(format=fmt, schema=schema)\nexcept ValueError as e:\n    if 'schema to be passed' in str(e):\n        schema = load_schema_from_config_store(topic)","preventionTips":["Store schemas alongside topics in config/registry (e.g. Schema Registry export).","Reject empty-string schemas in config validation.","Document per-format schema requirements in the pipeline template."],"tags":["python","apache-beam","yaml","pubsub","missing-argument"],"backgroundTag":"missing-required-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}