primitives

package
v2.40.0-RC1 Latest Latest
Warning

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

 Go to latest Published: Jun 19, 2022 License: Apache-2.0, BSD-3-Clause, MIT Imports: 16 Imported by: 0

Documentation

Overview

Package primitives contains integration tests for primitives in beam.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Checkpoints added in v2.40.0 

func Checkpoints(s beam.Scope)

Checkpoints is a small test pipeline to establish the correctness of the simple test case.

func CoGBK 

func CoGBK() *beam.Pipeline

CoGBK tests CoGBK.

func CreateAndSplit added in v2.35.0 

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

CreateAndSplit makes a KV PCollection from a list of stringPair types

func Drain added in v2.40.0 

func Drain(s beam.Scope)

Drain tests the SDF truncation during drain.

func Flatten 

func Flatten() *beam.Pipeline

Flatten tests flatten.

func FlattenDup 

func FlattenDup() *beam.Pipeline

FlattenDups tests flatten with the same input multiple times.

func Panes added in v2.37.0 

func Panes(s beam.Scope)

Panes constructs a teststream and applies a pardo to get the pane timings.

func PanesFn added in v2.37.0 

func PanesFn(pn beam.PaneInfo, value float64, emit func(int))

PanesFn is DoFn that simply emits the pane timing value.

func ParDoKVSideInput 

func ParDoKVSideInput() *beam.Pipeline

ParDoKVSideInput computes the sum of ints using a KV side input.

func ParDoMultiMapSideInput added in v2.35.0 

func ParDoMultiMapSideInput() *beam.Pipeline

ParDoMultiMapSideInput checks that the multimap side input access pattern works correctly, properly producing the correct output with an asymmetric join.

func ParDoMultiOutput 

func ParDoMultiOutput() *beam.Pipeline

ParDoMultiOutput test a DoFn with multiple output.

func ParDoPipelineOptions added in v2.37.0 

func ParDoPipelineOptions() *beam.Pipeline

ParDoPipelineOptions creates a pipeline with flag options to validate that a DoFn can access them as PipelineOptions.

func ParDoSideInput 

func ParDoSideInput() *beam.Pipeline

ParDoSideInput computes the sum of ints using a side input.

func Reshuffle 

func Reshuffle() *beam.Pipeline

Reshuffle tests Reshuffle.

func ReshuffleKV 

func ReshuffleKV() *beam.Pipeline

ReshuffleKV tests Reshuffle with KV PCollections.

func TestStreamBoolSequence 

func TestStreamBoolSequence() *beam.Pipeline

TestStreamBoolSequence tests the TestStream primitive by inserting boolean elements then advancing the watermark past the point where they were inserted.

func TestStreamByteSliceSequence 

func TestStreamByteSliceSequence() *beam.Pipeline

TestStreamByteSliceSequence tests the TestStream primitive by inserting byte slice elements then advancing the watermark to infinity and comparing the output..

func TestStreamFloat64Sequence 

func TestStreamFloat64Sequence() *beam.Pipeline

TestStreamFloat64Sequence tests the TestStream primitive by inserting float64 elements then advancing the watermark past the point where they were inserted.

func TestStreamInt64Sequence 

func TestStreamInt64Sequence() *beam.Pipeline

TestStreamInt64Sequence tests the TestStream primitive by inserting int64 elements then advancing the watermark past the point where they were inserted.

func TestStreamStrings 

func TestStreamStrings() *beam.Pipeline

TestStreamSequence tests the TestStream primitive by inserting string elements then advancing the watermark past the point where they were inserted.

func TestStreamTwoBoolSequences 

func TestStreamTwoBoolSequences() *beam.Pipeline

TestStreamTwoBoolSequences tests the TestStream primitive by inserting two sets of boolean elements that arrive on-time into the TestStream

func TestStreamTwoFloat64Sequences 

func TestStreamTwoFloat64Sequences() *beam.Pipeline

TestStreamTwoFloat64Sequences tests the TestStream primitive by inserting two sets of float64 elements that arrive on-time into the TestStream

func TestStreamTwoInt64Sequences 

func TestStreamTwoInt64Sequences() *beam.Pipeline

TestStreamTwoInt64Sequences tests the TestStream primitive by inserting two sets of int64 elements that arrive on-time into the TestStream

func TriggerAfterAll added in v2.40.0 

func TriggerAfterAll(s beam.Scope)

TriggerAfterAll tests AfterAll trigger. The output pane is fired when all triggers in the subtriggers are ready. In this test, since trigger.AfterCount(int32(5)) won't be ready unless we see 5 elements, trigger.Always() won't fire until we meet that condition. So we fire only once when we see the 5th element.

func TriggerAfterAny added in v2.40.0 

func TriggerAfterAny(s beam.Scope)

TriggerAfterAny tests AfterAny trigger. In this test, trigger.Always() gets ready everytime. So we would expect panes to be fired at every element irrespective of checking for other triggers.

func TriggerAfterEach added in v2.40.0 

func TriggerAfterEach(s beam.Scope)

TriggerAfterEach tests AfterEach trigger. The output pane is fired after each trigger is ready in the order set in subtriggers. In this test, since trigger.AfterCount(int32(3)) is first, first pane is fired after 3 elements, then a pane is fired each for trigger.Always() for element 5.0 and 8.0

func TriggerAfterEndOfWindow added in v2.34.0 

func TriggerAfterEndOfWindow(s beam.Scope)

TriggerAfterEndOfWindow tests the AfterEndOfWindow Trigger. With AfterCount(2) as the early firing trigger and AfterCount(1) as late firing trigger. It fires two times, one with early firing when there are two elements while the third elements waits in. This third element is fired in the late firing.

func TriggerAfterProcessingTime 

func TriggerAfterProcessingTime(s beam.Scope)

TriggerAfterProcessingTime tests the AfterProcessingTime Trigger, it fires output panes once 't' processing time has passed Not yet supported by the flink runner: java.lang.UnsupportedOperationException: Advancing Processing time is not supported by the Flink Runner.

func TriggerAfterSynchronizedProcessingTime added in v2.40.0 

func TriggerAfterSynchronizedProcessingTime(s beam.Scope)

TriggerAfterSynchronizedProcessingTime tests AfterSynchronizedProcessingTime trigger. It fires at the window expiration since the times doesn't synchronize in this test case.

func TriggerAlways 

func TriggerAlways(s beam.Scope)

TriggerAlways tests the Always trigger, it is expected to receive every input value as the output.

func TriggerDefault 

func TriggerDefault(s beam.Scope)

TriggerDefault tests the default trigger which fires the pane after the end of the window

func TriggerElementCount 

func TriggerElementCount(s beam.Scope)

TriggerElementCount tests the ElementCount Trigger, it waits for atleast N elements to be ready to fire an output pane

func TriggerNever added in v2.40.0 

func TriggerNever(s beam.Scope)

TriggerNever tests Never Trigger. It fires at the window expiration.

func TriggerOrFinally added in v2.40.0 

func TriggerOrFinally(s beam.Scope)

TriggerOrFinally tests OrFinally trigger. The main trigger in this test case trigger.Always() is always ready. But the output is produced only when finally trigger is ready. So it is ready at second element in first window and produces two output panes. Similarly, for the second window.

func TriggerRepeat 

func TriggerRepeat(s beam.Scope)

TriggerRepeat tests the repeat trigger. As of now is it is configure to take only one trigger as a subtrigger. In the below test, it is expected to receive three output panes with two elements each.

func ValidateWindowedSideInputs added in v2.35.0 

func ValidateWindowedSideInputs(s beam.Scope)

ValidateWindowedSideInputs checks that side inputs have accurate windowing information when used.

func WindowSums 

func WindowSums(s beam.Scope, sumPerKey func(beam.Scope, beam.PCollection) beam.PCollection)

WindowSums produces a pipeline that generates the numbers of a 3x3 magic square, and configures the pipeline so that PCollection. Sum is a closure to handle summing data over the window, in a few conditions.

func WindowSums_GBK 

func WindowSums_GBK(s beam.Scope)

func WindowSums_Lifted 

func WindowSums_Lifted(s beam.Scope)

Types

type RangeEstimator added in v2.40.0 

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

RangeEstimator implements the offsetrange.RangeEndEstimator interface. It provides the estimated end for a restriction.

func (*RangeEstimator) Estimate added in v2.40.0 

func (r *RangeEstimator) Estimate() int64

Estimate returns the estimated end.

func (*RangeEstimator) SetEstimate added in v2.40.0 

func (r *RangeEstimator) SetEstimate(estimate int64)

SetEstimate sets the estimated end.

type TruncateFn added in v2.40.0 

type TruncateFn struct {
	Estimator RangeEstimator
}

TruncateFn is an SDF.

func (*TruncateFn) CreateInitialRestriction added in v2.40.0 

func (fn *TruncateFn) CreateInitialRestriction(_ []byte) offsetrange.Restriction

CreateInitialRestriction creates an initial restriction

func (*TruncateFn) CreateTracker added in v2.40.0 

func (fn *TruncateFn) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker

CreateTracker wraps the given restriction into a LockRTracker type.

func (*TruncateFn) ProcessElement added in v2.40.0 

func (fn *TruncateFn) ProcessElement(rt *sdf.LockRTracker, _ []byte, emit func(int64)) sdf.ProcessContinuation

ProcessElement continually gets the start position of the restriction and emits the element as it is.

func (*TruncateFn) RestrictionSize added in v2.40.0 

func (fn *TruncateFn) RestrictionSize(_ []byte, rest offsetrange.Restriction) float64

RestrictionSize returns the size of the current restriction

func (*TruncateFn) SplitRestriction added in v2.40.0 

func (fn *TruncateFn) SplitRestriction(_ []byte, rest offsetrange.Restriction) []offsetrange.Restriction

SplitRestriction is similar to the one used in checkpointing.go test.

func (*TruncateFn) TruncateRestriction added in v2.40.0 

func (fn *TruncateFn) TruncateRestriction(rt *sdf.LockRTracker, _ []byte) offsetrange.Restriction

TruncateRestriction truncates the restriction during drain.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.