{"record":{"id":"1d9d2c22094535ea","repo":"apache/beam","slug":"mltransform-only-supports-globalwindows-when-producing","errorCode":null,"errorMessage":"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.","messagePattern":"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\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/ml/transforms/handlers.py","lineNumber":300,"sourceCode":"        raw_data_feature_spec)\n\n  def write_transform_artifacts(self, transform_fn, location):\n    \"\"\"\n    Write transform artifacts to the given location.\n    Args:\n      transform_fn: A transform_fn object.\n      location: A location to write the artifacts.\n    Returns:\n      A PCollection of WriteTransformFn writing a TF transform graph.\n    \"\"\"\n    return (\n        transform_fn\n        | 'Write Transform Artifacts' >>\n        transform_fn_io.WriteTransformFn(location))\n\n  def _fail_on_non_default_windowing(self, pcoll: beam.PCollection):\n    if not pcoll.windowing.is_default():\n      raise RuntimeError(\n          \"MLTransform only supports GlobalWindows when producing \"\n          \"artifacts such as min, max, variance etc over the dataset.\"\n          \"Please use beam.WindowInto(beam.transforms.window.GlobalWindows()) \"\n          \"to convert your PCollection to GlobalWindow.\")\n\n  def process_data_fn(\n      self, inputs: dict[str, common_types.ConsistentTensorType]\n  ) -> dict[str, common_types.ConsistentTensorType]:\n    \"\"\"\n    This method is used in the AnalyzeAndTransformDataset step. It applies\n    the transforms to the `inputs` in sequential order on the columns\n    provided for a given transform.\n    Args:\n      inputs: A dictionary of column names and data.\n    Returns:\n      A dictionary of column names and transformed data.\n    \"\"\"\n    outputs = inputs.copy()","sourceCodeStart":282,"sourceCodeEnd":318,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/ml/transforms/handlers.py#L282-L318","documentation":"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.","triggerScenarios":"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.","commonSituations":"Streaming pipelines with windowed PCollections, or batch pipelines where an earlier stage applied windowing for other aggregations and MLTransform is added downstream.","solutions":["Wrap the input PCollection with beam.WindowInto(beam.transforms.window.GlobalWindows()) immediately before MLTransform.","Restructure the pipeline so windowing for other aggregations happens after the MLTransform step.","If windowing is inherent to the source (streaming), batch the data or switch to a non-TFT transform that supports per-window processing."],"exampleFix":"# before\nwindowed = pcoll | beam.WindowInto(beam.window.FixedWindows(60))\nresult = windowed | MLTransform(...).with_write_artifact_location(loc)\n\n# after\nglobal_w = pcoll | beam.WindowInto(beam.transforms.window.GlobalWindows())\nresult = global_w | MLTransform(...).with_write_artifact_location(loc)","handlingStrategy":"validation","validationCode":"if not pcoll.windowing.is_default():\n    pcoll = pcoll | beam.WindowInto(beam.transforms.window.GlobalWindows())","typeGuard":"def uses_global_windows(pcoll) -> bool:\n    return pcoll.windowing.is_default()","tryCatchPattern":"try:\n    result = pcoll | MLTransform(...).with_write_artifact_location(loc)\nexcept RuntimeError as e:\n    if 'GlobalWindows' in str(e):\n        pcoll = pcoll | 'ReWindowToGlobal' >> beam.WindowInto(beam.transforms.window.GlobalWindows())\n        result = pcoll | MLTransform(...).with_write_artifact_location(loc)\n    else:\n        raise","preventionTips":["Place MLTransform before any WindowInto in batch pipelines that produce artifacts.","For streaming, re-window to GlobalWindows immediately before MLTransform.","Document the GlobalWindow requirement in pipeline construction helpers.","Check pcoll.windowing.is_default() in tests that build MLTransform pipelines."],"tags":["python","apache-beam","windowing","streaming"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}