snippets

Documentation

Overview

Package snippets contains code used in the Beam Programming Guide as examples for the Apache Beam Go SDK. These snippets are compiled and their tests run to ensure correctness. However, due to their piecemeal pedagogical use, they may not be the best example of production code.

The Beam Programming Guide can be found at https://beam.apache.org/documentation/programming-guide/.

Package snippets contains code used in the Beam Programming Guide as examples for the Apache Beam Go SDK. These snippets are compiled and their tests run to ensure correctness. However, due to their piecemeal pedagogical use, they may not be the best example of production code.

The Beam Programming Guide can be found at https://beam.apache.org/documentation/programming-guide/.

Index

• func AddBufferDoFn[V any](s beam.Scope, in beam.PCollection) beam.PCollection
• func AddDynamicTimerTagsDoFn[V hasAction](s beam.Scope, in beam.PCollection) beam.PCollection
• func AddEventTimeDoFn(s beam.Scope, in beam.PCollection) beam.PCollection
• func AddJoinDoFn(s beam.Scope, in beam.PCollection) beam.PCollection
• func AddProcessingTimeDoFn(s beam.Scope, in beam.PCollection) beam.PCollection
• func AddTimedOutputBatching[V any](s beam.Scope, in beam.PCollection) beam.PCollection
• func AddTimerGarbageCollection[V any](s beam.Scope, in beam.PCollection) beam.PCollection
• func AddTimestampDoFn(element LogEntry, emit func(beam.EventTime, LogEntry))
• func ComplexTriggers(s beam.Scope, pcollection beam.PCollection)
• func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection
• func Create()
• func CreateAndSplit(s beam.Scope, input []stringPair) beam.PCollection
• func LogicalTypeExample()
• func PipelineConstruction()
• func PipelineOptions()
• func PipelineOptionsCustom()
• func TriggerAfterEndOfWindow(s beam.Scope, pCollection beam.PCollection)
• func TriggerAlways(s beam.Scope, pCollection beam.PCollection)
• type ComputeWordLengthFn
  • func (fn *ComputeWordLengthFn) ProcessElement(word string, emit func(int))
• type CustomWatermarkEstimator
  • func (e *CustomWatermarkEstimator) CurrentWatermark() time.Time
  • func (e *CustomWatermarkEstimator) ObserveTimestamp(ts time.Time)
• type Event
• type JoinedEvent
• type LogEntry
• type MyCustomType
  • func (m MyCustomType) Bytes() []byte
  • func (m MyCustomType) FromBytes(_ []byte) MyCustomType
• type MyMetricsDoFn
  • func (fn *MyMetricsDoFn) ProcessElement(ctx context.Context, v beam.V, emit func(beam.V))
  • func (fn *MyMetricsDoFn) Setup()
• type Purchase
• type Record
• type ShippingAddress
• type SomeService
• type Student
• type TimestampNanos
  • func (tn TimestampNanos) Nanos() int32
  • func (tn TimestampNanos) Seconds() int64
• type TimestampNanosProvider
  • func (p *TimestampNanosProvider) BuildDecoder(rt reflect.Type) (func(io.Reader) (any, error), error)
  • func (p *TimestampNanosProvider) BuildEncoder(rt reflect.Type) (func(any, io.Writer) error, error)
  • func (p *TimestampNanosProvider) FromLogicalType(rt reflect.Type) (reflect.Type, error)
• type Transaction
• type WatermarkState

Functions

func AddBufferDoFn

func AddBufferDoFn[V any](s beam.Scope, in beam.PCollection) beam.PCollection

func AddDynamicTimerTagsDoFn

func AddDynamicTimerTagsDoFn[V hasAction](s beam.Scope, in beam.PCollection) beam.PCollection

func AddEventTimeDoFn

func AddEventTimeDoFn(s beam.Scope, in beam.PCollection) beam.PCollection

func AddJoinDoFn

func AddJoinDoFn(s beam.Scope, in beam.PCollection) beam.PCollection

func AddProcessingTimeDoFn

func AddProcessingTimeDoFn(s beam.Scope, in beam.PCollection) beam.PCollection

func AddTimedOutputBatching

func AddTimedOutputBatching[V any](s beam.Scope, in beam.PCollection) beam.PCollection

func AddTimerGarbageCollection

func AddTimerGarbageCollection[V any](s beam.Scope, in beam.PCollection) beam.PCollection

func AddTimestampDoFn

func AddTimestampDoFn(element LogEntry, emit func(beam.EventTime, LogEntry))

AddTimestampDoFn extracts an event time from a LogEntry.

func ComplexTriggers

func ComplexTriggers(s beam.Scope, pcollection beam.PCollection)

func CountWords

func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection

[START countwords_composite] CountWords is a function that builds a composite PTransform to count the number of times each word appears.

func Create

func Create()

Create demonstrates using beam.CreateList.

func CreateAndSplit

func CreateAndSplit(s beam.Scope, input []stringPair) beam.PCollection

CreateAndSplit is a helper function that creates

func LogicalTypeExample

func LogicalTypeExample()

func PipelineConstruction

func PipelineConstruction()

PipelineConstruction contains snippets for the initial sections of the Beam Programming Guide, from initializing to submitting a pipeline.

func PipelineOptions

func PipelineOptions()

PipelineOptions shows basic pipeline options using flags.

func PipelineOptionsCustom

func PipelineOptionsCustom()

PipelineOptionsCustom shows slightly less basic pipeline options using flags.

func TriggerAfterEndOfWindow

func TriggerAfterEndOfWindow(s beam.Scope, pCollection beam.PCollection)

func TriggerAlways

func TriggerAlways(s beam.Scope, pCollection beam.PCollection)

Types

type ComputeWordLengthFn

type ComputeWordLengthFn struct{}

ComputeWordLengthFn is the DoFn to perform on each element in the input PCollection.

func (*ComputeWordLengthFn) ProcessElement

func (fn *ComputeWordLengthFn) ProcessElement(word string, emit func(int))

ProcessElement is the method to execute for each element.

type CustomWatermarkEstimator

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

CustomWatermarkEstimator is a custom watermark estimator. You may use any type here, including some of Beam's built in watermark estimator types, e.g. sdf.WallTimeWatermarkEstimator, sdf.TimestampObservingWatermarkEstimator, and sdf.ManualWatermarkEstimator

func (*CustomWatermarkEstimator) CurrentWatermark

func (e *CustomWatermarkEstimator) CurrentWatermark() time.Time

CurrentWatermark returns the current watermark and is invoked on DoFn splits and self-checkpoints. Watermark estimators must implement CurrentWatermark() time.Time

func (*CustomWatermarkEstimator) ObserveTimestamp

func (e *CustomWatermarkEstimator) ObserveTimestamp(ts time.Time)

ObserveTimestamp is called on the output timestamps of all emitted elements to update the watermark. It is optional

type Event

type Event struct{}

type JoinedEvent

type JoinedEvent struct {
	View, Click *Event
}

type LogEntry

type LogEntry int

LogEntry is a dummy type for documentation purposes.

type MyCustomType

type MyCustomType struct{}

func (MyCustomType) Bytes

func (m MyCustomType) Bytes() []byte

func (MyCustomType) FromBytes

func (m MyCustomType) FromBytes(_ []byte) MyCustomType

type MyMetricsDoFn

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

func (*MyMetricsDoFn) ProcessElement

func (fn *MyMetricsDoFn) ProcessElement(ctx context.Context, v beam.V, emit func(beam.V))

func (*MyMetricsDoFn) Setup

func (fn *MyMetricsDoFn) Setup()

type Purchase

type Purchase struct {
	// ID of the user who made the purchase.
	UserID string `beam:"userId"`
	// Identifier of the item that was purchased.
	ItemID int64 `beam:"itemId"`
	// The shipping address, a nested type.
	ShippingAddress ShippingAddress `beam:"shippingAddress"`
	// The cost of the item in cents.
	Cost int64 `beam:"cost"`
	// The transactions that paid for this purchase.
	// A slice since the purchase might be spread out over multiple
	// credit cards.
	Transactions []Transaction `beam:"transactions"`
}

type Record

type Record struct{}

type ShippingAddress

type ShippingAddress struct {
	StreetAddress string  `beam:"streetAddress"`
	City          string  `beam:"city"`
	State         *string `beam:"state"`
	Country       string  `beam:"country"`
	PostCode      string  `beam:"postCode"`
}

type SomeService

type SomeService struct {
	ThrottlingErr error
}

type Student

type Student struct {
	Percentile int
}

type TimestampNanos

type TimestampNanos time.Time

TimestampNanos is a logical type using time.Time, but encodes as a schema type.

func (TimestampNanos) Nanos

func (tn TimestampNanos) Nanos() int32

func (TimestampNanos) Seconds

func (tn TimestampNanos) Seconds() int64

type TimestampNanosProvider

type TimestampNanosProvider struct{}

TimestampNanosProvider implements the beam.SchemaProvider interface.

func (*TimestampNanosProvider) BuildDecoder

func (p *TimestampNanosProvider) BuildDecoder(rt reflect.Type) (func(io.Reader) (any, error), error)

BuildDecoder builds a Beam schema decoder for the TimestampNanos type.

func (*TimestampNanosProvider) BuildEncoder

func (p *TimestampNanosProvider) BuildEncoder(rt reflect.Type) (func(any, io.Writer) error, error)

BuildEncoder builds a Beam schema encoder for the TimestampNanos type.

func (*TimestampNanosProvider) FromLogicalType

func (p *TimestampNanosProvider) FromLogicalType(rt reflect.Type) (reflect.Type, error)

FromLogicalType converts checks if the given type is TimestampNanos, and if so returns the storage type.

type Transaction

type Transaction struct {
	Bank           string  `beam:"bank"`
	PurchaseAmount float64 `beam:"purchaseAmount"`
}

type WatermarkState

type WatermarkState struct {
	Watermark time.Time
}

WatermarkState is a custom type.`

It is optional to write your own state type when making a custom estimator.