{"record":{"id":"c0aa9aac7fd68297","repo":"apache/beam","slug":"input-to-impulse-transform-must-be-a-pbegin-but-found-s","errorCode":null,"errorMessage":"Input to Impulse transform must be a PBegin but found %s","messagePattern":"Input to Impulse transform must be a PBegin but found (.+?)","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/core.py","lineNumber":4315,"sourceCode":"\n  @staticmethod\n  def _create_source_from_iterable(values, coder):\n    return Create._create_source(list(map(coder.encode, values)), coder)\n\n  @staticmethod\n  def _create_source(serialized_values, coder):\n    # type: (typing.Any, typing.Any) -> create_source._CreateSource\n    from apache_beam.transforms.create_source import _CreateSource\n\n    return _CreateSource(serialized_values, coder)\n\n\n@typehints.with_output_types(bytes)\nclass Impulse(PTransform):\n  \"\"\"Impulse primitive.\"\"\"\n  def expand(self, pbegin):\n    if not isinstance(pbegin, pvalue.PBegin):\n      raise TypeError(\n          'Input to Impulse transform must be a PBegin but found %s' % pbegin)\n    return pvalue.PCollection(pbegin.pipeline, element_type=bytes)\n\n  def get_windowing(self, inputs):\n    # type: (typing.Any) -> Windowing\n    return Windowing(GlobalWindows())\n\n  def infer_output_type(self, unused_input_type):\n    return bytes\n\n  def to_runner_api_parameter(self, unused_context):\n    # type: (PipelineContext) -> typing.Tuple[str, None]\n    return common_urns.primitives.IMPULSE.urn, None\n\n  @staticmethod\n  @PTransform.register_urn(common_urns.primitives.IMPULSE.urn, None)\n  def from_runner_api_parameter(\n      unused_ptransform, unused_parameter, unused_context):","sourceCodeStart":4297,"sourceCodeEnd":4333,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/core.py#L4297-L4333","documentation":"Impulse is a pipeline-root primitive that must take a PBegin as its input. Beam raises TypeError when something other than PBegin (e.g. a PCollection) is fed into the Impulse transform, because Impulse can only start a pipeline, not consume an existing PCollection.","triggerScenarios":"Applying beam.Impulse() to a PCollection (e.g. pcoll | beam.Impulse()) instead of to the pipeline object; composing Impulse as a downstream step inside another transform whose input is a PCollection.","commonSituations":"Misplacing Impulse mid-pipeline in test helpers; copy-paste from examples where Impulse was the first stage; trying to 'restart' a pipeline from within a branch.","solutions":["Apply Impulse to the pipeline root: beam.Impulse() must be the first transform: p | beam.Impulse() | ...","Remove Impulse from mid-pipeline positions; use an existing PCollection directly.","If you need to synthesize elements mid-pipeline, use beam.Create([...]) on the pipeline instead."],"exampleFix":"// before\nresult = pcoll | beam.Impulse() | beam.Map(lambda x: b'data')\n// after\nresult = p | beam.Impulse() | beam.Map(lambda x: b'data')","handlingStrategy":"type-guard","validationCode":"import apache_beam as beam\nfrom apache_beam import pvalue\nassert isinstance(root, pvalue.PBegin), 'Impulse must be applied to the pipeline root'","typeGuard":"def can_apply_impulse(x) -> bool:\n    from apache_beam import pvalue\n    return isinstance(x, pvalue.PBegin)","tryCatchPattern":"try:\n    out = source | beam.Impulse()\nexcept TypeError as e:\n    if 'must be a PBegin' in str(e):\n        out = pipeline | beam.Impulse()\n    else:\n        raise","preventionTips":["Only apply Impulse as the first stage of a pipeline","Use beam.Create for mid-pipeline synthetic data","Keep pipeline-graph construction code separate from transform composition"],"tags":["apache-beam","python","pipeline-graph"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}