apache/beam · error · TypeError

Pickling error encountered while running inference. This…

Error message

Pickling error encountered while running inference. This may be caused by trying to send unpickleable data to a model which is shared across processes. For more information, see https://beam.apache.org/documentation/ml/large-language-modeling/#pickling-errors

What it means

Raised when a pickle.PickleError occurs during RunInference on a model configured with share_model_across_processes. Sharing a model across processes requires data sent to the model to be picklable; unpickleable data (locks, open handles, lambdas, etc.) breaks serialization and is re-raised as this TypeError with a link to Beam's pickling-error docs.

Solutions

Make all data sent to the model picklable (replace lambdas with top-level functions, remove open handles/locks).
Disable share_model_across_processes() if sharing is not required, so each process loads its own model.
Use dill-based serialization or convert custom objects to plain picklable types (dicts, bytes) before inference.

Example fix

// before
class Ctx:
  def __init__(self): self.fh = open('data.bin')  # unpickleable
rows = p | RunInference(handler.share_model_across_processes()) with Ctx() inputs
// after
rows = p | Map(lambda x: (x, read_bytes('data.bin'))) | RunInference(handler)  # pass picklable data, don't share model

Defensive patterns

Strategy: try-catch

Try / catch

try:
    result = beam.RunInference(handler.share_model_across_processes())
except TypeError as e:
    if 'Pickling error' in str(e):
        logging.error('Unpickleable data with shared model: %s', e.__cause__)
        raise
catch (e) { if (String(e).includes('Pickling error')) { /* fix data picklability or unshare model */ } }

Prevention

Prefer top-level functions over lambdas in data flowing to shared models.
Ensure model inputs contain only picklable types (str, bytes, dict, list).
Only enable share_model_across_processes when data is known picklable.

When it happens

Trigger: Running inference on a model handler with share_model_across_processes() enabled while the batch contains objects that cannot be pickled (e.g. file handles, threads, local lambdas, C-extension objects).

Common situations: Passing custom non-picklable objects as model input; large-language-model pipelines with shared models on multi-worker runners; lambda functions defined in __main__ used in preprocessing outputs.

Understand the failure class

Background: json.Marshal / "failed to marshal" errors in Go: why "unsupported type" happens and how to fix it — this error's family across 22 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/ml/inference/base.py:2112

      model = self._model.next_model()
      if isinstance(model, str):
        # ModelManager with MultiProcessShared returns the model tag
        unique_tag = model
        model = multi_process_shared.MultiProcessShared(
            lambda: None, tag=model, always_proxy=True).acquire()
      try:
        result_generator = (OOMProtectedFn(self._model_handler.run_inference))(
            batch, model, inference_args)
      finally:
        # Always release the model so that it can be reloaded.
        if self.use_model_manager:
          self._model.release_model(self._model_tag, unique_tag)
    except BaseException as e:
      if self._metrics_collector:
        self._metrics_collector.failed_batches_counter.inc()
      if (e is pickle.PickleError and
          self._model_handler.share_model_across_processes()):
        raise TypeError(
            'Pickling error encountered while running inference. '
            'This may be caused by trying to send unpickleable '
            'data to a model which is shared across processes. '
            'For more information, see '
            'https://beam.apache.org/documentation/ml/large-language-modeling/#pickling-errors'  # pylint: disable=line-too-long
        ) from e
      raise e
    predictions = list(result_generator)

    end_time = _to_microseconds(self._clock.time_ns())
    inference_latency = end_time - start_time
    num_bytes = self._model_handler.get_num_bytes(batch)
    num_elements = len(batch)
    if self._metrics_collector:
      self._metrics_collector.update(num_elements, num_bytes, inference_latency)

    return predictions

View on GitHub (pinned to 12126d8942)