apache/beam · error · NotImplementedError

Unexpected type for bad_results

Error message

Unexpected type for bad_results: {type(bad_results)}

What it means

When MLTransform's exception handling collects failed results, bad_results must be a RunInferenceDLQ or a beam.PCollection; any other type cannot be mapped to error rows, so NotImplementedError is raised naming the unexpected type.

Solutions

1. Pass a beam.PCollection of failed records as bad_results.
2. For RunInference failures, pass the RunInferenceDLQ object returned by the model handler setup.
3. Convert custom error containers to a PCollection first (beam.Create(...) | ...).

Example fix

// before
mltransform.with_exception_handling(bad_results=my_error_list)
// after
bad = failed_records | beam.Map(lambda x: ...)  # a PCollection
mltransform.with_exception_handling(bad_results=bad)

from apache_beam.ml.inference.base import RunInferenceDLQ
assert isinstance(bad_results, (RunInferenceDLQ, __import__('apache_beam', fromlist=['PCollection']).PCollection)), f'bad bad_results type: {type(bad_results)}'

def is_valid_bad_results(br) -> bool:
    import apache_beam as beam
    from apache_beam.ml.inference.base import RunInferenceDLQ
    return isinstance(br, (RunInferenceDLQ, beam.PCollection))

try:
    out = mltransform.with_exception_handling(bad_results=br)
except NotImplementedError as e:
    if 'Unexpected type for bad_results' in str(e):
        br = failures | beam.Map(lambda x: x)
        out = mltransform.with_exception_handling(bad_results=br)

When it happens

Trigger: Using with_exception_handling and supplying a bad_results sink/collector that is neither RunInferenceDLQ nor a PCollection (e.g. a plain list, callable, or custom DLQ object).

Common situations: Wiring a custom dead-letter container class instead of a PCollection; passing a function that returns a PCollection rather than the PCollection itself.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• A sink must inherit iobase.Sink, iobase.NativeSink, or be a…
• Cannot encode payload for WriteToPubSub. Expected valid…
• Cannot interpret as seconds.
• Cannot interpret as subseconds.
• compression_type must be CompressionType object but was

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

Appendix: source

Thrown at sdks/python/apache_beam/ml/transforms/base.py:437

      for i in range(len(ptransform_list)):
        if hasattr(ptransform_list[i], 'artifact_mode'):
          ptransform_list[i].artifact_mode = self._artifact_mode

    transform_name = None
    for ptransform in ptransform_list:
      if self._with_exception_handling:
        if hasattr(ptransform, 'with_exception_handling'):
          ptransform = ptransform.with_exception_handling(
              **self._exception_handling_args)
          pcoll, bad_results = pcoll | ptransform
          # RunInference outputs a RunInferenceDLQ instead of a PCollection.
          # since TFTProcessHandler and RunInferene are supported, try to infer
          # the type of bad_results and append it to the list of errors.
          if isinstance(bad_results, RunInferenceDLQ):
            bad_results = bad_results.failed_inferences
            transform_name = ptransform.annotations()['model_handler']
          elif not isinstance(bad_results, beam.PCollection):
            raise NotImplementedError(
                f'Unexpected type for bad_results: {type(bad_results)}')
          bad_results = bad_results | beam.Map(
              lambda x: _map_errors_to_beam_row(x, transform_name))
          upstream_errors.append(bad_results)

      else:
        pcoll = pcoll | ptransform
    _ = (
        pcoll.pipeline
        | "MLTransformMetricsUsage" >> MLTransformMetricsUsage(self))

    if self._with_exception_handling:
      bad_pcoll = (upstream_errors | beam.Flatten())
      return pcoll, bad_pcoll  # type: ignore[return-value]
    return pcoll  # type: ignore[return-value]

  def with_transform(self, transform: MLTransformProvider):
    """

View on GitHub (pinned to 12126d8942)