sink

Documentation

Overview

Package sink provides a way for clients to write data to BigQuery through the usage of a streaming metaphor.

It is the responsibility of the client code to provide the schema and other specifics of the BigQuery dataset and table, and then this package handles the actual writing of data.

Currently, real streaming is not implemented, and this package rather caches rows in memory before flushing to BigQuery when the calling code sends a signal.

Index

• func Start(ctx context.Context, opts ...Option)
• type Option
  • func WithBigQuery(projectID, datasetID string) Option
  • func WithErrorChannel(errChan chan error) Option
  • func WithMetrics(m metrics.Metrics) Option
  • func WithTableOperations(op TableOperations) Option
• type Schema
• type SourceStream
  • func Stream(typ string, schema Schema) SourceStream
• type TableOperations

Functions

func Start

func Start(ctx context.Context, opts ...Option)

Start the internal functionality by creating and starting an internal handler per incoming streamImpl. Each handler is started in a separate go routine.

Types

type Option

type Option func(collector *optionsCollector)

Option for configuring this package.

func WithBigQuery

func WithBigQuery(projectID, datasetID string) Option

WithBigQuery initializes this package with the information needed to access Google BigQuery.

func WithErrorChannel

func WithErrorChannel(errChan chan error) Option

WithErrorChannel sets a channel that this module will use to communicate all errors out.

func WithMetrics

func WithMetrics(m metrics.Metrics) Option

WithMetrics initializes this package with the metrics service.

func WithTableOperations

func WithTableOperations(op TableOperations) Option

WithTableOperations sets the interface that is used to write to BigQuery internally. The point is to provide a way by which this package can be unit tested. This function should not be used in production.

type Schema

type Schema struct {
	BQSchema    *bigquery.TableMetadata
	Disposition bigquery.TableWriteDisposition
}

Schema wraps the BigQuery schema and write disposition.

type SourceStream

type SourceStream interface {
	// Type is the type of the stream, usually the same as the table that the data is written to in BigQuery.
	Type() string

	// Send sends the given value on the stream.
	Send(v bigquery.ValueSaver)

	// SendAll sends all the elements in the list on the stream.
	SendAll(v []bigquery.ValueSaver)

	// Flush writes all elements currently held in memory to BigQuery.
	Flush()

	// Complete sends the signal that the stream is now complete for this iteration to the receiver.
	Complete()
}

SourceStream is the streamImpl that the source of the data that shall be written to BigQuery uses in order to communicate with this package.

func Stream

func Stream(typ string, schema Schema) SourceStream

Stream creates and returns a streamImpl that client code chan be used to streamImpl objects that shall be written to BigQuery. This function has the side effect of caching the corresponding target streamImpl internally so that changes are handled when this package is started.

type TableOperations

type TableOperations interface {
	// Write writes the given value to the table in BigQuery
	Write(ctx context.Context, table *bigquery.Table, rows []bigquery.ValueSaver) error

	// CreateTable creates the table in BigQuery.
	CreateTable(ctx context.Context, dataset string, schema Schema) (*bigquery.Table, error)

	// CopyTable copies the content of the source table to the destination table.
	CopyTable(ctx context.Context, source, dest *bigquery.Table) error

	// DeleteTable deletes the table in BigQuery.
	DeleteTable(ctx context.Context, table *bigquery.Table) error

	// TableRef creates and returns a reference to the table with the given schema and schema. The name of the table
	// is embedded in the schema.
	TableRef(dataset string, schema Schema) *bigquery.Table
}

TableOperations provides an abstraction for the concrete operations that happen against BigQuery.