bigtableio

Documentation

Overview

Package bigtableio provides transformations and utilities to interact with Google Bigtable. See also: https://cloud.google.com/bigtable/docs

Index

• func Write(s beam.Scope, project, instanceID, table string, col beam.PCollection)
• func WriteBatch(s beam.Scope, project, instanceID, table string, col beam.PCollection)
• type Mutation
  • func NewMutation(rowKey string) *Mutation
  • func (m *Mutation) Set(family, column string, ts bigtable.Timestamp, value []byte)
  • func (m *Mutation) WithGroupKey(key string) *Mutation
• type Operation

Examples

• WriteBatch

Functions

func Write

func Write(s beam.Scope, project, instanceID, table string, col beam.PCollection)

Write writes the elements of the given PCollection<bigtableio.Mutation> to bigtable.

func WriteBatch

func WriteBatch(s beam.Scope, project, instanceID, table string, col beam.PCollection)

WriteBatch writes the elements of the given PCollection<bigtableio.Mutation> to bigtable using bigtable.ApplyBulk(). For the underlying bigtable.ApplyBulk function to work properly the maximum number of operations per bigtableio.Mutation of the input PCollection must not be greater than 100,000. For more information see https://cloud.google.com/bigtable/docs/writes#batch for more.

Example

pipeline := beam.NewPipeline()
s := pipeline.Root()

//sample PBCollection<bigtableio.Mutation>
bigtableioMutationCol := beam.CreateList(s, func() []Mutation {
	columnFamilyName := "stats_summary"
	timestamp := bigtable.Now()

	// var muts []bigtableio.Mutation
	var muts []Mutation

	deviceA := "tablet"
	rowKeyA := deviceA + "#a0b81f74#20190501"

	// bigtableio.NewMutation(rowKeyA).WithGroupKey(deviceA)
	mutA := NewMutation(rowKeyA).WithGroupKey(deviceA) // this groups bundles by device identifiers
	mutA.Set(columnFamilyName, "connected_wifi", timestamp, []byte("1"))
	mutA.Set(columnFamilyName, "os_build", timestamp, []byte("12155.0.0-rc1"))

	muts = append(muts, *mutA)

	deviceB := "phone"
	rowKeyB := deviceB + "#a0b81f74#20190502"

	mutB := NewMutation(rowKeyB).WithGroupKey(deviceB)
	mutB.Set(columnFamilyName, "connected_wifi", timestamp, []byte("1"))
	mutB.Set(columnFamilyName, "os_build", timestamp, []byte("12145.0.0-rc6"))

	muts = append(muts, *mutB)

	return muts
}())

// bigtableio.WriteBatch(...)
WriteBatch(s, "project", "instanceId", "tableName", bigtableioMutationCol)

Types

type Mutation

type Mutation struct {
	RowKey string
	Ops    []Operation

	// optional custom beam.GroupByKey key, default is a fixed key of 1.
	GroupKey string
}

Mutation represents a necessary serializable wrapper analogue to bigtable.Mutation containing a rowKey and the operations to be applied.

func NewMutation

func NewMutation(rowKey string) *Mutation

NewMutation returns a new *Mutation, analogue to bigtable.NewMutation().

func (*Mutation) Set

func (m *Mutation) Set(family, column string, ts bigtable.Timestamp, value []byte)

Set sets a value in a specified column, with the given timestamp, analogue to bigtable.Mutation.Set(). The timestamp will be truncated to millisecond granularity. A timestamp of ServerTime means to use the server timestamp.

func (*Mutation) WithGroupKey

func (m *Mutation) WithGroupKey(key string) *Mutation

WithGroupKey sets a custom group key to be utilised by beam.GroupByKey.

type Operation

type Operation struct {
	Family string
	Column string
	Ts     bigtable.Timestamp
	Value  []byte
}

Operation represents a raw change to be applied within a Mutation.