pipelinex

Documentation

Overview

Package pipelinex contains utilities for manipulating Beam proto pipelines. The utilities generally uses shallow copies and do not mutate their inputs.

Index

• Variables
• func ApplySdkImageOverrides(p *pipepb.Pipeline, patterns map[string]string) error
• func Bounded(p *pipepb.Pipeline) bool
• func ContainerImages(p *pipepb.Pipeline) []string
• func Normalize(p *pipepb.Pipeline) (*pipepb.Pipeline, error)
• func ShallowCloneFunctionSpec(p *pipepb.FunctionSpec) *pipepb.FunctionSpec
• func ShallowClonePTransform(t *pipepb.PTransform) *pipepb.PTransform
• func ShallowCloneParDoPayload(p *pipepb.ParDoPayload) *pipepb.ParDoPayload
• func ShallowCloneSideInput(p *pipepb.SideInput) *pipepb.SideInput
• func TopologicalSort(xforms map[string]*pipepb.PTransform, ids []string) []string
• func TrimCoders(coders map[string]*pipepb.Coder, ids ...string) map[string]*pipepb.Coder
• func Update(p *pipepb.Pipeline, values *pipepb.Components) (*pipepb.Pipeline, error)

Variables

var IdempotentNormalize bool = true

IdempotentNormalize determines whether to use the idempotent version of ensureUniqueNames or the legacy version. TODO(BEAM-12341): Cleanup once nothing depends on the legacy implementation.

Functions

func ApplySdkImageOverrides

func ApplySdkImageOverrides(p *pipepb.Pipeline, patterns map[string]string) error

ApplySdkImageOverrides takes a pipeline and a map of patterns to overrides, and proceeds to replace matching ContainerImages in any Environments present in the pipeline. Each environment is expected to match at most one pattern. If an environment matches two or more it is arbitrary which pattern will be applied.

func Bounded

func Bounded(p *pipepb.Pipeline) bool

Bounded returns true iff all PCollections are bounded.

func ContainerImages

func ContainerImages(p *pipepb.Pipeline) []string

ContainerImages returns the set of container images used in the given pipeline.

func Normalize

func Normalize(p *pipepb.Pipeline) (*pipepb.Pipeline, error)

Normalize recomputes derivative information in the pipeline, such as roots and input/output for composite transforms. It also ensures that unique names are so and topologically sorts each subtransform list.

func ShallowCloneFunctionSpec

func ShallowCloneFunctionSpec(p *pipepb.FunctionSpec) *pipepb.FunctionSpec

ShallowCloneFunctionSpec makes a shallow copy of the given FunctionSpec.

func ShallowClonePTransform

func ShallowClonePTransform(t *pipepb.PTransform) *pipepb.PTransform

ShallowClonePTransform makes a shallow copy of the given PTransform.

func ShallowCloneParDoPayload

func ShallowCloneParDoPayload(p *pipepb.ParDoPayload) *pipepb.ParDoPayload

ShallowCloneParDoPayload makes a shallow copy of the given ParDoPayload.

func ShallowCloneSideInput

func ShallowCloneSideInput(p *pipepb.SideInput) *pipepb.SideInput

ShallowCloneSideInput makes a shallow copy of the given SideInput.

func TopologicalSort

func TopologicalSort(xforms map[string]*pipepb.PTransform, ids []string) []string

TopologicalSort returns a topologically sorted list of the given ids, generally from the same scope/composite. Assumes acyclic graph.

func TrimCoders

func TrimCoders(coders map[string]*pipepb.Coder, ids ...string) map[string]*pipepb.Coder

TrimCoders returns the transitive closure of the given coders ids.

func Update

func Update(p *pipepb.Pipeline, values *pipepb.Components) (*pipepb.Pipeline, error)

Update merges a pipeline with the given components, which may add, replace or delete its values. It returns the merged pipeline. The input is not modified.