{"record":{"id":"a00a0a0d04a73e04","repo":"apache/beam","slug":"an-explicit-schema-is-required-to-write-non-schema-d","errorCode":null,"errorMessage":"An explicit schema is required to write non-schema'd PCollections.","messagePattern":"An explicit schema is required to write non-schema'd PCollections\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/avroio.py","lineNumber":435,"sourceCode":"        If set it overrides user windowing. Mandatory for GlobalWindow.\n\n    Returns:\n      A WriteToAvro transform usable for writing.\n    \"\"\"\n    self._schema = schema\n    self._sink_provider = lambda avro_schema: _create_avro_sink(\n        file_path_prefix, avro_schema, codec, file_name_suffix, num_shards,\n        shard_name_template, mime_type, triggering_frequency)\n\n  def expand(self, pcoll):\n    if self._schema:\n      avro_schema = self._schema\n      records = pcoll\n    else:\n      try:\n        beam_schema = schemas.schema_from_element_type(pcoll.element_type)\n      except TypeError as exn:\n        raise ValueError(\n            \"An explicit schema is required to write non-schema'd PCollections.\"\n        ) from exn\n      avro_schema = beam_schema_to_avro_schema(beam_schema)\n      records = pcoll | beam.Map(\n          beam_row_to_avro_dict(avro_schema, beam_schema))\n    self._sink = self._sink_provider(avro_schema)\n    if (not pcoll.is_bounded and self._sink.shard_name_template\n        == filebasedsink.DEFAULT_SHARD_NAME_TEMPLATE):\n      self._sink.shard_name_template = (\n          filebasedsink.DEFAULT_WINDOW_SHARD_NAME_TEMPLATE)\n      self._sink.shard_name_format = self._sink._template_to_format(\n          self._sink.shard_name_template)\n      self._sink.shard_name_glob_format = self._sink._template_to_glob_format(\n          self._sink.shard_name_template)\n\n    return records | beam.io.iobase.Write(self._sink)\n\n  def display_data(self):","sourceCodeStart":417,"sourceCodeEnd":453,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/avroio.py#L417-L453","documentation":"apache_beam.io.avroio.WriteToAvro.expand can derive an Avro schema automatically only when the incoming PCollection has a Beam schema (elements are schema'd types like Beam Rows, NamedTuples, or dataclasses). For plain PCollection element types, deriving the schema raises TypeError, which expand re-raises as ValueError telling you an explicit schema is required. WriteToAvro then needs a concrete Avro schema to create its sink.","triggerScenarios":"Writing a PCollection of plain dicts, bytes, or arbitrary classes (no registered Beam schema) to WriteToAvro without passing schema=..., so schemas.schema_from_element_type(pcoll.element_type) fails.","commonSituations":"Piping avroio.WriteToAvro after a plain Map producing dicts; forgetting to convert elements to a schema'd type (beam.Row, NamedTuple, dataclass with @dataclass and typing annotations); migrating code from the old avroio API that always required explicit schemas.","solutions":["Pass an explicit schema to WriteToAvro: WriteToAvro(path, schema=your_avro_schema_dict)","Or write schema'd elements: produce beam.Row(...) objects or NamedTuples/dataclasses with type annotations","Register a Beam schema for your custom class via beamAi schema registration so schema_from_element_type succeeds","Check pcoll.element_type: if it is not schema'd, add a Map to beam.Row before the sink"],"exampleFix":"// before\npcoll | beam.Map(lambda x: {'name': x[0], 'age': x[1]}) | avroio.WriteToAvro('out.avro')\n// after\npcoll | beam.Map(lambda x: beam.Row(name=x[0], age=x[1])) | avroio.WriteToAvro('out.avro')\n# or: avroio.WriteToAvro('out.avro', schema={'type':'record','name':'R','fields':[...]})","handlingStrategy":"validation","validationCode":"has_schema = isinstance(pcoll.element_type, (beam.RowTypeConstraint,)) or hasattr(pcoll.element_type, '_beam_schema')\nassert has_schema or schema is not None, 'pass schema= or write schema-d elements'","typeGuard":"def is_schema_d(pcoll):\n    try:\n        from apache_beam.typehint.schemas import schema_from_element_type\n        schema_from_element_type(pcoll.element_type)\n        return True\n    except TypeError:\n        return False","tryCatchPattern":"try:\n    result = pcoll | avroio.WriteToAvro(path)\nexcept ValueError as e:\n    if 'explicit schema is required' in str(e):\n        pcoll | beam.Map(lambda x: beam.Row(**x)) | avroio.WriteToAvro(path)","preventionTips":["Always pass schema= to WriteToAvro unless elements are Beam-schema'd types","Convert dicts to beam.Row() before writing","Prefer NamedTuples/dataclasses with type annotations as pipeline element types"],"tags":["python","apache-beam","avro","schema","io"],"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"}