apache/beam · error · NotImplementedError

Writing with dynamic schemas is not supported for this…

Error message

Writing with dynamic schemas is not supported for this write method.

What it means

StorageWriteToBigQuery.expand() raises NotImplementedError when schema is a callable, because the Storage Write API path resolves the schema once and cannot compute a schema dynamically per element or destination.

Solutions

Provide a static schema (string, dict, or ValueProvider) instead of a callable.
Resolve the schema yourself before the write (e.g. with a branch per destination and separate WriteToBigQuery transforms).
Use FILE_LOADS or STREAMING_INSERTS which support dynamic schemas.

Example fix

// before
WriteToBigQuery(method='STORAGE_WRITE_API', schema=lambda dest: schema_for(dest))
// after
WriteToBigQuery(method='STORAGE_WRITE_API', schema=schema_for(DEFAULT_DEST))

Defensive patterns

Strategy: validation

Validate before calling

if method == 'STORAGE_WRITE_API' and callable(schema):
    raise ValueError('STORAGE_WRITE_API needs a static schema')

Prevention

Resolve schemas per destination before the write using branch transforms.
Document that dynamic schemas are FILE_LOADS/STREAMING_INSERTS only.

When it happens

Trigger: WriteToBigQuery(method=STORAGE_WRITE_API, schema=lambda destination: ...) expanded.

Common situations: Migrating from FILE_LOADS (which supports callable schemas for dynamic destinations) to STORAGE_WRITE_API without flattening the callable.

Related errors

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

Appendix: source

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

    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:
        input_beam_rows = input
      else:
        input_beam_rows = (
            input

View on GitHub (pinned to 12126d8942)