etl

Documentation

Index

• Constants
• Variables
• func Build(transform *Transform) ([]string, error)
• func Field(val *zed.Value, field string) (*zed.Value, error)
• func FieldAsInt(val *zed.Value, field string) (int64, error)
• func FieldAsString(val *zed.Value, field string) (string, error)
• func NewArrayFromReader(zr zio.Reader) (*zbuf.Array, error)
• type Pipeline
  • func NewPipeline(ctx context.Context, transform *Transform, service lakeapi.Interface) (*Pipeline, error)
  • func (p *Pipeline) Run(ctx context.Context) (int, error)
• type Pool
  • func OpenPool(ctx context.Context, poolName string, server lakeapi.Interface) (*Pool, error)
  • func (p *Pool) LoadBatch(ctx context.Context, zctx *zed.Context, batch *zbuf.Array) (ksuid.KSUID, error)
  • func (p *Pool) NextProducerOffsets(ctx context.Context) (map[string]int64, error)
  • func (p *Pool) Query(ctx context.Context, src string) (*zbuf.Array, error)
• type Route
• type Routes
  • func (r *Routes) InputsOf(output string) []string
  • func (r *Routes) LookupPool(topic string) string
  • func (r *Routes) Outputs() []string
• type Rule
• type Transform
  • func Load(path string) (*Transform, error)
  • func (t *Transform) Load(path string) error

Constants

const KafkaOffsetEarliest = -2

KafkaOffsetEarliest is used to begin consuming at the earliest (oldest) offset.

Variables

var ErrBadPoolKey = errors.New("pool key must be 'kafka.offset' in ascending order")

Functions

func Build

func Build(transform *Transform) ([]string, error)

func Field

func Field(val *zed.Value, field string) (*zed.Value, error)

func FieldAsInt

func FieldAsInt(val *zed.Value, field string) (int64, error)

func FieldAsString

func FieldAsString(val *zed.Value, field string) (string, error)

func NewArrayFromReader

func NewArrayFromReader(zr zio.Reader) (*zbuf.Array, error)

Types

type Pipeline

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

func NewPipeline

func NewPipeline(ctx context.Context, transform *Transform, service lakeapi.Interface) (*Pipeline, error)

func (*Pipeline) Run

func (p *Pipeline) Run(ctx context.Context) (int, error)

type Pool

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

func OpenPool

func OpenPool(ctx context.Context, poolName string, server lakeapi.Interface) (*Pool, error)

func (*Pool) LoadBatch

func (p *Pool) LoadBatch(ctx context.Context, zctx *zed.Context, batch *zbuf.Array) (ksuid.KSUID, error)

func (*Pool) NextProducerOffsets

func (p *Pool) NextProducerOffsets(ctx context.Context) (map[string]int64, error)

func (*Pool) Query

func (p *Pool) Query(ctx context.Context, src string) (*zbuf.Array, error)

type Route

type Route struct {
	Topic string `yaml:"topic"`
	Pool  string `yaml:"pool"`
}

type Routes

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

func (*Routes) InputsOf

func (r *Routes) InputsOf(output string) []string

func (*Routes) LookupPool

func (r *Routes) LookupPool(topic string) string

func (*Routes) Outputs

func (r *Routes) Outputs() []string

type Rule

type Rule struct {
	Type  string `yaml:"type"`
	In    string `yaml:"in"`
	Left  string `yaml:"left"`
	Right string `yaml:"right"`
	Join  string `yaml:"join-on"`
	Out   string `yaml:"out"`
	Where string `yaml:"where"`
	Zed   string `yaml:"zed"`
}

type Transform

type Transform struct {
	Inputs []Route `yaml:"inputs"`
	Output Route   `yaml:"output"`
	ETLs   []Rule  `yaml:"transforms"`
}

func Load

func Load(path string) (*Transform, error)

func (*Transform) Load

func (t *Transform) Load(path string) error