func CrossLanguagePayload(pl any) []byte

func NewPipelineWithRoot() (*Pipeline, Scope)

func ParDo0(s Scope, dofn any, col PCollection, opts ...Option)

func ParDo2(s Scope, dofn any, col PCollection, opts ...Option) (PCollection, PCollection)

func ParDo3(s Scope, dofn any, col PCollection, opts ...Option) (PCollection, PCollection, PCollection)

func ParDo4(s Scope, dofn any, col PCollection, opts ...Option) (PCollection, PCollection, PCollection, PCollection)

func RegisterDoFn(dofn any)

func RegisterFunction(fn any)

func RegisterInit(hook func())

func UnnamedOutputTag() string

DecodeCoder decodes a coder. Any custom coder function symbol must be resolvable via the runtime.GlobalSymbolResolver. The types must be encodable.

func NewCoder(t FullType) Coder

NewCoder infers a Coder for any bound full type.

func (c Coder) IsValid() bool

IsValid returns true iff the Coder is valid. Any use of an invalid Coder will result in a panic.

func (c Coder) Type() FullType

Type returns the full type 'A' of elements the coder can encode and decode. 'A' must be a concrete full type, such as int or KV<int,string>.

func NewCounter(namespace, name string) Counter

NewCounter returns the Counter with the given namespace and name.

Dec decrements the counter within by the given amount. The context must be provided by the framework, or the value will not be recorded.

package main

import (
	"context"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

// A beam_test global context var to improve how the examples look.
var ctx = context.Background()

func main() {
	c := beam.NewCounter("example", "size")
	c.Dec(ctx, int64(len("foobar")))
}

Inc increments the counter within by the given amount. The context must be provided by the framework, or the value will not be recorded.

package main

import (
	"context"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

// A beam_test global context var to improve how the examples look.
var ctx = context.Background()

func main() {
	c := beam.NewCounter("example", "size")
	c.Inc(ctx, int64(len("foobar")))
}

func NewDistribution(namespace, name string) Distribution

NewDistribution returns the Distribution with the given namespace and name.

Update adds an observation to this distribution. The context must be provided by the framework, or the value will not be recorded.

package main

import (
	"context"
	"time"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

// A beam_test global context var to improve how the examples look.
var ctx = context.Background()

func main() {
	t := time.Millisecond * 42
	d := beam.NewDistribution("example", "latency_micros")
	d.Update(ctx, int64(t/time.Microsecond))
}

NewElementDecoder returns an ElementDecoder the given type.

NewElementEncoder returns a new encoding function for the given type.

type EncodedCoder struct {
	
	Coder Coder
}

MarshalJSON returns the JSON encoding this value.

UnmarshalJSON sets the state of this instance from the passed in JSON.

MarshalJSON returns the JSON encoding this value.

UnmarshalJSON sets the state of this instance from the passed in JSON.

MarshalJSON returns the JSON encoding this value.

UnmarshalJSON sets the state of this instance from the passed in JSON.

func NewGauge(namespace, name string) Gauge

NewGauge returns the Gauge with the given namespace and name.

Set sets the current value for this gauge. The context must be provided by the framework, or the value will not be recorded.

package main

import (
	"context"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

// A beam_test global context var to improve how the examples look.
var ctx = context.Background()

func main() {
	g := beam.NewGauge("example", "progress")
	g.Set(ctx, 42)
}

type Option interface {
	
}

type PCollection struct {
	
}

func AddFixedKey(s Scope, col PCollection) PCollection

AddFixedKey adds a fixed key (0) to every element.

func CoGroupByKey(s Scope, cols ...PCollection) PCollection

CoGroupByKey inserts a CoGBK transform into the pipeline.

func Combine(s Scope, combinefn any, col PCollection, opts ...Option) PCollection

Combine inserts a global Combine transform into the pipeline. It expects a PCollection<T> as input where T is a concrete type. Combine supports TypeDefinition options for binding generic types in combinefn.

func CombinePerKey(s Scope, combinefn any, col PCollection, opts ...Option) PCollection

CombinePerKey inserts a GBK and per-key Combine transform into the pipeline. It expects a PCollection<KV<K,T>>. The CombineFn may optionally take a key parameter. CombinePerKey supports TypeDefinition options for binding generic types in combinefn.

func Create(s Scope, values ...any) PCollection

Create inserts a fixed non-empty set of values into the pipeline. The values must be of the same type 'A' and the returned PCollection is of type A.

The returned PCollections can be used as any other PCollections. The values are JSON-coded. Each runner may place limits on the sizes of the values and Create should generally only be used for small collections.

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

var s = beam.Scope{}

func main() {
	beam.Create(s, 5, 6, 7, 8, 9)               // PCollection<int>
	beam.Create(s, []int{5, 6}, []int{7, 8, 9}) // PCollection<[]int>
	beam.Create(s, []int{5, 6, 7, 8, 9})        // PCollection<[]int>
	beam.Create(s, "a", "b", "c")               // PCollection<string>
}

func CreateList(s Scope, list any) PCollection

CreateList inserts a fixed set of values into the pipeline from a slice or array. Unlike Create this supports the creation of an empty PCollection.

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

var s = beam.Scope{}

func main() {
	beam.CreateList(s, []int{5, 6, 7, 8, 9}) // PCollection<int>
}

func DropKey(s Scope, col PCollection) PCollection

DropKey drops the key for an input PCollection<KV<A,B>>. It returns a PCollection<B>.

func DropValue(s Scope, col PCollection) PCollection

DropValue drops the value for an input PCollection<KV<A,B>>. It returns a PCollection<A>.

func Explode(s Scope, col PCollection) PCollection

Explode is a PTransform that takes a single PCollection<[]A> and returns a PCollection<A> containing all the elements for each incoming slice.

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

var s = beam.Scope{}

func main() {
	d := beam.Create(s, []int{1, 2, 3, 4, 5}) // PCollection<[]int>
	beam.Explode(s, d)                        // PCollection<int>
}

External defines a Beam external transform. The interpretation of this primitive is runner specific. The runner is responsible for parsing the payload based on the URN provided to implement the behavior of the operation. Transform libraries should expose an API that captures the user's intent and serialize the payload as a byte slice that the runner will deserialize.

Use ExternalTagged if the runner will need to associate the PTransforms local PCollection tags with values in the payload.

func Flatten(s Scope, cols ...PCollection) PCollection

Flatten is a PTransform that takes either multiple PCollections of type 'A' and returns a single PCollection of type 'A' containing all the elements in all the input PCollections. The name "Flatten" suggests taking a list of lists and flattening them into a single list.

By default, the Coder of the output PCollection is the same as the Coder of the first PCollection.

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/textio"
)

var s = beam.Scope{}

func main() {
	a := textio.Read(s, "...some file path...") // PCollection<string>
	b := textio.Read(s, "...some other file path...")
	c := textio.Read(s, "...some third file path...")

	beam.Flatten(s, a, b, c) // PCollection<String>
}

func GroupByKey(s Scope, a PCollection) PCollection

GroupByKey is a PTransform that takes a PCollection of type KV<A,B>, groups the values by key and windows, and returns a PCollection of type GBK<A,B> representing a map from each distinct key and window of the input PCollection to an iterable over all the values associated with that key in the input per window. Each key in the output PCollection is unique within each window.

GroupByKey is analogous to converting a multi-map into a uni-map, and related to GROUP BY in SQL. It corresponds to the "shuffle" step between the Mapper and the Reducer in the MapReduce framework.

Two keys of type A are compared for equality by first encoding each of the keys using the Coder of the keys of the input PCollection, and then comparing the encoded bytes. This admits efficient parallel evaluation. Note that this requires that the Coder of the keys be deterministic.

By default, input and output PCollections share a key Coder and iterable values in the input and output PCollection share an element Coder.

GroupByKey is a key primitive in data-parallel processing, since it is the main way to efficiently bring associated data together into one location. It is also a key determiner of the performance of a data-parallel pipeline.

See CoGroupByKey for a way to group multiple input PCollections by a common key at once.

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

var s = beam.Scope{}

func main() {
	type Doc struct{}
	var urlDocPairs beam.PCollection             // PCollection<KV<string, Doc>>
	urlToDocs := beam.GroupByKey(s, urlDocPairs) // PCollection<CoGBK<string, Doc>>

	// CoGBK parameters receive an iterator function with all values associated
	// with the same key.
	beam.ParDo0(s, func(key string, values func(*Doc) bool) {
		var cur Doc
		for values(&cur) {
			// ... process all docs having that url ...
		}
	}, urlToDocs) // PCollection<KV<string, []Doc>>
}

func Impulse(s Scope) PCollection

Impulse emits a single empty []byte into the global window. The resulting PCollection is a singleton of type []byte.

The purpose of Impulse is to trigger another transform, such as ones that take all information as side inputs.

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

var s = beam.Scope{}

func main() {
	beam.Impulse(s) // PCollection<[]byte>
}

func ImpulseValue(s Scope, value []byte) PCollection

ImpulseValue emits the supplied byte slice into the global window. The resulting PCollection is a singleton of type []byte.

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

var s = beam.Scope{}

func main() {
	beam.ImpulseValue(s, []byte{}) // PCollection<[]byte>
}

Must returns the input, but panics if err != nil.

MustN returns the input, but panics if err != nil.

func ParDo(s Scope, dofn any, col PCollection, opts ...Option) PCollection

ParDo is the core element-wise PTransform in Apache Beam, invoking a user-specified function on each of the elements of the input PCollection to produce zero or more output elements, all of which are collected into the output PCollection. Use one of the ParDo variants for a different number of output PCollections. The PCollections do not need to have the same types.

Elements are processed independently, and possibly in parallel across distributed cloud resources. The ParDo processing style is similar to what happens inside the "Mapper" or "Reducer" class of a MapReduce-style algorithm.

DoFns ¶

The function to use to process each element is specified by a DoFn, either as single function or as a struct with methods, notably ProcessElement. The struct may also define Setup, StartBundle, FinishBundle and Teardown methods. The struct is JSON-serialized and may contain construction-time values.

Functions and types used as DoFns must be registered with beam using the beam `register` package, so they may execute on distributed workers. Functions must not be anonymous or closures, or they will fail at execution time.

Conceptually, when a ParDo transform is executed, the elements of the input PCollection are first divided up into some number of "bundles". These are farmed off to distributed worker machines (or locally on a local runner instance). For each bundle of input elements processing proceeds as follows:

• If a struct, a fresh instance of the argument DoFn is created on a worker from json serialization, and the Setup method is called on this instance, if present. A runner may reuse DoFn instances for multiple bundles. A DoFn that has terminated abnormally (by returning an error) will never be reused.
• The DoFn's StartBundle method, if provided, is called to initialize it.
• The DoFn's ProcessElement method is called on each of the input elements in the bundle.
• The DoFn's FinishBundle method, if provided, is called to complete its work. After FinishBundle is called, the framework will not again invoke ProcessElement or FinishBundle until a new call to StartBundle has occurred.
• If any of Setup, StartBundle, ProcessElement or FinishBundle methods return an error, the Teardown method, if provided, will be called on the DoFn instance.
• If a runner will no longer use a DoFn, the Teardown method, if provided, will be called on the discarded instance.

Each of the calls to any of the DoFn's processing methods can produce zero or more output elements. All of the of output elements from all of the DoFn instances are included in an output PCollection.

For example:

func stringLen(word string) int { return len(word)	}
func init() { register.Function1x1(stringLen) }

words := beam.ParDo(s, &Foo{...}, ...)
lengths := beam.ParDo(s, stringLen, words)

Each output element has the same timestamp and is in the same windows as its corresponding input element. The timestamp can be accessed and/or emitted by including a EventTime-typed parameter. The name of the function or struct is used as the DoFn name. Function literals do not have stable names and should thus not be used in production code.

Side Inputs ¶

While a ParDo processes elements from a single "main input" PCollection, it can take additional "side input" PCollections. These SideInput along with the DoFn parameter form express styles of accessing PCollection computed by earlier pipeline operations, passed in to the ParDo transform using SideInput options, and their contents accessible to each of the DoFn operations. For example:

func filterLessThanCutoff(word string, cutoff int, emit func(string)) {
	if len(word) < cutoff {
		emit(word)
	}
}
func init() { register.Function3x0(filterLessThanCutoff) }

words := ...
cufoff := ...  // Singleton PCollection<int>
smallWords := beam.ParDo(s, filterLessThanCutoff, words, beam.SideInput{Input: cutoff})

Additional Outputs ¶

Optionally, a ParDo transform can produce zero or multiple output PCollections. Note the use of ParDo2 to specfic 2 outputs. For example:

func partitionAtCutoff(word string, cutoff int, small, big func(string)) {
	if len(word) < cutoff {
		small(word)
	} else {
		big(word)
	}
}
func init() { register.Function4x0(partitionAtCutoff) }

words := ...
cufoff := ...  // Singleton PCollection<int>
small, big := beam.ParDo2(s, partitionAtCutoff, words, beam.SideInput{Input: cutoff})

By default, the Coders for the elements of each output PCollections is inferred from the concrete type.

No Global Shared State ¶

There are three main ways to initialize the state of a DoFn instance processing a bundle:

• Define public instance variable state. This state will be automatically JSON serialized and then deserialized in the DoFn instances created for bundles. This method is good for state known when the original DoFn is created in the main program, if it's not overly large. This is not suitable for any state which must only be used for a single bundle, as DoFn's may be used to process multiple bundles.

• Compute the state as a singleton PCollection and pass it in as a side input to the DoFn. This is good if the state needs to be computed by the pipeline, or if the state is very large and so is best read from file(s) rather than sent as part of the DoFn's serialized state.

• Initialize the state in each DoFn instance, in a StartBundle method. This is good if the initialization doesn't depend on any information known only by the main program or computed by earlier pipeline operations, but is the same for all instances of this DoFn for all program executions, say setting up empty caches or initializing constant data.

ParDo operations are intended to be able to run in parallel across multiple worker machines. This precludes easy sharing and updating mutable state across those machines. There is no support in the Beam model for communicating and synchronizing updates to shared state across worker machines, so programs should not access any mutable global variable state in their DoFn, without understanding that the Go processes for the main program and workers will each have its own independent copy of such state, and there won't be any automatic copying of that state across Java processes. All information should be communicated to DoFn instances via main and side inputs and serialized state, and all output should be communicated from a DoFn instance via output PCollections, in the absence of external communication mechanisms written by user code.

Splittable DoFns ¶

Splittable DoFns are DoFns that are able to split work within an element, as opposed to only at element boundaries like normal DoFns. This is useful for DoFns that emit many outputs per input element and can distribute that work among multiple workers. The most common examples of this are sources.

In order to split work within an element, splittable DoFns use the concept of restrictions, which are objects that are associated with an element and describe a portion of work on that element. For example, a restriction associated with a filename might describe what byte range within that file to process. In addition to restrictions, splittable DoFns also rely on restriction trackers to track progress and perform splits on a restriction currently being processed. See the `RTracker` interface in core/sdf/sdf.go for more details.

Splitting ¶

Splitting means taking one restriction and splitting into two or more that cover the entire input space of the original one. In other words, processing all the split restrictions should produce identical output to processing the original one.

Splitting occurs in two stages. The initial splitting occurs before any restrictions have started processing. This step is used to split large restrictions into smaller ones that can then be distributed among multiple workers for processing. Initial splitting is user-defined and optional.

Dynamic splitting occurs during the processing of a restriction in runners that have implemented it. If there are available workers, runners may split the unprocessed portion of work from a busy worker and shard it to available workers in order to better distribute work. With unsplittable DoFns this can only occur on element boundaries, but for splittable DoFns this split can land within a restriction and will require splitting that restriction.

• Note: Dataflow is currently the only runner with support for both initial and dynamic splitting. Other runners do not support dynamic splitting, and stragglers will therefore not be split during execution with liquid sharding.

Splittable DoFn Methods ¶

Making a splittable DoFn requires the following methods to be implemented on a DoFn in addition to the usual DoFn requirements. In the following method signatures `elem` represents the main input elements to the DoFn, and should match the types used in ProcessElement. `restriction` represents the user-defined restriction, and can be any type as long as it is consistent throughout all the splittable DoFn methods:

• `CreateInitialRestriction(context.Context?, elem) (restriction, error?)` CreateInitialRestriction creates an initial restriction encompassing an entire element. The restriction created stays associated with the element it describes.
• `SplitRestriction(context.Context?, elem, restriction) ([]restriction, error?)` SplitRestriction takes an element and its initial restriction, and optionally performs an initial split on it, returning a slice of all the split restrictions. If no splits are desired, the method returns a slice containing only the original restriction. This method will always be called on each newly created restriction before they are processed.
• `RestrictionSize(context.Context?, elem, restriction) (float64, error?)` RestrictionSize returns a cheap size estimation for a restriction. This size is an abstract non-negative scalar value that represents how much work a restriction takes compared to other restrictions in the same DoFn. For example, a size of 200 represents twice as much work as a size of 100, but the numbers do not represent anything on their own. Size is used by runners to estimate work for dynamic work rebalancing. Must be thread safe. Will be invoked concurrently during bundle processing due to runner initiated splitting and progress estimation.
• `CreateTracker(context.Context?, restriction) (restrictionTracker, error?)` CreateTracker creates and returns a restriction tracker (a concrete type implementing the `sdf.RTracker` interface) given a restriction. The restriction tracker is used to track progress processing a restriction, and to allow for dynamic splits. This method is called on each restriction right before processing begins.
• `ProcessElement(context.Context?, sdf.RTracker, elem, func emit(output)) (sdf.ProcessContinuation?, error?)` For splittable DoFns, ProcessElement requires a restriction tracker before inputs, and generally requires emits to be used for outputs, since restrictions will generally produce multiple outputs. For more details on processing restrictions in a splittable DoFn, see `sdf.RTracker`. ProcessElement can optionally return a `sdf.ProcessContinuation` to signal to the runner that processing should be resumed at a later time, if not all data within the restriction can be processed within the lifetime of a single bundle. The runner tries to respect the resume time, however it is not guaranteed.

A splittable DoFn can also implement the following optional method:

• `TruncateRestriction(context.Context?, sdf.RTracker, elem) (restriction, error?)` TruncateRestriction is triggered when a pipeline starts to drain on runners that support pipeline draining. It helps finish the pipeline faster by truncating the restriction. If not implemented, the default behavior for bounded restrictions is to process the remainder of the restriction, and for unbounded restrictions to process until the next SDF-initiated checkpoint or runner-initiated split occurs.

Fault Tolerance ¶

In a distributed system, things can fail: machines can crash, machines can be unable to communicate across the network, etc. While individual failures are rare, the larger the job, the greater the chance that something, somewhere, will fail. Beam runners may strive to mask such failures by retrying failed DoFn bundles. This means that a DoFn instance might process a bundle partially, then crash for some reason, then be rerun (often as a new process) on that same bundle and on the same elements as before. Sometimes two or more DoFn instances will be running on the same bundle simultaneously, with the system taking the results of the first instance to complete successfully. Consequently, the code in a DoFn needs to be written such that these duplicate (sequential or concurrent) executions do not cause problems. If the outputs of a DoFn are a pure function of its inputs, then this requirement is satisfied. However, if a DoFn's execution has external side-effects, such as performing updates to external HTTP services, then the DoFn's code needs to take care to ensure that those updates are idempotent and that concurrent updates are acceptable. This property can be difficult to achieve, so it is advisable to strive to keep DoFns as pure functions as much as possible.

Optimization ¶

Beam runners may choose to apply optimizations to a pipeline before it is executed. A key optimization, fusion, relates to ParDo operations. If one ParDo operation produces a PCollection that is then consumed as the main input of another ParDo operation, the two ParDo operations will be fused together into a single ParDo operation and run in a single pass; this is "producer-consumer fusion". Similarly, if two or more ParDo operations have the same PCollection main input, they will be fused into a single ParDo that makes just one pass over the input PCollection; this is "sibling fusion".

If after fusion there are no more unfused references to a PCollection (e.g., one between a producer ParDo and a consumer ParDo), the PCollection itself is "fused away" and won't ever be written to disk, saving all the I/O and space expense of constructing it.

When Beam runners apply fusion optimization, it is essentially "free" to write ParDo operations in a very modular, composable style, each ParDo operation doing one clear task, and stringing together sequences of ParDo operations to get the desired overall effect. Such programs can be easier to understand, easier to unit-test, easier to extend and evolve, and easier to reuse in new programs. The predefined library of PTransforms that come with Beam makes heavy use of this modular, composable style, trusting to the runner to "flatten out" all the compositions into highly optimized stages.

See https://beam.apache.org/documentation/programming-guide/#pardo for the web documentation for ParDo

package main

import (
	"github.com/apache/beam/sdks/v2/go/pkg/beam"
)

var s = beam.Scope{}

func main() {
	var words beam.PCollection  // PCollection<string>
	var cutoff beam.PCollection // Singleton PCollection<int>
	small, big := beam.ParDo2(s, func(word string, cutoff int, small, big func(string)) {
		if len(word) < cutoff {
			small(word)
		} else {
			big(word)
		}
	}, words, beam.SideInput{Input: cutoff})

	_, _ = small, big
}

func ParDoN(s Scope, dofn any, col PCollection, opts ...Option) []PCollection

ParDoN inserts a ParDo with any number of outputs into the pipeline.

Partition takes a PCollection<T> and a PartitionFn, uses the PartitionFn to split the elements of the input PCollection into N partitions, and returns a []PCollection<T> that bundles N PCollection<T>s containing the split elements.

A PartitionFn has the signature `func(T) int.`

func lenToTen(s string) int {
	if len(s) > 9 {
		return 10
	}
	return len(s)
}

// Partition functions must be registered with Beam, and must not be closures.
func init() { register.Function1x1(lenToTen) }

// The number of partitions goes up to 11 since we can return 0 through 10
wordsByLength := beam.Partition(s, 11, lenToTen, inputStrings)

T is permitted to be a KV.

func Reshuffle(s Scope, col PCollection) PCollection

Reshuffle copies a PCollection of the same kind and using the same element coder, and maintains the same windowing information. Importantly, it allows the result PCollection to be processed with a different sharding, in a different stage than the input PCollection.

For example, if a computation needs a lot of parallelism but produces only a small amount of output data, then the computation producing the data can run with as much parallelism as needed, while the output file is written with a smaller amount of parallelism, using the following pattern:

pc := bigHairyComputationNeedingParallelism(scope) // PCollection<string>
resharded := beam.Reshuffle(scope, pc)                // PCollection<string>

Another use case is when one has a non-deterministic DoFn followed by one that performs externally-visible side effects. Inserting a Reshuffle between these DoFns ensures that retries of the second DoFn will always be the same, which is necessary to make side effects idempotent.

A Reshuffle will force a break in the optimized pipeline. Consequently, this operation should be used sparingly, only after determining that the pipeline without reshuffling is broken in some way and performing an extra operation is worth the cost.

func Seq(s Scope, col PCollection, dofns ...any) PCollection

Seq is a convenience helper to chain single-input/single-output ParDos together in a sequence.

package main

import (
	"math"
	"strconv"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/textio"
)

var s = beam.Scope{}

func main() {
	a := textio.Read(s, "...some file path...") // PCollection<string>

	beam.Seq(s, a,
		strconv.Atoi, // string to int
		func(i int) float64 { return float64(i) }, // int to float64
		math.Signbit, // float64 to bool
	) // PCollection<bool>
}

func SwapKV(s Scope, col PCollection) PCollection

SwapKV swaps the key and value for an input PCollection<KV<A,B>>. It returns a PCollection<KV<B,A>>.

func TryCoGroupByKey(s Scope, cols ...PCollection) (PCollection, error)

TryCoGroupByKey inserts a CoGBK transform into the pipeline. Returns an error on failure.

func TryCombine(s Scope, combinefn any, col PCollection, opts ...Option) (PCollection, error)

TryCombine attempts to insert a global Combine transform into the pipeline. It may fail for multiple reasons, notably that the combinefn is not valid or cannot be bound -- due to type mismatch, say -- to the incoming PCollections.

func TryCombinePerKey(s Scope, combinefn any, col PCollection, opts ...Option) (PCollection, error)

TryCombinePerKey attempts to insert a per-key Combine transform into the pipeline. It may fail for multiple reasons, notably that the combinefn is not valid or cannot be bound -- due to type mismatch, say -- to the incoming PCollection.

TryCreate inserts a fixed non-empty set of values into the pipeline. The values must be of the same type.

TryCreateList inserts a fixed set of values into the pipeline from a slice or array. The values must be of the same type. Unlike TryCreate this supports the creation of an empty PCollection.

TryExternal attempts to perform the work of External, returning an error indicating why the operation failed.

func TryFlatten(s Scope, cols ...PCollection) (PCollection, error)

TryFlatten merges incoming PCollections of type 'A' to a single PCollection of type 'A'. Returns an error indicating the set of PCollections that could not be flattened.

TryGroupByKey inserts a GBK transform into the pipeline. Returns an error on failure.

TryParDo attempts to insert a ParDo transform into the pipeline. It may fail for multiple reasons, notably that the dofn is not valid or cannot be bound -- due to type mismatch, say -- to the incoming PCollections.

TryReshuffle inserts a Reshuffle into the pipeline, and returns an error if the pcollection's unable to be reshuffled.

TryWindowInto attempts to insert a WindowInto transform.

WindowInto applies the windowing strategy to each element.

func (p PCollection) Coder() Coder

Coder returns the coder for the collection. The Coder is of type 'A'.

IsValid returns true iff the PCollection is valid and part of a Pipeline. Any use of an invalid PCollection will result in a panic.

SetCoder set the coder for the collection. The Coder must be of type 'A'.

func (p PCollection) Type() FullType

Type returns the full type 'A' of the elements. 'A' must be a concrete type, such as int or KV<int,string>.

func NewPipeline() *Pipeline

NewPipeline creates a new empty pipeline.

Build validates the Pipeline and returns a lower-level representation for execution. It is called by runners only.

func (p *Pipeline) Root() Scope

Root returns the root scope of the pipeline.

Run executes the pipeline using the selected registred runner. It is customary to define a "runner" with no default as a flag to let users control runner selection.

func (s Scope) IsValid() bool

IsValid returns true iff the Scope is valid. Any use of an invalid Scope will result in a panic.

Scope returns a sub-scope with the given name. The name provided may be augmented to ensure uniqueness.

WithContext creates a named subscope with an attached context for the represented composite transform. Values from that context may be extracted and added to the composite PTransform or generate a new environment for scoped transforms.

If you're not sure whether these apply to your transform, use Scope instead, and do not set a context.

type SideInput struct {
	Input PCollection
}

type WindowIntoOption interface {
	
}

AllowedLateness configures for how long data may arrive after the end of a window.

func PanesAccumulate() WindowIntoOption

PanesAccumulate applies an Accumulating AccumulationMode to the window. After a pane fires, already processed elements will accumulate and elements will be repeated in subseqent firings for the window.

func PanesDiscard() WindowIntoOption

PanesDiscard applies a Discarding AccumulationMode to the window. After a pane fires, already processed elements will be discarded and not included in later firings for the window.

Trigger applies the given trigger to the window.