apache/beam · error · NotImplementedError

Currently, dynamic clustering and timepartitioning is not…

Error message

Currently, dynamic clustering and timepartitioning is not supported for STORAGE_WRITE_API write method.

What it means

StorageWriteToBigQuery.expand() raises NotImplementedError when additional_bq_parameters is callable. A callable implies per-destination dynamic time partitioning/clustering, which the Storage Write API path does not support; it only reads static 'clustering' fields from a dict.

Solutions

Pass additional_bq_parameters as a static dict, e.g. {'timePartitioning': {...}, 'clustering': {'fields': [...]}}.
Split the pipeline into multiple WriteToBigQuery transforms, one per partitioning configuration.
Use FILE_LOADS if dynamic per-destination clustering/partitioning is required.

Example fix

// before
WriteToBigQuery(method='STORAGE_WRITE_API', additional_bq_parameters=lambda dest: {'clustering': {'fields': ['date']}})
// after
WriteToBigQuery(method='STORAGE_WRITE_API', additional_bq_parameters={'clustering': {'fields': ['date']}})

Defensive patterns

Strategy: validation

Validate before calling

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

Type guard

def is_static_bq_params(p):
    return isinstance(p, dict) or p is None

Prevention

Use a static dict for partitioning/clustering with STORAGE_WRITE_API.
Split per-destination configs into separate writes instead of callables.

When it happens

Trigger: WriteToBigQuery(method=STORAGE_WRITE_API, additional_bq_parameters=lambda destination: {...}) expanded.

Common situations: Reusing the callable additional_bq_parameters pattern from FILE_LOADS dynamic destinations when migrating to STORAGE_WRITE_API.

Related errors

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

Appendix: source

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

                        StorageWriteToBigQuery.RECORD: row[1]
                    })).with_output_types(
                        RowTypeConstraint.from_fields([
                            (StorageWriteToBigQuery.DESTINATION, str),
                            (StorageWriteToBigQuery.RECORD, input.element_type)
                        ])))
      # otherwise, convert to Beam Rows
      else:
        input_beam_rows = (
            input_rows
            | "Convert dict to Beam Row" >> self.ConvertToBeamRows(
                schema, True, self._type_overrides).with_output_types())
      # communicate to Java that this write should use dynamic destinations
      table = StorageWriteToBigQuery.DYNAMIC_DESTINATIONS

    clustering_fields = []
    if self.additional_bq_parameters:
      if callable(self.additional_bq_parameters):
        raise NotImplementedError(
            "Currently, dynamic clustering and timepartitioning is not "
            "supported for STORAGE_WRITE_API write method.")
      clustering_fields = (
          self.additional_bq_parameters.get("clustering", {}).get("fields", []))

    output = (
        input_beam_rows
        | SchemaAwareExternalTransform(
            identifier=StorageWriteToBigQuery.IDENTIFIER,
            expansion_service=self._expansion_service,
            rearrange_based_on_discovery=True,
            table=table,
            create_disposition=self._create_disposition,
            write_disposition=self._write_disposition,
            triggering_frequency_seconds=self._triggering_frequency,
            auto_sharding=self._with_auto_sharding,
            num_streams=self._num_storage_api_streams,
            use_at_least_once_semantics=self._use_at_least_once,

View on GitHub (pinned to 12126d8942)