apache/beam · error · ValueError

A schema is required to write non-schema'd data.

Error message

A schema is required to write non-schema'd data.

What it means

_WriteToParquet.expand needs a schema to know how to convert incoming rows into Arrow tables. If the sink was constructed without a `schema` and the input PCollection's element type is not a Beam schema'd row class (so schema_from_element_type fails), the writer cannot proceed and raises this ValueError.

Source

Thrown at sdks/python/apache_beam/io/parquetio.py:623

          mime_type,
          triggering_frequency
      )

  def expand(self, pcoll):
    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)

    if self._schema is None:
      try:
        beam_schema = schemas.schema_from_element_type(pcoll.element_type)
      except TypeError as exn:
        raise ValueError(
            "A schema is required to write non-schema'd data.") from exn
      self._sink._schema = (
          arrow_type_compatibility.arrow_schema_from_beam_schema(beam_schema))
      convert_fn = _BeamRowsToArrowTable()
    else:
      convert_fn = _RowDictionariesToArrowTable(
          self._schema, self._row_group_buffer_size, self._record_batch_size)
    if pcoll.is_bounded:
      return pcoll | ParDo(convert_fn) | Write(self._sink)
    else:
      self._sink.convert_fn = convert_fn
      return pcoll | Write(self._sink)

  def display_data(self):
    return {
        'sink_dd': self._sink,
        'row_group_buffer_size': str(self._row_group_buffer_size)
    }

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass schema= to WriteToParquet, e.g. WriteToParquet(path, schema=pa.schema([('name', pa.string()), ('age', pa.int64())])).
2. Convert input to a Beam schema'd row class (beam.Row or NamedTuple registered via @dataclass with beam type hints) so the schema can be inferred.
3. Define the schema from the data with beam.Row(...) type hints on the PCollection.

Example fix

// before
result | WriteToParquet('out.parquet')  # dict rows, no schema
// after
result | WriteToParquet('out.parquet', schema=pa.schema([('name', pa.string()), ('age', pa.int64())]))

if schema is None and not hasattr(element_type, 'beam_schema'):
    schema = pa.schema([('col1', pa.string())])  # derive or define explicitly

def has_beam_schema(element_type) -> bool:
    try:
        schemas.schema_from_element_type(element_type)
        return True
    except TypeError:
        return False

try:
    result |= WriteToParquet(path, schema=schema)
except ValueError as e:
    if "schema is required" in str(e):
        result |= WriteToParquet(path, schema=derive_schema(pcoll))

When it happens

Trigger: beam.io.parquetio.WriteToParquet(file_path_prefix) without a `schema=` argument, writing a PCollection of plain dicts or other non-schema'd types. If the element type is a schema'd row class, the schema is inferred instead and no error occurs.

Common situations: Writing dict rows after migrating code that previously passed a schema; forgetting schema= after refactoring the input element type away from a NamedTuple/Beam Row.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• An explicit schema is required to write non-schema'd PCollec
• WriteToText requires an input schema with exactly one field.
• WriteToText requires an input schema with exactly one field,
• Encountered an Atomic type that is not currently supported b
• Schema with id {schema.id} has encoding_positions_set=True,

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