apache/beam · error · ValueError

A schema is required in order to prepare rows for writing…

Error message

A schema is required in order to prepare rows for writing with STORAGE_WRITE_API.

What it means

StorageWriteToBigQuery.expand() requires a schema: if none was supplied and none can be inferred from the element type (schema_from_element_type raises TypeError), it re-raises as ValueError. The Storage Write API needs a proto/schema to encode rows.

Solutions

Pass an explicit schema argument to WriteToBigQuery.
Use typed elements (Beam Rows / namedtuples) or apply a schema via apache_beam.pvalue.AsSchema / beam.Map to a schema'd type.
Ensure element type is not untyped dict before the write.

Example fix

// before
rows | WriteToBigQuery(table='proj:ds.tbl', method='STORAGE_WRITE_API')
// after
rows | WriteToBigQuery(table='proj:ds.tbl', method='STORAGE_WRITE_API', schema='id:STRING,value:FLOAT')

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.typehints.schemas import schema_from_element_type
try:
    schema_from_element_type(pcoll.element_type)
except TypeError:
    pass  # need explicit schema

Try / catch

try:
    result = (rows | WriteToBigQuery(method='STORAGE_WRITE_API'))
except ValueError as e:
    if 'schema is required' in str(e):
        result = (rows | WriteToBigQuery(method='STORAGE_WRITE_API', schema=my_schema))

Prevention

Attach a Beam schema to your elements before STORAGE_WRITE_API writes.
Always pass schema explicitly for Storage API writes.

When it happens

Trigger: WriteToBigQuery(method=STORAGE_WRITE_API) with schema=None on a PCollection of plain dicts (no Beam schema registered).

Common situations: Streaming pipelines of untyped dict rows switched to STORAGE_WRITE_API; forgetting to define a Beam schema on the element type.

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

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

Appendix: source

Thrown at sdks/python/apache_beam/io/gcp/bigquery.py:2763

    self.additional_bq_parameters = additional_bq_parameters
    self._triggering_frequency = triggering_frequency
    self._use_at_least_once = use_at_least_once
    self._with_auto_sharding = with_auto_sharding
    self._num_storage_api_streams = num_storage_api_streams
    self._use_cdc_writes = use_cdc_writes
    self._primary_key = primary_key
    self._big_lake_configuration = big_lake_configuration
    self._type_overrides = type_overrides
    self._expansion_service = expansion_service or BeamJarExpansionService(
        'sdks:java:io:google-cloud-platform:expansion-service:build')

  def expand(self, input):
    if self._schema is None:
      try:
        schema = schema_from_element_type(input.element_type)
        is_rows = True
      except TypeError as exn:
        raise ValueError(
            "A schema is required in order to prepare rows "
            "for writing with STORAGE_WRITE_API.") from exn
    elif callable(self._schema):
      raise NotImplementedError(
          "Writing with dynamic schemas is not "
          "supported for this write method.")
    elif isinstance(self._schema, vp.ValueProvider):
      schema = self._schema.get()
      is_rows = False
    else:
      schema = self._schema
      is_rows = False

    table = bigquery_tools.get_hashable_destination(self._table)

    # if writing to one destination, just convert to Beam rows and send over
    if not callable(table):
      if is_rows:

View on GitHub (pinned to 12126d8942)