apache/beam · error · ValueError

Model Handler does not implement a default preprocess…

Error message

Model Handler does not implement a default preprocess method. Please define a preprocessing method using the 'preprocess' tag. This is required in most cases because most models will have a different input shape, so the model cannot generalize how the input Row should be transformed. For an example preprocess method, see VertexAIModelHandlerJSONProvider

What it means

Beam YAML requires a preprocessing function for ML transforms because model input shapes vary. `default_preprocess_fn` (yaml_ml.py:115) is a placeholder that always raises ValueError telling the user to define a `preprocess` config; it is invoked when no custom preprocess function was supplied for the model handler.

Solutions

Add a `preprocess` spec to the transform: {path: preprocess.py, name: my_preprocess_fn} or an inline `callable`.
Implement a default_preprocess_fn on a custom ModelHandlerProvider so YAML users get sensible defaults.
Use VertexAIModelHandlerJSONProvider (or another handler with a built-in default preprocess) if no custom preprocessing is needed.

Example fix

# before
- type: RunInference
  model_handler: {type: MyHandler, ...}
# after
- type: RunInference
  model_handler: {type: MyHandler, ...}
  preprocess: {path: preprocess.py, name: preprocess_fn}

Defensive patterns

Strategy: fallback

Validate before calling

handler_provider = ModelHandlerProvider.handler_types.get(spec['type'])
if handler_provider and handler_provider.default_preprocess_fn.__qualname__ == 'default_preprocess_fn':
    assert 'preprocess' in cfg, 'this handler has no default preprocess; define one'

Type guard

def needs_explicit_preprocess(provider_cls):
    import inspect
    src = inspect.getsource(provider_cls.default_preprocess_fn)
    return 'raise' in src  # placeholder that raises

Try / catch

try:
    transform = RunInferenceYamlTransform(cfg)
except ValueError as e:
    if 'default preprocess' in str(e):
        cfg['preprocess'] = {'path': 'preprocess.py', 'name': 'preprocess_fn'}
        transform = RunInferenceYamlTransform(cfg)
    else:
        raise

Prevention

Always define a preprocess step for custom model handlers in YAML pipelines.
Implement default_preprocess_fn on custom providers when a sensible default exists.
Prefer handlers with built-in defaults (e.g. VertexAIModelHandlerJSONProvider) for zero-config use.

When it happens

Trigger: Running an ML transform (e.g. RunInference via YAML) on a model handler that lacks a built-in default preprocess, without specifying a `preprocess` callable/path in the transform config.

Common situations: First-time users wiring a custom model handler without implementing a default preprocess; omitting the preprocess block because examples with VertexAIModelHandlerJSONProvider (which has a default) worked before.

Understand the failure class

Background: "X is required", "must be set", "cannot be empty": the missing-required-config error family, from Vertex AI project/location to WeChat keys — this error's family across 18 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_ml.py:115

      elif callable:
        return python_callable.PythonCallableWithSource(callable)
      else:
        raise ValueError(
            f"Must specify one of 'callable' or 'path' and 'name' for {typ} "
            f"function.")

    if processing_transform:
      if isinstance(processing_transform, dict):
        return _parse_config(**processing_transform)
      else:
        raise ValueError("Invalid model_handler specification.")

  def underlying_handler(self):
    return self._handler

  @staticmethod
  def default_preprocess_fn():
    raise ValueError(
        'Model Handler does not implement a default preprocess '
        'method. Please define a preprocessing method using the '
        '\'preprocess\' tag. This is required in most cases because '
        'most models will have a different input shape, so the model '
        'cannot generalize how the input Row should be transformed. For '
        'an example preprocess method, see VertexAIModelHandlerJSONProvider')

  def _preprocess_fn_internal(self):
    return lambda row: (row, self._preprocess_fn(row))

  @staticmethod
  def default_postprocess_fn():
    return lambda x: x

  def _postprocess_fn_internal(self):
    return lambda result: (result[0], self._postprocess_fn(result[1]))

  @staticmethod

View on GitHub (pinned to 12126d8942)