apache/beam · error · NotImplementedError

with_exception_handling with TensorFlow Transform-based…

Error message

with_exception_handling with TensorFlow Transform-based MLTransform operations is not supported. To enable exception handling for those operations, please create a separate MLTransform instance

What it means

MLTransform backed by TensorFlow Transform explicitly overrides with_exception_handling() to raise NotImplementedError. Exception handling (bad-row routing) is not implemented for TFT-based transforms, so callers must not invoke this method on TFT MLTransform instances.

Solutions

Remove the with_exception_handling() call from the TFT-based MLTransform chain.
Create a separate MLTransform instance for the TFT operations as the message suggests, and handle failures manually via a beam.Map/DoFn with try/except around data prep.
Use the non-TFT transform implementations in apache_beam.ml.transforms that do support exception handling if bad-row routing is required.

Example fix

# before
result = pcoll | MLTransform(tft.ScaleToZScore('x')).with_exception_handling().with_write_artifact_location(loc)

# after
result = pcoll | MLTransform(tft.ScaleToZScore('x')).with_write_artifact_location(loc)

Defensive patterns

Strategy: type-guard

Validate before calling

def is_tft_transform(cfg) -> bool:
    import apache_beam.ml.transforms.tft as tft
    return isinstance(cfg, tuple(c for c in vars(tft).values() if isinstance(c, type) and issubclass(c, object) and c.__module__ == tft.__name__))

Type guard

def supports_exception_handling(mltransform) -> bool:
    # TFTProcessHandler-backed MLTransform raises NotImplementedError
    import inspect
    try:
        mltransform.with_exception_handling
    except NotImplementedError:
        return False
    return True

Try / catch

try:
    t = mltransform.with_exception_handling()
except NotImplementedError:
    t = mltransform  # proceed without exception handling; handle bad rows manually

Prevention

Only call with_exception_handling() on non-TFT MLTransform pipelines.
Wrap raw data validation in a prior beam.Map with try/except for TFT pipelines.
Check the Beam version's docs for TFT handler limitations before enabling bad-row routing.

When it happens

Trigger: Calling .with_exception_handling() on an MLTransform whose transforms come from apache_beam.ml.transforms.tft (TFT-based handlers), e.g. ScaleToZScore, ComputeAndApplyVocab, etc.

Common situations: Copy-pasting exception-handling configuration from non-TFT (e.g. torch/sklearn handlers) MLTransform usage onto a TFT pipeline.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/ml/transforms/handlers.py:479

      # transformed_dataset.
      del self.transformed_schema[_TEMP_KEY]
      row_type = RowTypeConstraint.from_fields(
          list(self.transformed_schema.items()))

      # Decode the extra columns that were encoded as bytes.
      transformed_dataset = (
          transformed_dataset
          |
          "DecodeUnmodifiedColumns" >> beam.Map(lambda x: data_coder.decode(x)))
      # The schema only contains the columns that are transformed.
      transformed_dataset = (
          transformed_dataset
          | "ConvertToRowType" >>
          beam.Map(lambda x: beam.Row(**x)).with_output_types(row_type))
      return transformed_dataset

  def with_exception_handling(self):
    raise NotImplementedError(
        "with_exception_handling with TensorFlow Transform-based MLTransform "
        "operations is not supported. To enable exception handling for those "
        "operations, please create a separate MLTransform instance")

View on GitHub (pinned to 12126d8942)