apache/beam · error · ValueError

An unsupported sink was specified

Error message

An unsupported sink was specified: '%s'. Please specify one of the following sinks: %s

What it means

Managed transform constructor validates that the `sink` argument is one of the supported write transforms (e.g. 'iceberg', 'bigquery'). It lowercases the sink string and looks it up in the _WRITE_TRANSFORMS map; if absent, ValueError is raised listing valid sinks.

Solutions

1. Use one of the sinks listed in the error message (e.g. 'iceberg', 'bigquery').
2. Check the exact spelling of the sink name against apache_beam/transforms/managed.py _WRITE_TRANSFORMS.
3. Upgrade apache_beam if the sink you need was added in a newer release.

Example fix

// before
beam.managed.Write('iceburg')
// after
beam.managed.Write('iceberg')

from apache_beam.transforms.managed import Write
SUPPORTED = list(Write._WRITE_TRANSFORMS.keys())
assert sink and sink.lower() in SUPPORTED, f"sink must be one of {SUPPORTED}"

When it happens

Trigger: Calling apache_beam.transforms.managed.Write(sink='iceburg') (typo), or with a sink name not in _WRITE_TRANSFORMS keys, or an empty/None sink string.

Common situations: Typos in sink names, copying examples from an older Beam version whose managed API supported different sinks, passing uppercase/mixed-case names that don't match (handled by .lower(), so usually typos or unsupported sinks).

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• At least one of --render_port or --render_output must be…
• buffer_sec must be >= 0, got
• Cannot skip negative number of header lines
• change_function must be 'CHANGES' or 'APPENDS', got
• CustomQueryConfig must provide a valid query_fn

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/managed.py:171

      ICEBERG: ManagedTransforms.Urns.ICEBERG_WRITE.urn,
      KAFKA: ManagedTransforms.Urns.KAFKA_WRITE.urn,
      BIGQUERY: ManagedTransforms.Urns.BIGQUERY_WRITE.urn,
      POSTGRES: ManagedTransforms.Urns.POSTGRES_WRITE.urn,
      MYSQL: ManagedTransforms.Urns.MYSQL_WRITE.urn,
      SQL_SERVER: ManagedTransforms.Urns.SQL_SERVER_WRITE.urn
  }

  def __init__(
      self,
      sink: str,
      config: Optional[dict[str, Any]] = None,
      config_url: Optional[str] = None,
      expansion_service=None):
    super().__init__()
    self._sink = sink
    identifier = self._WRITE_TRANSFORMS.get(sink.lower())
    if not identifier:
      raise ValueError(
          f"An unsupported sink was specified: '{sink}'. Please specify "
          f"one of the following sinks: {list(self._WRITE_TRANSFORMS.keys())}")

    # Store parameters for deferred expansion service creation
    self._identifier = identifier
    self._provided_expansion_service = expansion_service
    self._underlying_identifier = identifier
    self._yaml_config = yaml.dump(config)
    self._config_url = config_url

  def expand(self, input):
    # Create expansion service with access to pipeline options
    expansion_service = _resolve_expansion_service(
        self._sink,
        self._identifier,
        self._provided_expansion_service,
        pipeline_options=input.pipeline._options)

View on GitHub (pinned to 12126d8942)