apache/beam · error · ValueError

An explicit schema is required to write non-schema'd PCollec

Error message

An explicit schema is required to write non-schema'd PCollections.

What it means

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.

Source

Thrown at sdks/python/apache_beam/io/avroio.py:435

        If set it overrides user windowing. Mandatory for GlobalWindow.

    Returns:
      A WriteToAvro transform usable for writing.
    """
    self._schema = schema
    self._sink_provider = lambda avro_schema: _create_avro_sink(
        file_path_prefix, avro_schema, codec, file_name_suffix, num_shards,
        shard_name_template, mime_type, triggering_frequency)

  def expand(self, pcoll):
    if self._schema:
      avro_schema = self._schema
      records = pcoll
    else:
      try:
        beam_schema = schemas.schema_from_element_type(pcoll.element_type)
      except TypeError as exn:
        raise ValueError(
            "An explicit schema is required to write non-schema'd PCollections."
        ) from exn
      avro_schema = beam_schema_to_avro_schema(beam_schema)
      records = pcoll | beam.Map(
          beam_row_to_avro_dict(avro_schema, beam_schema))
    self._sink = self._sink_provider(avro_schema)
    if (not pcoll.is_bounded and self._sink.shard_name_template
        == filebasedsink.DEFAULT_SHARD_NAME_TEMPLATE):
      self._sink.shard_name_template = (
          filebasedsink.DEFAULT_WINDOW_SHARD_NAME_TEMPLATE)
      self._sink.shard_name_format = self._sink._template_to_format(
          self._sink.shard_name_template)
      self._sink.shard_name_glob_format = self._sink._template_to_glob_format(
          self._sink.shard_name_template)

    return records | beam.io.iobase.Write(self._sink)

  def display_data(self):

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass an explicit schema to WriteToAvro: WriteToAvro(path, schema=your_avro_schema_dict)
2. Or write schema'd elements: produce beam.Row(...) objects or NamedTuples/dataclasses with type annotations
3. Register a Beam schema for your custom class via beamAi schema registration so schema_from_element_type succeeds
4. Check pcoll.element_type: if it is not schema'd, add a Map to beam.Row before the sink

Example fix

// before
pcoll | beam.Map(lambda x: {'name': x[0], 'age': x[1]}) | avroio.WriteToAvro('out.avro')
// after
pcoll | beam.Map(lambda x: beam.Row(name=x[0], age=x[1])) | avroio.WriteToAvro('out.avro')
# or: avroio.WriteToAvro('out.avro', schema={'type':'record','name':'R','fields':[...]})

has_schema = isinstance(pcoll.element_type, (beam.RowTypeConstraint,)) or hasattr(pcoll.element_type, '_beam_schema')
assert has_schema or schema is not None, 'pass schema= or write schema-d elements'

def is_schema_d(pcoll):
    try:
        from apache_beam.typehint.schemas import schema_from_element_type
        schema_from_element_type(pcoll.element_type)
        return True
    except TypeError:
        return False

try:
    result = pcoll | avroio.WriteToAvro(path)
except ValueError as e:
    if 'explicit schema is required' in str(e):
        pcoll | beam.Map(lambda x: beam.Row(**x)) | avroio.WriteToAvro(path)

When it happens

Trigger: 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.

Common situations: 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.

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

• Unable to convert {avro_type} to a Beam schema.
• Only strings allowed as map keys when converting from AVRO,
• Unrecognized type_info: {type_info!r}
• Unknown BigQuery field mode: {}
• A schema is required to write non-schema'd data.

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