apache/beam · error

invalid pipeline

Error message

invalid pipeline

What it means

After optional strict validation, the direct runner calls p.Build() to turn the pipeline into a graph of edges. If graph construction fails (invalid transform wiring, nil inputs, duplicate consumption of a PCollection), the error is wrapped as 'invalid pipeline'. It signals the pipeline could not be compiled into an execution graph at all.

Source

Thrown at sdks/go/pkg/beam/runners/direct/direct.go:65

	if !beam.Initialized() {
		log.Warn(ctx, "Beam has not been initialized. Call beam.Init() before pipeline construction.")
	}

	log.Info(ctx, "Pipeline:")
	log.Info(ctx, p)
	ctx = metrics.SetBundleID(ctx, "direct") // Ensure a metrics.Store exists.

	if *jobopts.Strict {
		log.Info(ctx, "Strict mode enabled, applying additional validation.")
		if _, err := vet.Execute(ctx, p); err != nil {
			return nil, errors.Wrap(err, "strictness check failed")
		}
		log.Info(ctx, "Strict mode validation passed.")
	}

	edges, _, err := p.Build()
	if err != nil {
		return nil, errors.Wrap(err, "invalid pipeline")
	}
	plan, err := Compile(edges)
	if err != nil {
		return nil, errors.Wrap(err, "translation failed")
	}
	beam.PipelineOptions.LoadOptionsFromFlags(nil)
	log.Info(ctx, plan)

	if err = plan.Execute(ctx, "", exec.DataContext{}); err != nil {
		plan.Down(ctx) // ignore any teardown errors
		return nil, err
	}
	if err = plan.Down(ctx); err != nil {
		return nil, err
	}

	return newDirectPipelineResult(ctx)
}

View on GitHub (pinned to 12126d8942)

Solutions

1. Inspect the inner error from p.Build() — it identifies the edge/node that failed.
2. Ensure each PCollection is consumed exactly once; fork it via beam.ParDo if branching is needed.
3. Check that all transforms are added to the same beam.Pipeline instance and inputs are non-nil.
4. If building programmatically, switch to beam.TryNewPipeline-style error-checked construction to catch the failure at the right place.

Example fix

// before
out1 := beam.ParDo(s, extractFn, col)
out2 := beam.ParDo(s, otherFn, col) // col consumed twice -> invalid pipeline
// after
out1 := beam.ParDo(s, extractFn, col)
out2 := beam.ParDo(s, otherFn, out1)

if _, err := p.Build(); err != nil {
    return fmt.Errorf("pipeline graph invalid: %w", err)
}

if _, err := beam.Run(ctx, runner, p); err != nil && strings.Contains(err.Error(), "invalid pipeline") {
    log.Printf("check PCollection reuse/nil inputs: %v", err)
}

When it happens

Trigger: Calling beam.Run (direct runner) on a pipeline whose graph is invalid: consuming a PCollection after it was already consumed, using a nil/zero PCollection as input, building transforms outside beam.TryNewPipeline scope, or misuse of composite transforms.

Common situations: Copy-pasted transform code reusing the same output twice; pipelines constructed with a pipeline value from a different package instance; panics recovered into errors inside custom graph manipulations; tests that partially build pipelines.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• strictness check failed
• translation failed
• unexpected edge: %v
• too few params
• too few inputs: forgot an input or to annotate options?

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