beam

package

v2.71.0Latest Latest

This package is not in the latest version of its module.

Go to latest Published: Jan 16, 2026 License:Apache-2.0, BSD-3-Clause, MITImports:31Imported by:905

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/apache/beam

Documentation ¶

Overview¶

Package beam is an implementation of the Apache Beam (https://beam.apache.org)programming model in Go. Beam provides a simple, powerful model forbuilding both batch and streaming parallel data processing pipelines.

For more on the Beam model see:https://beam.apache.org/documentation/programming-guide

For design choices this implementation makes see:https://s.apache.org/beam-go-sdk-design-rfc

Example (GettingStarted)¶

package mainimport ("context""fmt""regexp""strings""github.com/apache/beam/sdks/v2/go/pkg/beam""github.com/apache/beam/sdks/v2/go/pkg/beam/io/textio""github.com/apache/beam/sdks/v2/go/pkg/beam/runners/direct")func main() {// In order to start creating the pipeline for execution, a Pipeline object is needed.p := beam.NewPipeline()s := p.Root()// The pipeline object encapsulates all the data and steps in your processing task.// It is the basis for creating the pipeline's data sets as PCollections and its operations// as transforms.// The PCollection abstraction represents a potentially distributed,// multi-element data set. You can think of a PCollection as “pipeline” data;// Beam transforms use PCollection objects as inputs and outputs. As such, if// you want to work with data in your pipeline, it must be in the form of a// PCollection.// Transformations are applied in a scoped fashion to the pipeline. The scope// can be obtained from the pipeline object.// Start by reading text from an input files, and receiving a PCollection.lines := textio.Read(s, "protocol://path/file*.txt")// Transforms are added to the pipeline so they are part of the work to be// executed.  Since this transform has no PCollection as an input, it is// considered a 'root transform'// A pipeline can have multiple root transformsmoreLines := textio.Read(s, "protocol://other/path/file*.txt")// Further transforms can be applied, creating an arbitrary, acyclic graph.// Subsequent transforms (and the intermediate PCollections they produce) are// attached to the same pipeline.all := beam.Flatten(s, lines, moreLines)wordRegexp := regexp.MustCompile(`[a-zA-Z]+('[a-z])?`)words := beam.ParDo(s, func(line string, emit func(string)) {for _, word := range wordRegexp.FindAllString(line, -1) {emit(word)}}, all)formatted := beam.ParDo(s, strings.ToUpper, words)textio.Write(s, "protocol://output/path", formatted)// Applying a transform adds it to the pipeline, rather than executing it// immediately.  Once the whole pipeline of transforms is constructed, the// pipeline can be executed by a PipelineRunner.  The direct runner executes the// transforms directly, sequentially, in this one process, which is useful for// unit tests and simple experiments:if _, err := direct.Execute(context.Background(), p); err != nil {fmt.Printf("Pipeline failed: %v", err)}}

Example (MetricsDeclaredAnywhere)¶

package mainimport ("context""regexp""github.com/apache/beam/sdks/v2/go/pkg/beam""github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics")func ctxWithPtransformID(id string) context.Context {ctx := context.Background()ctx = metrics.SetBundleID(ctx, "exampleBundle")ctx = metrics.SetPTransformID(ctx, id)return ctx}var wordRE = regexp.MustCompile(`[a-zA-Z]+('[a-z])?`)func main() {// Metrics can be declared outside DoFns, and used inside..outside := beam.NewCounter("example.namespace", "count")extractWordsDofn := func(ctx context.Context, line string, emit func(string)) {// They can be defined at time of use within a DoFn, if necessary.inside := beam.NewDistribution("example.namespace", "characters")for _, word := range wordRE.FindAllString(line, -1) {emit(word)outside.Inc(ctx, 1)inside.Update(ctx, int64(len(word)))}}ctx := ctxWithPtransformID("example")extractWordsDofn(ctx, "this has six words in it", func(string) {})extractWordsDofn(ctx, "this has seven words in it, see?", func(string) {})metrics.DumpToOutFromContext(ctx)}

Output:PTransformID: "example"example.namespace.characters - count: 13 sum: 43 min: 2 max: 5example.namespace.count - value: 13

Example (MetricsReusable)¶

package mainimport ("context""regexp""github.com/apache/beam/sdks/v2/go/pkg/beam""github.com/apache/beam/sdks/v2/go/pkg/beam/core/metrics")// A beam_test global context var to improve how the examples look.var ctx = context.Background()var wordRE = regexp.MustCompile(`[a-zA-Z]+('[a-z])?`)func main() {// Metric proxies can be used in multiple DoFnsc := beam.NewCounter("example.reusable", "count")extractWordsDofn := func(ctx context.Context, line string, emit func(string)) {for _, word := range wordRE.FindAllString(line, -1) {emit(word)c.Inc(ctx, 1)}}extractRunesDofn := func(ctx context.Context, line string, emit func(rune)) {for _, r := range line {emit(r)c.Inc(ctx, 1)}}ctx = metrics.SetBundleID(ctx, "exampleBundle")extractWordsDofn(metrics.SetPTransformID(ctx, "extract1"), "this has six words in it", func(string) {})extractRunesDofn(metrics.SetPTransformID(ctx, "extract2"), "seven thousand", func(rune) {})metrics.DumpToOutFromContext(ctx)}

Output:PTransformID: "extract1"example.reusable.count - value: 6PTransformID: "extract2"example.reusable.count - value: 14

Index¶

Examples¶

Constants¶

This section is empty.

Variables¶

View Source

var (TType =typex.TTypeUType =typex.UTypeVType =typex.VTypeWType =typex.WTypeXType =typex.XTypeYType =typex.YTypeZType =typex.ZType)

These are the reflect.Type instances of the universal types, which are usedwhen binding actual types to "generic" DoFns that use Universal Types.

View Source

var EnableSchemasbool =true

EnableSchemas is a temporary configuration variableto use Beam Schema encoding by default instead of JSON.Before it is removed, it will be set to true by defaultand then eventually removed.

Only users who rely on default JSON marshalling behaviour should setthis explicitly, and file an issue on the BEAM repo so the issue maybe resolved.https://github.com/apache/beam/issues/new/choose

View Source

var EventTimeType =typex.EventTimeType

EventTimeType is the reflect.Type of EventTime.

View Source

var PipelineOptions =runtime.GlobalOptions

PipelineOptions are global options for the active pipeline. Options canbe defined any time before execution and are re-created by the harness onremote execution workers. Global options should be used sparingly.

Functions¶

funcCrossLanguage ¶

func CrossLanguage(sScope,urnstring,payload []byte,expansionAddrstring,namedInputs map[string]PCollection,namedOutputTypes map[string]FullType,) map[string]PCollection

CrossLanguage is a low-level transform for executing cross-language transforms written in otherSDKs. Because this is low-level, it is recommended to use one of the higher-level IO-specificwrappers where available. These can be found in the pkg/beam/io/xlang subdirectory.CrossLanguage is useful for executing cross-language transforms which do not have any existingIO wrappers.

Usage requires an address for an expansion service accessible during pipeline construction, aURN identifying the desired transform, an optional payload with configuration information, andinput and output names. It outputs a map of named output PCollections.

For more information on expansion services and other aspects of cross-language transforms ingeneral, refer to the Beam programming guide:https://beam.apache.org/documentation/programming-guide/#multi-language-pipelines

Payload¶

Payloads are configuration data that some cross-language transforms require for expansion.Consult the documentation of the transform in the source SDK to find out what payload data itrequires. If no payload is required, pass in nil.

CrossLanguage accepts payloads as a []byte containing an encoded ExternalConfigurationPayloadprotobuf. The helper function beam.CrossLanguagePayload is the recommended way to easily encodea standard Go struct for use as a payload.

Inputs and Outputs¶

Like most transforms, any input PCollections must be provided. Unlike most transforms, outputtypes must be provided because Go cannot infer output types from external transforms.

Inputs and outputs to a cross-language transform may be either named or unnamed. Namedinputs/outputs are used when there are more than one input/output, and are provided as maps withnames as keys. Unnamed inputs/outputs are used when there is only one, and a map can be quicklyconstructed with the UnnamedInput and UnnamedOutput methods.

An example of defining named inputs and outputs:

namedInputs := map[string]beam.PCollection{"pcol1": pcol1, "pcol2": pcol2}namedOutputTypes := map[string]typex.FullType{    "main": typex.New(reflectx.String),    "side": typex.New(reflectx.Int64),}

CrossLanguage outputs a map of PCollections with associated names. These names will match thosefrom provided named outputs. If the beam.UnnamedOutput method was used, the PCollection can beretrieved with beam.UnnamedOutputTag().

An example of retrieving named outputs from a call to CrossLanguage:

outputs := beam.CrossLanguage(...)mainPcol := outputs["main"]sidePcol := outputs["side"]

Example¶

This example shows using CrossLanguage to execute the Prefix cross-language transform using anexpansion service running on localhost:8099. Prefix requires a payload containing a prefix toprepend to every input string.

type prefixPayload struct {    Data string}encodedPl := beam.CrossLanguagePayload(prefixPayload{Data: "foo"})urn := "beam:transforms:xlang:test:prefix"expansionAddr := "localhost:8099"outputType := beam.UnnamedOutput(typex.New(reflectx.String))input := beam.UnnamedInput(inputPcol)outs := beam.CrossLanguage(s, urn, encodedPl, expansionAddr, input, outputType)outPcol := outputs[beam.UnnamedOutputTag()]

Alternative Expansion Handlers¶

The xlangx.RegisterHandler function can be used to register alternative expansionhandlers to a namespace, for use with this function. This allows for custom handlingof expansion addresses or starting up expansion services automatically beneath theCrossLanguage call.

In addition, urns can be bound to specific expansion addresses, usingxlangx.RegisterOverrideForUrn. This allows for testing specific overrides, or othercustom implementations to be used instead.

To ignore overrides regardless of URN, wrapping the expansion address ina call to xlangx.Require, will force expansion using the given address.

funcCrossLanguagePayload ¶

func CrossLanguagePayload(plany) []byte

CrossLanguagePayload encodes a native Go struct into a payload for cross-language transforms.payloads are []byte encoded ExternalConfigurationPayload protobufs. In order to fill thecontents of the protobuf, the provided struct will be used to converted to a row encodedrepresentation with an accompanying schema, so the input struct must be compatible with schemas.

Seehttps://beam.apache.org/documentation/programming-guide/#schemas for basic information onschemas, and pkg/beam/core/runtime/graphx/schema for details on schemas in the Go SDK.

Example:

type stringPayload struct {    Data string}encodedPl := beam.CrossLanguagePayload(stringPayload{Data: "foo"})

funcExternalTagged ¶

func ExternalTagged(sScope,urnstring,payload []byte,namedInputs map[string]PCollection,namedOutputTypes map[string]FullType,boundedbool) map[string]PCollection

ExternalTagged defines an external PTransform, and allows re-specifying the tags for the inputand output PCollections. The interpretation of this primitive is runner specific.The runner is responsible for parsing the payload based on the URN provided to implementthe behavior of the operation. Transform libraries should expose an API that capturesthe 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 tagswith values in the payload. Otherwise, prefer External.

funcInit ¶

func Init()

Init is the hook that all user code must call after flags processing andother static initialization, for now.

funcInitialized ¶

func Initialized()bool

Initialized exposes the initialization status for runners.

funcMustTaggedN ¶

func MustTaggedN(ret map[string]PCollection, errerror) map[string]PCollection

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

funcNewPipelineWithRoot ¶

func NewPipelineWithRoot() (*Pipeline,Scope)

NewPipelineWithRoot creates a new empty pipeline and its root scope.

funcParDo0 ¶

func ParDo0(sScope, dofnany, colPCollection, opts ...Option)

ParDo0 inserts a ParDo with zero output transform into the pipeline.

funcParDo2 ¶

func ParDo2(sScope, dofnany, colPCollection, opts ...Option) (PCollection,PCollection)

ParDo2 inserts a ParDo with 2 outputs into the pipeline.

funcParDo3 ¶

func ParDo3(sScope, dofnany, colPCollection, opts ...Option) (PCollection,PCollection,PCollection)

ParDo3 inserts a ParDo with 3 outputs into the pipeline.

funcParDo4 ¶

func ParDo4(sScope, dofnany, colPCollection, opts ...Option) (PCollection,PCollection,PCollection,PCollection)

ParDo4 inserts a ParDo with 4 outputs into the pipeline.

funcParDo5 ¶

func ParDo5(sScope, dofnany, colPCollection, opts ...Option) (PCollection,PCollection,PCollection,PCollection,PCollection)

ParDo5 inserts a ParDo with 5 outputs into the pipeline.

funcParDo6 ¶

func ParDo6(sScope, dofnany, colPCollection, opts ...Option) (PCollection,PCollection,PCollection,PCollection,PCollection,PCollection)

ParDo6 inserts a ParDo with 6 outputs into the pipeline.

funcParDo7 ¶

func ParDo7(sScope, dofnany, colPCollection, opts ...Option) (PCollection,PCollection,PCollection,PCollection,PCollection,PCollection,PCollection)

ParDo7 inserts a ParDo with 7 outputs into the pipeline.

funcRegisterCoder ¶

func RegisterCoder(treflect.Type, encoder, decoderany)

RegisterCoder registers a user defined coder for a given type, and willbe used if there is no existing beam coder for that type.Must be called prior to beam.Init(), preferably in an init() function.

The coder used for a given type follows this ordering:

Coders for Known Beam types.
Coders registered for specific types
Coders registered for interfaces types
Default coder (JSON)

Coders for interface types are iterated over to check if a typesatisfies them, and the most recent one registered will be used.

Repeated registrations of the same type overrides prior ones.

RegisterCoder additionally registers the type, and coder functionsas per RegisterType and RegisterFunction to avoid redundant calls.

Supported Encoder Signatures

func(T) []bytefunc(reflect.Type, T) []bytefunc(T) ([]byte, error)func(reflect.Type, T) ([]byte, error)

Supported Decoder Signatures

func([]byte) Tfunc(reflect.Type, []byte) Tfunc([]byte) (T, error)func(reflect.Type, []byte) (T, error)

where T is the matching user type.

funcRegisterDoFn ¶

func RegisterDoFn(dofnany)

RegisterDoFn is a convenience function to handle registering a DoFn and allrelated types. Use this instead of calling RegisterType or RegisterFunction.Like all the Register* functions, RegisterDoFn should be called in`init()` only.

In particular, it will call RegisterFunction for functional DoFns, andRegisterType for the parameter and return types for that function.StructuralDoFns will have RegisterType called for itself and the parameter andreturn types.

RegisterDoFn will panic if the argument type is not a DoFn.

Usage:

   func init() {    beam.RegisterDoFn(FunctionalDoFn)    beam.RegisterDoFn(reflect.TypeOf((*StructuralDoFn)(nil)).Elem())   }

funcRegisterFunction ¶

func RegisterFunction(fnany)

RegisterFunction allows function registration. It is beneficial for performanceand is needed for functions -- such as custom coders -- serialized during unittests, where the underlying symbol table is not available. It should be calledin `init()` only.

funcRegisterInit ¶

func RegisterInit(hook func())

RegisterInit registers an Init hook. Hooks are expected to be able tofigure out whether they apply on their own, notably if invoked in a remoteexecution environment. They are all executed regardless of the runner.

funcRegisterRunner ¶

func RegisterRunner(namestring, fn func(ctxcontext.Context, p *Pipeline) (PipelineResult,error))

RegisterRunner associates the name with the supplied runner, making it availableto execute a pipeline via Run.

funcRegisterSchemaProvider ¶

func RegisterSchemaProvider(rtreflect.Type, providerany)

RegisterSchemaProvider allows pipeline authors to provide special handlingto convert types to schema representations, when those types are used asfields in types being encoded as schema rows.

At present, the only supported provider interface is SchemaProvider,though this may change in the future.

Providers only need to support a limited set of types for conversion,specifically a single struct type or a pointer to struct type,or an interface type, which they are registered with.

Providers have three tasks with respect to a given supported logical type:

Producing schema representative types for their logical types.
Producing schema encoders for values of that type, writing beamschema encoded bytes for a value, matching the schema representative type.
Producing schema decoders for values of that type, reading beamschema encoded bytes, and producing a value of that type.

Representative Schema types must be structs with only exported fields.

A provider should be thread safe, but it's not required that a producedencoder or decoder is thread safe, since a separate encoder or decoderwill be used for simultaneously executed bundles.

If the supported type is an interface, that interface must have a non-emptymethod set. That is, it cannot be the empty interface.

RegisterSchemaProvider must be called before beam.Init(), and conventionallyis called in a package init() function.

Example¶

// Licensed to the Apache Software Foundation (ASF) under one or more// contributor license agreements.  See the NOTICE file distributed with// this work for additional information regarding copyright ownership.// The ASF licenses this file to You under the Apache License, Version 2.0// (the "License"); you may not use this file except in compliance with// the License.  You may obtain a copy of the License at////    http://www.apache.org/licenses/LICENSE-2.0//// Unless required by applicable law or agreed to in writing, software// distributed under the License is distributed on an "AS IS" BASIS,// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.// See the License for the specific language governing permissions and// limitations under the License.package mainimport ("bytes""fmt""io""reflect""github.com/apache/beam/sdks/v2/go/pkg/beam""github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/coder""github.com/google/go-cmp/cmp")// RegisterSchemaProvider must be called before beam.Init, and conventionally in a package init block.func init() {beam.RegisterSchemaProvider(reflect.TypeOf((*Alphabet)(nil)).Elem(), &AlphabetProvider{})// TODO(BEAM-9615): Registerying a self encoding type causes a cycle. Needs resolving.// beam.RegisterType(reflect.TypeOf((*Cyrillic)(nil)))beam.RegisterType(reflect.TypeOf((*Latin)(nil)))beam.RegisterType(reflect.TypeOf((*Ελληνικά)(nil)))}type Alphabet interface {alphabet() string}type Cyrillic struct {A, B int}func (*Cyrillic) alphabet() string {return "Cyrillic"}type Latin struct {// Unexported fields are not serializable by beam schemas by default// so we need to handle this ourselves.c uint64d *float32}func (*Latin) alphabet() string {return "Latin"}type Ελληνικά struct {q stringG func() string}func (*Ελληνικά) alphabet() string {return "Ελληνικά"}// AlphabetProvider provides encodings for types that implement the Alphabet interface.type AlphabetProvider struct {enc *coder.RowEncoderBuilderdec *coder.RowDecoderBuilder}var (typeCyrillic = reflect.TypeOf((*Cyrillic)(nil))typeLatin    = reflect.TypeOf((*Latin)(nil))typeΕλληνικά = reflect.TypeOf((*Ελληνικά)(nil)))func (p *AlphabetProvider) FromLogicalType(rt reflect.Type) (reflect.Type, error) {// FromLogicalType produces schema representative types, which match the encoders// and decoders that this function generates for this type.// While this example uses statically assigned schema representative types, it's// possible to generate the returned reflect.Type dynamically instead, using the// reflect package.switch rt {// The Cyrillic type is able to be encoded by default, so we simply use it directly// as it's own representative type.case typeCyrillic:return typeCyrillic, nilcase typeLatin:// The Latin type only has unexported fields, so we need to make the equivalent// have exported fields.return reflect.TypeOf((*struct {C uint64D *float32})(nil)).Elem(), nilcase typeΕλληνικά:return reflect.TypeOf((*struct{ Q string })(nil)).Elem(), nil}return nil, fmt.Errorf("Unknown Alphabet: %v", rt)}// BuildEncoder returns beam schema encoder functions for types with the Alphabet interface.func (p *AlphabetProvider) BuildEncoder(rt reflect.Type) (func(any, io.Writer) error, error) {switch rt {case typeCyrillic:if p.enc == nil {p.enc = &coder.RowEncoderBuilder{}}// Since Cyrillic is by default encodable, defer to the standard schema row decoder for the type.return p.enc.Build(rt)case typeLatin:return func(iface any, w io.Writer) error {v := iface.(*Latin)// Beam Schema Rows have a header that indicates which fields if any, are nil.if err := coder.WriteRowHeader(2, func(i int) bool {if i == 1 {return v.d == nil}return false}, w); err != nil {return err}// Afterwards, each field is encoded using the appropriate helper.if err := coder.EncodeVarUint64(v.c, w); err != nil {return err}// Nil fields have nothing written for them other than the header.if v.d != nil {if err := coder.EncodeDouble(float64(*v.d), w); err != nil {return err}}return nil}, nilcase typeΕλληνικά:return func(iface any, w io.Writer) error {// Since the representation for Ελληνικά never has nil fields// we can use the simple header helper.if err := coder.WriteSimpleRowHeader(1, w); err != nil {return err}v := iface.(*Ελληνικά)if err := coder.EncodeStringUTF8(v.q, w); err != nil {return fmt.Errorf("decoding string field A: %v", err)}return nil}, nil}return nil, fmt.Errorf("Unknown Alphabet: %v", rt)}// BuildDecoder returns beam schema decoder functions for types with the Alphabet interface.func (p *AlphabetProvider) BuildDecoder(rt reflect.Type) (func(io.Reader) (any, error), error) {switch rt {case typeCyrillic:if p.dec == nil {p.dec = &coder.RowDecoderBuilder{}}// Since Cyrillic is by default encodable, defer to the standard schema row decoder for the type.return p.dec.Build(rt)case typeLatin:return func(r io.Reader) (any, error) {// Since the d field can be nil, we use the header get the nil bits.n, nils, err := coder.ReadRowHeader(r)if err != nil {return nil, err}// Header returns the number of fields, so we check if it has what we// expect. This allows schemas to evolve if necessary.if n != 2 {return nil, fmt.Errorf("expected 2 fields, but got %v", n)}c, err := coder.DecodeVarUint64(r)if err != nil {return nil, err}// Check if the field is nil before trying to decode a value for it.var d *float32if !coder.IsFieldNil(nils, 1) {f, err := coder.DecodeDouble(r)if err != nil {return nil, err}f32 := float32(f)d = &f32}return &Latin{c: c,d: d,}, nil}, nilcase typeΕλληνικά:return func(r io.Reader) (any, error) {// Since the representation for Ελληνικά never has nil fields// we can use the simple header helper. Returns an error if// something unexpected occurs.if err := coder.ReadSimpleRowHeader(1, r); err != nil {return nil, err}q, err := coder.DecodeStringUTF8(r)if err != nil {return nil, fmt.Errorf("decoding string field A: %v", err)}return &Ελληνικά{q: q,}, nil}, nil}return nil, nil}// Schema providers work on fields of schema encoded types.type translation struct {C *CyrillicL *LatinE *Ελληνικά}func main() {f := float32(42.789)want := translation{C: &Cyrillic{A: 123, B: 456},L: &Latin{c: 789, d: &f},E: &Ελληνικά{q: "testing"},}rt := reflect.TypeOf((*translation)(nil)).Elem()enc, err := coder.RowEncoderForStruct(rt)if err != nil {panic(err)}dec, err := coder.RowDecoderForStruct(rt)if err != nil {panic(err)}var buf bytes.Bufferif err := enc(want, &buf); err != nil {panic(err)}got, err := dec(&buf)if err != nil {panic(err)}if d := cmp.Diff(want, got,cmp.AllowUnexported(Latin{}, Ελληνικά{})); d != "" {fmt.Printf("diff in schema encoding translation: (-want,+got)\n%v\n", d)} else {fmt.Println("No diffs!")}}

Output:No diffs!

funcRegisterSchemaProviderWithURN ¶added inv2.43.0

func RegisterSchemaProviderWithURN(rtreflect.Type, providerany, urnstring)

RegisterSchemaProviderWithURN is for internal use only. Users are recommended to usebeam.RegisterSchemaProvider() instead.RegisterSchemaProviderWithURN registers a new schema provider for a new logical type definedin pkg/beam/model/pipeline_v1/schema.pb.go

RegisterSchemaProviderWithURN must be called before beam.Init(), and conventionallyis called in a package init() function.

funcRegisterType ¶

func RegisterType(treflect.Type)

RegisterType inserts "external" types into a global type registry to bypassserialization and preserve full method information. It should be called in`init()` only.TODO(wcn): the canonical definition of "external" is in v1.proto. We need userfacing copy for this important concept.

funcTryCrossLanguage ¶

func TryCrossLanguage(sScope,urnstring,payload []byte,expansionAddrstring,namedInputs map[string]PCollection,namedOutputTypes map[string]FullType,) (map[string]PCollection,error)

TryCrossLanguage coordinates the core functions required to execute the cross-language transform.See CrossLanguage for user documentation.

funcTryExternalTagged ¶

func TryExternalTagged(sScope,urnstring,payload []byte,namedInputs map[string]PCollection,namedOutputTypes map[string]FullType,boundedbool) (map[string]PCollection,error)

TryExternalTagged attempts to perform the work of ExternalTagged, returning an errorindicating why the operation failed.

funcUnnamedInput ¶

func UnnamedInput(colPCollection) map[string]PCollection

UnnamedInput is a helper function for passing single unnamed inputs tobeam.CrossLanguage.

Example:

beam.CrossLanguage(s, urn, payload, addr, UnnamedInput(input), outputs)

funcUnnamedOutput ¶

func UnnamedOutput(tFullType) map[string]FullType

UnnamedOutput is a helper function for passing single unnamed output types tobeam.CrossLanguage. The associated output can be accessed with beam.UnnamedOutputTag.

Example:

resultMap := beam.CrossLanguage(s, urn, payload, addr, inputs, UnnamedOutput(output));result := resultMap[beam.UnnamedOutputTag()]

funcUnnamedOutputTag ¶added inv2.34.0

func UnnamedOutputTag()string

UnnamedOutputTag provides the output tag used for an output passed to beam.UnnamedOutput.Needed to retrieve the unnamed output PCollection from the result of beam.CrossLanguage.

funcValidateKVType ¶

func ValidateKVType(colPCollection) (typex.FullType,typex.FullType)

ValidateKVType panics if the type of the PCollection is not KV<A,B>.It returns (A,B).

funcValidateNonCompositeType ¶

func ValidateNonCompositeType(colPCollection)typex.FullType

ValidateNonCompositeType panics if the type of the PCollection is not acomposite type. It returns the type.

Types¶

typeBundleFinalization ¶added inv2.39.0

type BundleFinalization =typex.BundleFinalization

BundleFinalization represents the parameter used to register callbacks tobe run once the runner has durably persisted output for a bundle.See typex.BundleFinalization for more details.

typeCoder ¶

type Coder struct {// contains filtered or unexported fields}

Coder defines how to encode and decode values of type 'A' into byte streams.Coders are attached to PCollections of the same type. For PCollectionsconsumed by GBK, the attached coders are required to be deterministic.

funcDecodeCoder ¶

func DecodeCoder(datastring) (Coder,error)

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

funcNewCoder ¶

func NewCoder(tFullType)Coder

NewCoder infers a Coder for any bound full type.

func (Coder)IsValid ¶

func (cCoder) IsValid()bool

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

func (Coder)String ¶

func (cCoder) String()string

func (Coder)Type ¶

func (cCoder) 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>.

typeCounter ¶

type Counter struct {*metrics.Counter}

Counter is a metric that can be incremented and decremented,and is aggregated by the sum.

Counters are safe to use in multiple bundles simultaneously, butnot generally threadsafe. Your DoFn needs to manage the threadsafety of Beam metrics for any additional concurrency it uses.

funcNewCounter ¶

func NewCounter(namespace, namestring)Counter

NewCounter returns the Counter with the given namespace and name.

func (Counter)Dec ¶

func (cCounter) Dec(ctxcontext.Context, vint64)

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

Example¶

package mainimport ("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")))}

func (Counter)Inc ¶

func (cCounter) Inc(ctxcontext.Context, vint64)

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

Example¶

package mainimport ("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")))}

typeDistribution ¶

type Distribution struct {*metrics.Distribution}

Distribution is a metric that records various statistics about the distributionof reported values.

Distributions are safe to use in multiple bundles simultaneously, butnot generally threadsafe. Your DoFn needs to manage the threadsafety of Beam metrics for any additional concurrency it uses.

funcNewDistribution ¶

func NewDistribution(namespace, namestring)Distribution

NewDistribution returns the Distribution with the given namespace and name.

func (Distribution)Update ¶

func (cDistribution) Update(ctxcontext.Context, vint64)

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

Example¶

package mainimport ("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 * 42d := beam.NewDistribution("example", "latency_micros")d.Update(ctx, int64(t/time.Microsecond))}

typeElementDecoder ¶

type ElementDecoder =coder.ElementDecoder

ElementDecoder encapsulates being able to decode an element from a reader.

funcNewElementDecoder ¶

func NewElementDecoder(treflect.Type)ElementDecoder

NewElementDecoder returns an ElementDecoder the given type.

typeElementEncoder ¶

type ElementEncoder =coder.ElementEncoder

ElementEncoder encapsulates being able to encode an element into a writer.

funcNewElementEncoder ¶

func NewElementEncoder(treflect.Type)ElementEncoder

NewElementEncoder returns a new encoding function for the given type.

typeEncodedCoder ¶

type EncodedCoder struct {// Coder is the coder to preserve across serialization.CoderCoder}

EncodedCoder is a serialization wrapper around a coder for convenience.

func (EncodedCoder)MarshalJSON ¶

func (wEncodedCoder) MarshalJSON() ([]byte,error)

MarshalJSON returns the JSON encoding this value.

func (*EncodedCoder)UnmarshalJSON ¶

func (w *EncodedCoder) UnmarshalJSON(buf []byte)error

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

typeEncodedFunc ¶

type EncodedFunc struct {// Fn is the function to preserve across serialization.Fnreflectx.Func}

EncodedFunc is a serialization wrapper around a function for convenience.

func (EncodedFunc)MarshalJSON ¶

func (wEncodedFunc) MarshalJSON() ([]byte,error)

MarshalJSON returns the JSON encoding this value.

func (*EncodedFunc)UnmarshalJSON ¶

func (w *EncodedFunc) UnmarshalJSON(buf []byte)error

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

typeEncodedType ¶

type EncodedType struct {// T is the type to preserve across serialization.Treflect.Type}

EncodedType is a serialization wrapper around a type for convenience.

func (EncodedType)MarshalJSON ¶

func (wEncodedType) MarshalJSON() ([]byte,error)

MarshalJSON returns the JSON encoding this value.

func (*EncodedType)UnmarshalJSON ¶

func (w *EncodedType) UnmarshalJSON(buf []byte)error

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

typeEventTime ¶

type EventTime =typex.EventTime

EventTime represents the time of the event that generated an element.This is distinct from the time when an element is processed.

typeFullType ¶

type FullType =typex.FullType

FullType represents the tree structure of data types processed by the graph.It allows representation of composite types, such as KV<int, string> orCoGBK<int, int>, as well as "generic" such types, KV<int,T> or CoGBK<X,Y>,where the free "type variables" are the fixed universal types: T, X, etc.

typeGauge ¶

type Gauge struct {*metrics.Gauge}

Gauge is a metric that can have its new value set, and is aggregated by takingthe last reported value.

Gauge are safe to use in multiple bundles simultaneously, butnot generally threadsafe. Your DoFn needs to manage the threadsafety of Beam metrics for any additional concurrency it uses.

funcNewGauge ¶

func NewGauge(namespace, namestring)Gauge

NewGauge returns the Gauge with the given namespace and name.

func (Gauge)Set ¶

func (cGauge) Set(ctxcontext.Context, vint64)

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

Example¶

package mainimport ("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)}

typeMetricResult ¶added inv2.35.0

type MetricResult =metrics.SingleResult

MetricResult represents a single metric value, for use in writing predicate functions to query PipelineResults.

typeOption ¶

type Option interface {// contains filtered or unexported methods}

Option is an optional value or context to a transformation, used at pipelineconstruction time. The primary use case is providing side inputs.

typePCollection ¶

type PCollection struct {// contains filtered or unexported fields}

PCollection is an immutable collection of values of type 'A', which must bea concrete type, such as int or KV<int,string>. A PCollection can containeither a bounded or unbounded number of elements. Bounded and unboundedPCollections are produced as the output of PTransforms (including rootPTransforms like textio.Read), and can be passed as the inputs of otherPTransforms. Some root transforms produce bounded PCollections and othersproduce unbounded ones.

Each element in a PCollection has an associated timestamp. Sources assigntimestamps to elements when they create PCollections, and other PTransformspropagate these timestamps from their input to their output implicitly orexplicitly.

Additionally, each element is assigned to a set of windows. By default, allelements are assigned into a single default window, GlobalWindow.

funcAddFixedKey ¶

func AddFixedKey(sScope, colPCollection)PCollection

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

funcCoGroupByKey ¶

func CoGroupByKey(sScope, cols ...PCollection)PCollection

CoGroupByKey inserts a CoGBK transform into the pipeline.

funcCombine ¶

func Combine(sScope, combinefnany, colPCollection, opts ...Option)PCollection

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

funcCombinePerKey ¶

func CombinePerKey(sScope, combinefnany, colPCollection, opts ...Option)PCollection

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

funcCreate ¶

func Create(sScope, values ...any)PCollection

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

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

Example¶

package mainimport ("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>}

funcCreateList ¶

func CreateList(sScope, listany)PCollection

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

Example¶

package mainimport ("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>}

funcDropKey ¶

func DropKey(sScope, colPCollection)PCollection

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

funcDropValue ¶

func DropValue(sScope, colPCollection)PCollection

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

funcExplode ¶

func Explode(sScope, colPCollection)PCollection

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

Example¶

package mainimport ("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>}

funcExternal ¶

func External(sScope, urnstring, payload []byte, in []PCollection, out []FullType, boundedbool) []PCollection

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

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

funcFlatten ¶

func Flatten(sScope, 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 inall the input PCollections. The name "Flatten" suggests taking a list of listsand flattening them into a single list.

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

Example¶

package mainimport ("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>}

funcGroupByKey ¶

func GroupByKey(sScope, aPCollection)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 typeGBK<A,B> representing a map from each distinct key and window of theinput PCollection to an iterable over all the values associated withthat key in the input per window. Each key in the output PCollection isunique within each window.

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

Two keys of type A are compared for equality by first encoding each of thekeys using the Coder of the keys of the input PCollection, and thencomparing 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 iterablevalues in the input and output PCollection share an element Coder.

GroupByKey is a key primitive in data-parallel processing, since it is themain 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 commonkey at once.

Example¶

package mainimport ("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 Docfor values(&cur) {// ... process all docs having that url ...}}, urlToDocs) // PCollection<KV<string, []Doc>>}

funcImpulse ¶

func Impulse(sScope)PCollection

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

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

Example¶

package mainimport ("github.com/apache/beam/sdks/v2/go/pkg/beam")var s = beam.Scope{}func main() {beam.Impulse(s) // PCollection<[]byte>}

funcImpulseValue ¶

func ImpulseValue(sScope, value []byte)PCollection

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

Example¶

package mainimport ("github.com/apache/beam/sdks/v2/go/pkg/beam")var s = beam.Scope{}func main() {beam.ImpulseValue(s, []byte{}) // PCollection<[]byte>}

funcMust ¶

func Must(aPCollection, errerror)PCollection

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

funcMustN ¶

func MustN(list []PCollection, errerror) []PCollection

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

funcParDo ¶

func ParDo(sScope, dofnany, colPCollection, opts ...Option)PCollection

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

Elements are processed independently, and possibly in parallel acrossdistributed cloud resources. The ParDo processing style is similar to whathappens inside the "Mapper" or "Reducer" class of a MapReduce-stylealgorithm.

DoFns¶

The function to use to process each element is specified by a DoFn, either assingle function or as a struct with methods, notably ProcessElement. Thestruct 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 thebeam `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 inputPCollection are first divided up into some number of "bundles". These arefarmed 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 aworker from json serialization, and the Setup method is called on thisinstance, if present. A runner may reuse DoFn instances for multiplebundles. 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 elementsin the bundle.
The DoFn's FinishBundle method, if provided, is called to complete itswork. After FinishBundle is called, the framework will not again invokeProcessElement or FinishBundle until a new call to StartBundle hasoccurred.
If any of Setup, StartBundle, ProcessElement or FinishBundle methodsreturn an error, the Teardown method, if provided, will be called on theDoFn 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 zeroor more output elements. All of the of output elements from all of the DoFninstances 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 itscorresponding input element. The timestamp can be accessed and/or emitted byincluding a EventTime-typed parameter. The name of the function or struct isused as the DoFn name. Function literals do not have stable names and shouldthus not be used in production code.

Side Inputs¶

While a ParDo processes elements from a single "main input" PCollection, itcan take additional "side input" PCollections. These SideInput along withthe DoFn parameter form express styles of accessing PCollection computed byearlier pipeline operations, passed in to the ParDo transform using SideInputoptions, and their contents accessible to each of the DoFn operations. Forexample:

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 outputPCollections. 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 isinferred from the concrete type.

No Global Shared State¶

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

Define public instance variable state. This state will be automaticallyJSON serialized and then deserialized in the DoFn instances created forbundles. This method is good for state known when the original DoFn iscreated in the main program, if it's not overly large. This is notsuitable for any state which must only be used for a single bundle, asDoFn's may be used to process multiple bundles.
Compute the state as a singleton PCollection and pass it in as a sideinput to the DoFn. This is good if the state needs to be computed by thepipeline, 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 informationknown only by the main program or computed by earlier pipelineoperations, but is the same for all instances of this DoFn for allprogram executions, say setting up empty caches or initializing constantdata.

ParDo operations are intended to be able to run in parallel across multipleworker machines. This precludes easy sharing and updating mutable stateacross those machines. There is no support in the Beam model forcommunicating and synchronizing updates to shared state across workermachines, so programs should not access any mutable global variable state intheir DoFn, without understanding that the Go processes for the main programand workers will each have its own independent copy of such state, and therewon't be any automatic copying of that state across Java processes. Allinformation should be communicated to DoFn instances via main and sideinputs and serialized state, and all output should be communicated from aDoFn instance via output PCollections, in the absence of externalcommunication 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 usefulfor DoFns that emit many outputs per input element and can distribute thatwork among multiple workers. The most common examples of this are sources.

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

Splitting¶

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

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

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

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

Splittable DoFn Methods¶

Making a splittable DoFn requires the following methods to be implemented ona DoFn in addition to the usual DoFn requirements. In the followingmethod signatures `elem` represents the main input elements to the DoFn, andshould match the types used in ProcessElement. `restriction` represents theuser-defined restriction, and can be any type as long as it is consistentthroughout all the splittable DoFn methods:

`CreateInitialRestriction(context.Context?, elem) (restriction, error?)`CreateInitialRestriction creates an initial restriction encompassing anentire element. The restriction created stays associated with the elementit describes.
`SplitRestriction(context.Context?, elem, restriction) ([]restriction, error?)`SplitRestriction takes an element and its initial restriction, andoptionally performs an initial split on it, returning a slice of all thesplit restrictions. If no splits are desired, the method returns a slicecontaining only the original restriction. This method will always becalled 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. Thissize is an abstract non-negative scalar value that represents how muchwork 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 of100, but the numbers do not represent anything on their own. Size isused by runners to estimate work for dynamic work rebalancing. Must bethread safe. Will be invoked concurrently during bundle processing due torunner initiated splitting and progress estimation.
`CreateTracker(context.Context?, restriction) (restrictionTracker, error?)`CreateTracker creates and returns a restriction tracker (a concrete typeimplementing the `sdf.RTracker` interface) given a restriction. Therestriction tracker is used to track progress processing a restriction,and to allow for dynamic splits. This method is called on eachrestriction right before processing begins.
`ProcessElement(context.Context?, sdf.RTracker, elem, func emit(output))(sdf.ProcessContinuation?, error?)`For splittable DoFns, ProcessElement requires a restriction trackerbefore inputs, and generally requires emits to be used for outputs, sincerestrictions will generally produce multiple outputs. For more detailson processing restrictions in a splittable DoFn, see `sdf.RTracker`.ProcessElement can optionally return a `sdf.ProcessContinuation` tosignal to the runner that processing should be resumed at a later time,if not all data within the restriction can be processed within thelifetime 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 runnersthat support pipeline draining. It helps finish the pipeline faster bytruncating the restriction. If not implemented, the default behavior forbounded restrictions is to process the remainder of the restriction, andfor unbounded restrictions to process until the next SDF-initiatedcheckpoint or runner-initiated split occurs.

Fault Tolerance¶

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

Optimization¶

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

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 itselfis "fused away" and won't ever be written to disk, saving all the I/O andspace expense of constructing it.

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

Seehttps://beam.apache.org/documentation/programming-guide/#pardofor the web documentation for ParDo

Example (AdditionalOutputs)¶

Optionally, a ParDo transform can produce zero or multiple outputPCollections. Note the use of ParDo2 to specify 2 outputs.

package mainimport ("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}

funcParDoN ¶

func ParDoN(sScope, dofnany, colPCollection, opts ...Option) []PCollection

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

funcPartition ¶

func Partition(sScope, nint, fnany, colPCollection) []PCollection

Partition takes a PCollection<T> and a PartitionFn, uses the PartitionFn tosplit the elements of the input PCollection into N partitions, and returnsa []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 10wordsByLength := beam.Partition(s, 11, lenToTen, inputStrings)

T is permitted to be a KV.

funcReshuffle ¶

func Reshuffle(sScope, colPCollection)PCollection

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

For example, if a computation needs a lot of parallelism butproduces only a small amount of output data, then the computationproducing the data can run with as much parallelism as needed,while the output file is written with a smaller amount ofparallelism, 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 onethat performs externally-visible side effects. Inserting a Reshufflebetween these DoFns ensures that retries of the second DoFn will always bethe 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 thepipeline without reshuffling is broken in some way and performing an extraoperation is worth the cost.

funcSeq ¶

func Seq(sScope, colPCollection, dofns ...any)PCollection

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

Example¶

package mainimport ("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 intfunc(i int) float64 { return float64(i) }, // int to float64math.Signbit, // float64 to bool) // PCollection<bool>}

funcSwapKV ¶

func SwapKV(sScope, colPCollection)PCollection

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

funcTryCoGroupByKey ¶

func TryCoGroupByKey(sScope, cols ...PCollection) (PCollection,error)

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

funcTryCombine ¶

func TryCombine(sScope, combinefnany, colPCollection, opts ...Option) (PCollection,error)

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

funcTryCombinePerKey ¶

func TryCombinePerKey(sScope, combinefnany, colPCollection, opts ...Option) (PCollection,error)

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

funcTryCreate ¶

func TryCreate(sScope, values ...any) (PCollection,error)

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

funcTryCreateList ¶

func TryCreateList(sScope, listany) (PCollection,error)

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

funcTryExternal ¶

func TryExternal(sScope, urnstring, payload []byte, in []PCollection, out []FullType, boundedbool) ([]PCollection,error)

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

funcTryFlatten ¶

func TryFlatten(sScope, cols ...PCollection) (PCollection,error)

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

funcTryGroupByKey ¶

func TryGroupByKey(sScope, aPCollection) (PCollection,error)

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

funcTryParDo ¶

func TryParDo(sScope, dofnany, colPCollection, opts ...Option) ([]PCollection,error)

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

funcTryReshuffle ¶

func TryReshuffle(sScope, colPCollection) (PCollection,error)

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

funcTryWindowInto ¶

func TryWindowInto(sScope, wfn *window.Fn, colPCollection, opts ...WindowIntoOption) (PCollection,error)

TryWindowInto attempts to insert a WindowInto transform.

funcWindowInto ¶

func WindowInto(sScope, ws *window.Fn, colPCollection, opts ...WindowIntoOption)PCollection

WindowInto applies the windowing strategy to each element.

func (PCollection)Coder ¶

func (pPCollection) Coder()Coder

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

func (PCollection)IsValid ¶

func (pPCollection) IsValid()bool

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

func (PCollection)SetCoder ¶

func (pPCollection) SetCoder(cCoder)error

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

func (PCollection)String ¶

func (pPCollection) String()string

func (PCollection)Type ¶

func (pPCollection) Type()FullType

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

typePaneInfo ¶added inv2.38.0

type PaneInfo =typex.PaneInfo

PaneInfo represents a PaneInfo that provides information about current firing when triggers are used.

typePipeline ¶

type Pipeline struct {// contains filtered or unexported fields}

Pipeline manages a directed acyclic graph of primitive PTransforms, and thePCollections that the PTransforms consume and produce. Each Pipeline isself-contained and isolated from any other Pipeline. The Pipeline owns thePCollections and PTransforms and they can be used by that Pipeline only.Pipelines can safely be executed concurrently.

funcNewPipeline ¶

func NewPipeline() *Pipeline

NewPipeline creates a new empty pipeline.

func (*Pipeline)Build ¶

func (p *Pipeline) Build() ([]*graph.MultiEdge, []*graph.Node,error)

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

func (*Pipeline)Root ¶

func (p *Pipeline) Root()Scope

Root returns the root scope of the pipeline.

func (*Pipeline)String ¶

func (p *Pipeline) String()string

typePipelineResult ¶

type PipelineResult interface {Metrics()metrics.ResultsJobID()string}

PipelineResult is the result of beamx.RunWithMetrics.

funcRun ¶

func Run(ctxcontext.Context, runnerstring, p *Pipeline) (PipelineResult,error)

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

typeSchemaProvider ¶

type SchemaProvider interface {FromLogicalType(reflect.Type) (reflect.Type,error)BuildEncoder(rtreflect.Type) (func(any,io.Writer)error,error)BuildDecoder(rtreflect.Type) (func(io.Reader) (any,error),error)}

SchemaProvider specializes schema handling for complex types, including conversion to avalid schema base type,

In particular, they are intended to handle schema for interface types.

Sepearated out the acting type from the provider implementation is good.

typeScope ¶

type Scope struct {// contains filtered or unexported fields}

Scope is a hierarchical grouping for composite transforms. Scopes can beenclosed in other scopes and for a tree structure. For pipeline updates,the scope chain form a unique name. The scope chain can also be used formonitoring and visualization purposes.

func (Scope)IsValid ¶

func (sScope) IsValid()bool

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

func (Scope)Scope ¶

func (sScope) Scope(namestring)Scope

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

func (Scope)String ¶

func (sScope) String()string

func (Scope)WithContext ¶added inv2.53.0

func (sScope) WithContext(ctxcontext.Context, namestring)Scope

WithContext creates a named subscope with an attached context for therepresented composite transform. Values from that context may beextracted and added to the composite PTransform or generate a newenvironment for scoped transforms.

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

typeSideInput ¶

type SideInput struct {InputPCollection}

SideInput provides a view of the given PCollection to the transformation.

Example¶

package mainimport ("github.com/apache/beam/sdks/v2/go/pkg/beam")var s = beam.Scope{}func main() {// words and sample are PCollection<string>var words, sample beam.PCollection// analyzeFn emits values from the primary based on the singleton side input.analyzeFn := func(primary string, side string, emit func(string)) {}// Use beam.SideInput to declare that the sample PCollection is the side input.beam.ParDo(s, analyzeFn, words, beam.SideInput{Input: sample})}

typeT ¶

type T =typex.T

T is a Universal Type used to represent "generic" types in DoFn andPCollection signatures. Each universal type is distinct from all others.

typeTypeDefinition ¶

type TypeDefinition struct {// Var is the universal type defined.Varreflect.Type// T is the type it is bound to.Treflect.Type}

TypeDefinition provides construction-time type information that the platformcannot infer, such as structured storage sources. These types are universal typesthat appear as output only. Types that are inferrable should not be conveyed viathis mechanism.

typeU ¶

type U =typex.U

U is a Universal Type used to represent "generic" types in DoFn andPCollection signatures. Each universal type is distinct from all others.

typeV ¶

type V =typex.V

V is a Universal Type used to represent "generic" types in DoFn andPCollection signatures. Each universal type is distinct from all others.

typeW ¶

type W =typex.W

W is a Universal Type used to represent "generic" types in DoFn andPCollection signatures. Each universal type is distinct from all others.

typeWindow ¶

type Window =typex.Window

Window represents the aggregation window of this element. An element canbe a part of multiple windows, based on the element's event time.

typeWindowIntoOption ¶

type WindowIntoOption interface {// contains filtered or unexported methods}

funcAllowedLateness ¶added inv2.34.0

func AllowedLateness(delaytime.Duration)WindowIntoOption

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

funcPanesAccumulate ¶added inv2.34.0

func PanesAccumulate()WindowIntoOption

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

funcPanesDiscard ¶added inv2.34.0

func PanesDiscard()WindowIntoOption

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

funcTrigger ¶added inv2.34.0

func Trigger(trtrigger.Trigger)WindowIntoOption

Trigger applies the given trigger to the window.

typeX ¶

type X =typex.X

X is a Universal Type used to represent "generic" types in DoFn andPCollection signatures. Each universal type is distinct from all others.

typeY ¶

type Y =typex.Y

Y is a Universal Type used to represent "generic" types in DoFn andPCollection signatures. Each universal type is distinct from all others.

typeZ ¶

type Z =typex.Z

Z is a Universal Type used to represent "generic" types in DoFn andPCollection signatures. Each universal type is distinct from all others.

Source Files¶

View all Source files

Directories¶

?	: This menu
/	: Search site
f orF	: Jump to
y orY	: Canonical URL

Movatterモバイル変換

Details

Repository

Links

Documentation¶

Overview¶

Index¶

Examples¶

Constants¶

Variables¶

Functions¶

funcCrossLanguage¶

Payload¶

Inputs and Outputs¶

Example¶

Alternative Expansion Handlers¶

funcCrossLanguagePayload¶

funcExternalTagged¶

funcInit¶

funcInitialized¶

funcMustTaggedN¶

funcNewPipelineWithRoot¶

funcParDo0¶

funcParDo2¶

funcParDo3¶

funcParDo4¶

funcParDo5¶

funcParDo6¶

funcParDo7¶

funcRegisterCoder¶

funcRegisterDoFn¶

funcRegisterFunction¶

funcRegisterInit¶

funcRegisterRunner¶

funcRegisterSchemaProvider¶

funcRegisterSchemaProviderWithURN¶added inv2.43.0

funcRegisterType¶

funcTryCrossLanguage¶

funcTryExternalTagged¶

funcUnnamedInput¶

funcUnnamedOutput¶

funcUnnamedOutputTag¶added inv2.34.0

funcValidateKVType¶

funcValidateNonCompositeType¶

Types¶

typeBundleFinalization¶added inv2.39.0

typeCoder¶

funcDecodeCoder¶

funcNewCoder¶

func (Coder)IsValid¶

func (Coder)String¶

func (Coder)Type¶

typeCounter¶

funcNewCounter¶

func (Counter)Dec¶

func (Counter)Inc¶

typeDistribution¶

funcNewDistribution¶

func (Distribution)Update¶

typeElementDecoder¶

funcNewElementDecoder¶

typeElementEncoder¶

funcNewElementEncoder¶

typeEncodedCoder¶

func (EncodedCoder)MarshalJSON¶

func (*EncodedCoder)UnmarshalJSON¶

typeEncodedFunc¶

func (EncodedFunc)MarshalJSON¶

func (*EncodedFunc)UnmarshalJSON¶

typeEncodedType¶

func (EncodedType)MarshalJSON¶

func (*EncodedType)UnmarshalJSON¶

typeEventTime¶

typeFullType¶

typeGauge¶

funcNewGauge¶

func (Gauge)Set¶

typeMetricResult¶added inv2.35.0

typeOption¶

typePCollection¶

Documentation ¶

funcCrossLanguage ¶

funcCrossLanguagePayload ¶

funcExternalTagged ¶

funcInit ¶

funcInitialized ¶

funcMustTaggedN ¶

funcNewPipelineWithRoot ¶

funcParDo0 ¶

funcParDo2 ¶

funcParDo3 ¶

funcParDo4 ¶

funcParDo5 ¶

funcParDo6 ¶

funcParDo7 ¶

funcRegisterCoder ¶

funcRegisterDoFn ¶

funcRegisterFunction ¶

funcRegisterInit ¶

funcRegisterRunner ¶

funcRegisterSchemaProvider ¶

funcRegisterSchemaProviderWithURN ¶added inv2.43.0

funcRegisterType ¶

funcTryCrossLanguage ¶

funcTryExternalTagged ¶

funcUnnamedInput ¶

funcUnnamedOutput ¶

funcUnnamedOutputTag ¶added inv2.34.0

funcValidateKVType ¶

funcValidateNonCompositeType ¶

typeBundleFinalization ¶added inv2.39.0

typeCoder ¶

funcDecodeCoder ¶

funcNewCoder ¶

func (Coder)IsValid ¶

func (Coder)String ¶

func (Coder)Type ¶

typeCounter ¶

funcNewCounter ¶

func (Counter)Dec ¶

func (Counter)Inc ¶

typeDistribution ¶

funcNewDistribution ¶

func (Distribution)Update ¶

typeElementDecoder ¶

funcNewElementDecoder ¶

typeElementEncoder ¶

funcNewElementEncoder ¶

typeEncodedCoder ¶

func (EncodedCoder)MarshalJSON ¶

func (*EncodedCoder)UnmarshalJSON ¶

typeEncodedFunc ¶

func (EncodedFunc)MarshalJSON ¶

func (*EncodedFunc)UnmarshalJSON ¶

typeEncodedType ¶

func (EncodedType)MarshalJSON ¶

func (*EncodedType)UnmarshalJSON ¶

typeEventTime ¶

typeFullType ¶

typeGauge ¶

funcNewGauge ¶

func (Gauge)Set ¶

typeMetricResult ¶added inv2.35.0

typeOption ¶

typePCollection ¶

funcAddFixedKey ¶

funcCoGroupByKey ¶

funcCombine ¶

funcCombinePerKey ¶

funcCreate ¶

funcCreateList ¶

funcDropKey ¶

funcDropValue ¶

funcExplode ¶

funcExternal ¶

funcFlatten ¶

funcGroupByKey ¶

funcImpulse ¶

funcImpulseValue ¶

funcMust ¶