aggregate

README

Aggregation

This module implements KPL record aggregation.

https://github.com/awslabs/amazon-kinesis-producer/blob/master/aggregation-format.md

Generating the aggregate.pb.go file from aggregate.proto

Install protoc

https://developers.google.com/protocol-buffers/docs/downloads

Install protoc-gen-go

go get google.golang.org/protobuf/cmd/protoc-gen-g
go install google.golang.org/protobuf/cmd/protoc-gen-go

Install protobuf go library

go get github.com/golang/protobuf

Generating the protobuf go code

protoc -I=. --go_out=. aggregate.proto

Documentation

Index

• Variables
• type AggregatedRecord
  • func (*AggregatedRecord) Descriptor() ([]byte, []int)
  • func (x *AggregatedRecord) GetExplicitHashKeyTable() []string
  • func (x *AggregatedRecord) GetPartitionKeyTable() []string
  • func (x *AggregatedRecord) GetRecords() []*Record
  • func (*AggregatedRecord) ProtoMessage()
  • func (x *AggregatedRecord) ProtoReflect() protoreflect.Message
  • func (x *AggregatedRecord) Reset()
  • func (x *AggregatedRecord) String() string
• type Aggregator
  • func NewAggregator(stringGen *util.RandomStringGenerator, cfg *Config) *Aggregator
  • func (a *Aggregator) AddRecord(partitionKey string, hasPartitionKey bool, data []byte) (entry *kinesis.PutRecordsRequestEntry, err error)
  • func (a *Aggregator) AggregateRecords() (entry *kinesis.PutRecordsRequestEntry, err error)
  • func (a *Aggregator) GetRecordCount() int
• type Config
• type Record
  • func (*Record) Descriptor() ([]byte, []int)
  • func (x *Record) GetData() []byte
  • func (x *Record) GetExplicitHashKeyIndex() uint64
  • func (x *Record) GetPartitionKeyIndex() uint64
  • func (x *Record) GetTags() []*Tag
  • func (*Record) ProtoMessage()
  • func (x *Record) ProtoReflect() protoreflect.Message
  • func (x *Record) Reset()
  • func (x *Record) String() string
• type Tag
  • func (*Tag) Descriptor() ([]byte, []int)
  • func (x *Tag) GetKey() string
  • func (x *Tag) GetValue() string
  • func (*Tag) ProtoMessage()
  • func (x *Tag) ProtoReflect() protoreflect.Message
  • func (x *Tag) Reset()
  • func (x *Tag) String() string

Variables

var File_aggregate_aggregate_proto protoreflect.FileDescriptor

Types

type AggregatedRecord

type AggregatedRecord struct {
	PartitionKeyTable    []string  `protobuf:"bytes,1,rep,name=partition_key_table,json=partitionKeyTable" json:"partition_key_table,omitempty"`
	ExplicitHashKeyTable []string  `protobuf:"bytes,2,rep,name=explicit_hash_key_table,json=explicitHashKeyTable" json:"explicit_hash_key_table,omitempty"`
	Records              []*Record `protobuf:"bytes,3,rep,name=records" json:"records,omitempty"`
	// contains filtered or unexported fields
}

func (*AggregatedRecord) Descriptor

func (*AggregatedRecord) Descriptor() ([]byte, []int)

Deprecated: Use AggregatedRecord.ProtoReflect.Descriptor instead.

func (*AggregatedRecord) GetExplicitHashKeyTable

func (x *AggregatedRecord) GetExplicitHashKeyTable() []string

func (*AggregatedRecord) GetPartitionKeyTable

func (x *AggregatedRecord) GetPartitionKeyTable() []string

func (*AggregatedRecord) GetRecords

func (x *AggregatedRecord) GetRecords() []*Record

func (*AggregatedRecord) ProtoMessage

func (*AggregatedRecord) ProtoMessage()

func (*AggregatedRecord) ProtoReflect

func (x *AggregatedRecord) ProtoReflect() protoreflect.Message

func (*AggregatedRecord) Reset

func (x *AggregatedRecord) Reset()

func (*AggregatedRecord) String

func (x *AggregatedRecord) String() string

type Aggregator

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

Aggregator kinesis aggregator

func NewAggregator

func NewAggregator(stringGen *util.RandomStringGenerator, cfg *Config) *Aggregator

NewAggregator create a new aggregator

func (*Aggregator) AddRecord

func (a *Aggregator) AddRecord(partitionKey string, hasPartitionKey bool, data []byte) (entry *kinesis.PutRecordsRequestEntry, err error)

AddRecord to the aggregate buffer. Will return a kinesis PutRecordsRequest once buffer is full, or if the data exceeds the aggregate limit.

func (*Aggregator) AggregateRecords

func (a *Aggregator) AggregateRecords() (entry *kinesis.PutRecordsRequestEntry, err error)

AggregateRecords will flush proto-buffered records into a put request

func (*Aggregator) GetRecordCount

func (a *Aggregator) GetRecordCount() int

GetRecordCount gets number of buffered records

type Config

type Config struct {
	MaximumRecordSize *int
	MaxAggRecordSize  *int
}

Config is for aggregation customizations.

type Record

type Record struct {
	PartitionKeyIndex    *uint64 `protobuf:"varint,1,req,name=partition_key_index,json=partitionKeyIndex" json:"partition_key_index,omitempty"`
	ExplicitHashKeyIndex *uint64 `protobuf:"varint,2,opt,name=explicit_hash_key_index,json=explicitHashKeyIndex" json:"explicit_hash_key_index,omitempty"`
	Data                 []byte  `protobuf:"bytes,3,req,name=data" json:"data,omitempty"`
	Tags                 []*Tag  `protobuf:"bytes,4,rep,name=tags" json:"tags,omitempty"`
	// contains filtered or unexported fields
}

func (*Record) Descriptor

func (*Record) Descriptor() ([]byte, []int)

Deprecated: Use Record.ProtoReflect.Descriptor instead.

func (*Record) GetData

func (x *Record) GetData() []byte

func (*Record) GetExplicitHashKeyIndex

func (x *Record) GetExplicitHashKeyIndex() uint64

func (*Record) GetPartitionKeyIndex

func (x *Record) GetPartitionKeyIndex() uint64

func (*Record) GetTags

func (x *Record) GetTags() []*Tag

func (*Record) ProtoMessage

func (*Record) ProtoMessage()

func (*Record) ProtoReflect

func (x *Record) ProtoReflect() protoreflect.Message

func (*Record) Reset

func (x *Record) Reset()

func (*Record) String

func (x *Record) String() string

type Tag

type Tag struct {
	Key   *string `protobuf:"bytes,1,req,name=key" json:"key,omitempty"`
	Value *string `protobuf:"bytes,2,opt,name=value" json:"value,omitempty"`
	// contains filtered or unexported fields
}

func (*Tag) Descriptor

func (*Tag) Descriptor() ([]byte, []int)

Deprecated: Use Tag.ProtoReflect.Descriptor instead.

func (*Tag) GetKey

func (x *Tag) GetKey() string

func (*Tag) GetValue

func (x *Tag) GetValue() string

func (*Tag) ProtoMessage

func (*Tag) ProtoMessage()

func (*Tag) ProtoReflect

func (x *Tag) ProtoReflect() protoreflect.Message

func (*Tag) Reset

func (x *Tag) Reset()

func (*Tag) String

func (x *Tag) String() string