teststream

Documentation

Overview

Package teststream contains code configuring the TestStream primitive for use in testing code that is meant to be run on streaming data sources.

See https://beam.apache.org/blog/test-stream/ for more information.

TestStream is supported on the Flink runner and currently supports int64, float64, and boolean types.

TODO(BEAM-12753): Flink currently displays unexpected behavior with TestStream, should not be used until this issue is resolved.

Index

• func Create(s beam.Scope, c Config) beam.PCollection
• func CreateWithEndpoint(s beam.Scope, url string, elementType beam.FullType) beam.PCollection
• type Config
  • func NewConfig() Config
  • func (c *Config) AddElementList(timestamp int64, elements interface{}) error
  • func (c *Config) AddElements(timestamp int64, elements ...interface{}) error
  • func (c *Config) AdvanceProcessingTime(duration int64)
  • func (c *Config) AdvanceProcessingTimeToInfinity()
  • func (c *Config) AdvanceWatermark(timestamp int64) error
  • func (c *Config) AdvanceWatermarkToInfinity() error

Functions

func Create

func Create(s beam.Scope, c Config) beam.PCollection

Create inserts a TestStream primitive into a pipeline, taking a scope and a Config object and producing an output PCollection. The TestStream must be the first PTransform in the pipeline.

func CreateWithEndpoint

func CreateWithEndpoint(s beam.Scope, url string, elementType beam.FullType) beam.PCollection

CreateWithEndpoint inserts a TestStream primitive into a pipeline, taking a scope, a url to a TestStreamService, and a FullType object describing the elements that will be returned by the TestStreamService. Authentication is currently not supported, so the service the URL points to should be accessed in a trusted context.

Types

type Config

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

Config holds information used to create a TestStreamPayload object.

func NewConfig

func NewConfig() Config

NewConfig returns a Config to build a sequence of a test stream's events. Requires that users provide the coder for the elements they are trying to emit.

func (*Config) AddElementList

func (c *Config) AddElementList(timestamp int64, elements interface{}) error

AddElementList inserts a slice of elements into the stream at the specified event timestamp. Must be called with at least one element.

Calls into AddElements, which panics if an inserted type does not match a previously inserted element type.

Element types must have built-in coders in Beam.

func (*Config) AddElements

func (c *Config) AddElements(timestamp int64, elements ...interface{}) error

AddElements adds a number of elements to the stream at the specified event timestamp. Must be called with at least one element.

On the first call, a type will be inferred from the passed in elements, which must be of all the same type. Type mismatches on this or subsequent calls will cause AddElements to return an error.

Element types must have built-in coders in Beam.

func (*Config) AdvanceProcessingTime

func (c *Config) AdvanceProcessingTime(duration int64)

AdvanceProcessingTime adds an event advancing the processing time by a given duration. This advancement is applied to all of the PCollections output by the TestStream.

func (*Config) AdvanceProcessingTimeToInfinity

func (c *Config) AdvanceProcessingTimeToInfinity()

AdvanceProcessingTimeToInfinity moves the TestStream processing time to the largest possible timestamp.

func (*Config) AdvanceWatermark

func (c *Config) AdvanceWatermark(timestamp int64) error

AdvanceWatermark adds an event to the Config Events struct advancing the watermark for the PCollection to the given timestamp. Timestamp is in milliseconds

func (*Config) AdvanceWatermarkToInfinity

func (c *Config) AdvanceWatermarkToInfinity() error

AdvanceWatermarkToInfinity advances the watermark to the maximum timestamp.