beam

type AccumulatorCreator ¶

type AccumulatorCreator[A Element] interface {
	CreateAccumulator() A
	AccumulatorMerger[A]
}

AccumulatorCreator is an interface to allow combiners to produce a more sophisticated accumulator type, when the zero value is inappropriate for accumulation.

type AccumulatorMerger ¶

type AccumulatorMerger[A Element] interface {
	MergeAccumulators(A, A) A
}

AccumulatorMerger is an interface for combiners that only need a binary merge, and the input, output, and accumulator types are all the same.

type AfterBundle ¶

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

AfterBundle allows a DoFn to register a function that runs after the bundle has been durably committed. Emiting elements here will fail.

TODO consider moving this to a simple interface function. Upside, not likely to try to incorrectly emit in the closure. Downside, the caching for anything to finalize needs to be stored in the DoFn struct this violates the potential of a ConfigOnly DoFn.

func (*AfterBundle) Do(dfc bundleFinalizer, finalizeBundle func() error)

type Combiner ¶

type Combiner[A, I, O Element, AM AccumulatorMerger[A]] struct {
	// contains filtered or unexported fields
}

Combiners represent an optimizable approach to aggregating, by breaking down the aggregation into 3 component types.

func AddMerge[A, I Element, IA InputAdder[A, I]](c IA) Combiner[A, I, A, IA]

func FullCombine[A, I, O Element, C FullCombiner[A, I, O]](c C) Combiner[A, I, O, C]

func MergeExtract[A, O Element, OE OutputExtractor[A, O]](c OE) Combiner[A, A, O, OE]

func SimpleMerge[A Element, AM AccumulatorMerger[A]](c AM) Combiner[A, A, A, AM]

type Composite ¶

type Composite[O any] interface {
	Expand(s *Scope) O
}

Composite transforms allow structural re-use of sub pipelines.

type Counter ¶

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

func (c *Counter) Inc(dfc metricSource, diff int64)

type DFC ¶

type DFC[E Element] struct {
	// contains filtered or unexported fields
}

DFC is the DoFn Context for simple DoFns.

func (c *DFC[E]) Process(perElm Process[E]) error

func (c *DFC[E]) ToElmC(eventTime time.Time) ElmC

type Element ¶

type Element interface {
	any // Sadly, can't really restrict this without breaking iterators in GBK results.
}

type ElmC ¶

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

ElmC is the catch all context for the current element.

This includes * Key (state and timers) * Windows * Timestamp * Pane

Provides the downstream emission context, so it actually sends data to the next DoFn.

func (e *ElmC) EventTime() time.Time

type FullCombiner ¶

type FullCombiner[A, I, O Element] interface {
	InputAdder[A, I]
	AccumulatorMerger[A]
	OutputExtractor[A, O]
}

type InputAdder ¶

type InputAdder[A, I Element] interface {
	AddInput(A, I) A
	AccumulatorMerger[A]
}

InputAdder is an interface to allow combiners to incorporate an input type

type Iter ¶

type Iter[V Element] struct {
	// contains filtered or unexported fields
}

func (it *Iter[V]) All() func(perElm func(elm V) bool)

type KV ¶

type KV[K, V Element] struct {
	Key   K
	Value V
}

type Keys ¶

type Keys interface {
	comparable
}

type ObserveWindow ¶

type ObserveWindow struct{}

ObserveWindow indicates this DoFn needs to be aware of windows explicitly. Typical use is to embed ObserveWindows as a field.

func (*ObserveWindow) Of(ec ElmC) any

type OnBundleFinish ¶

type OnBundleFinish struct{}

OnBundleFinish allows a DoFn to register a function that runs just before a bundle finishes. Elements may be emitted downstream, if an ElmC is retrieved from the DFC.

func (*OnBundleFinish) Do(dfc bundleFinisher, finishBundle func() error)

type Options ¶

type Options = beamopts.Options

Options configure Run, ParDo, and Combine with specific features. Each function takes a variadic list of options, where properties set in later options override the value of previously set properties.

func Endpoint(endpoint string) Options

func Name(name string) Options

type Output ¶

type Output[E Element] struct {
	// contains filtered or unexported fields
}

Output represents an output of a DoFn.

At pipeline construction time, they represent an output PCollection, and can be connected as inputs to downstream DoFns.

At pipeline execution time, they are used in a ProcessBundle method to emit elements and pass along per element context, such as the EventTime and Window.

func CombinePerKey[K Keys, A, I, O Element, AM AccumulatorMerger[A]](s *Scope, input Output[KV[K, I]], comb Combiner[A, I, O, AM]) Output[KV[K, O]]

func Flatten[E Element](s *Scope, inputs ...Output[E]) Output[E]

func GBK[K Keys, V Element](s *Scope, input Output[KV[K, V]], opts ...Options) Output[KV[K, Iter[V]]]

func Impulse(s *Scope) Output[[]byte]

func Reshuffle[E Element](s *Scope, input Output[E], opts ...Options) Output[E]

func (emt *Output[E]) Emit(ec ElmC, elm E)

type OutputExtractor ¶

type OutputExtractor[A, O Element] interface {
	AccumulatorMerger[A]
	ExtractOutput(A) O
}

type Pipeline ¶

type Pipeline struct {
	Counters map[string]int64
}

Pipeline is a handle to a running or terminated pipeline for programmatic access to the given runner.

func Run(ctx context.Context, expand func(*Scope) error, opts ...Options) (Pipeline, error)

type Process ¶

type Process[E Element] func(ElmC, E) error

Process is the function type for handling a single element in a bundle.

Typically a closure returned from a Transform's ProcessBundle method.

Errors returned from Process functions abort bundle processing, and may cause pipeline termination. A runner may retry a bundle that has failed.

type Scope ¶

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

Scope is used for building pipeline graphs.

Scope is a hierarchical grouping for composite transforms. Scopes can be enclosed 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 for monitoring and visualization purposes.

func (s *Scope) String() string

type SideInputIter ¶

type SideInputIter[E Element] struct {
	// contains filtered or unexported fields
}

func AsSideIter[E Element](emt Output[E]) SideInputIter[E]

func (si *SideInputIter[E]) All(ec ElmC) func(perElm func(elm E) bool)

type SideInputMap ¶

type SideInputMap[K, V Element] struct {
	// contains filtered or unexported fields
}

SideInputMap allows a side input to be accessed via multip-map key lookups.

func AsSideMap[K, V Element](emt Output[KV[K, V]]) SideInputMap[K, V]

func (si *SideInputMap[K, V]) Get(ec ElmC, k K) func(perElm func(elm V) bool)

func (si *SideInputMap[K, V]) Keys(ec ElmC) func(perElm func(elm K) bool)

type StateBag ¶

type StateBag[E Element] struct {
	// contains filtered or unexported fields
}

type StateCombining ¶

type StateCombining[E Element] struct {
	// contains filtered or unexported fields
}

type StateMap ¶

type StateMap[K, V Element] struct {
	// contains filtered or unexported fields
}

type StateSet ¶

type StateSet[E Element] struct {
	// contains filtered or unexported fields
}

type StateValue ¶

type StateValue[E Element] struct {
	// contains filtered or unexported fields
}

type TimerEvent ¶

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

type TimerProcessing ¶

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

type Transform ¶

type Transform[E Element] interface {
	ProcessBundle(ctx context.Context, dfc *DFC[E]) error
}

Transform is the only interface that needs to be implemented by most DoFns.