primitives

package
v2.40.0 Latest Latest
Warning

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

Go to latest
Published: Jun 23, 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.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL