apache/beam · error · RuntimeError

MLTransform only supports GlobalWindows when producing artif

Error message

MLTransform only supports GlobalWindows when producing artifacts such as min, max, variance etc over the dataset.Please use beam.WindowInto(beam.transforms.window.GlobalWindows()) to convert your PCollection to GlobalWindow.

What it means

MLTransform with TensorFlow Transform must compute dataset-level statistics (min, max, variance, vocab, etc.), which requires all data in a single GlobalWindow. _fail_on_non_default_windowing raises a RuntimeError if the input PCollection uses any non-default windowing (fixed, sliding, session windows), since per-window artifact computation is unsupported.

Source

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

        raw_data_feature_spec)

  def write_transform_artifacts(self, transform_fn, location):
    """
    Write transform artifacts to the given location.
    Args:
      transform_fn: A transform_fn object.
      location: A location to write the artifacts.
    Returns:
      A PCollection of WriteTransformFn writing a TF transform graph.
    """
    return (
        transform_fn
        | 'Write Transform Artifacts' >>
        transform_fn_io.WriteTransformFn(location))

  def _fail_on_non_default_windowing(self, pcoll: beam.PCollection):
    if not pcoll.windowing.is_default():
      raise RuntimeError(
          "MLTransform only supports GlobalWindows when producing "
          "artifacts such as min, max, variance etc over the dataset."
          "Please use beam.WindowInto(beam.transforms.window.GlobalWindows()) "
          "to convert your PCollection to GlobalWindow.")

  def process_data_fn(
      self, inputs: dict[str, common_types.ConsistentTensorType]
  ) -> dict[str, common_types.ConsistentTensorType]:
    """
    This method is used in the AnalyzeAndTransformDataset step. It applies
    the transforms to the `inputs` in sequential order on the columns
    provided for a given transform.
    Args:
      inputs: A dictionary of column names and data.
    Returns:
      A dictionary of column names and transformed data.
    """
    outputs = inputs.copy()

View on GitHub (pinned to 12126d8942)

Solutions

1. Wrap the input PCollection with beam.WindowInto(beam.transforms.window.GlobalWindows()) immediately before MLTransform.
2. Restructure the pipeline so windowing for other aggregations happens after the MLTransform step.
3. If windowing is inherent to the source (streaming), batch the data or switch to a non-TFT transform that supports per-window processing.

Example fix

# before
windowed = pcoll | beam.WindowInto(beam.window.FixedWindows(60))
result = windowed | MLTransform(...).with_write_artifact_location(loc)

# after
global_w = pcoll | beam.WindowInto(beam.transforms.window.GlobalWindows())
result = global_w | MLTransform(...).with_write_artifact_location(loc)

if not pcoll.windowing.is_default():
    pcoll = pcoll | beam.WindowInto(beam.transforms.window.GlobalWindows())

def uses_global_windows(pcoll) -> bool:
    return pcoll.windowing.is_default()

try:
    result = pcoll | MLTransform(...).with_write_artifact_location(loc)
except RuntimeError as e:
    if 'GlobalWindows' in str(e):
        pcoll = pcoll | 'ReWindowToGlobal' >> beam.WindowInto(beam.transforms.window.GlobalWindows())
        result = pcoll | MLTransform(...).with_write_artifact_location(loc)
    else:
        raise

When it happens

Trigger: Applying a beam.WindowInto with fixed/sliding/session windows (or inheriting windowing from a streaming source) upstream of an MLTransform that produces artifacts via write_artifact_location.

Common situations: Streaming pipelines with windowed PCollections, or batch pipelines where an earlier stage applied windowing for other aggregations and MLTransform is added downstream.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• To write a GlobalWindow unbounded PCollection, triggering_fr
• GroupByKey cannot be applied to an unbounded PCollection wit
• GroupByKey cannot be applied to non-bounded PCollection in t
• UnboundedSource restriction was neither finished nor checkpo
• Watermark must be monotonically increasing.Provided watermar

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