etl

Oct 6, 2022

Changes in this version

+ const KafkaOffsetEarliest

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

+ 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 struct

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

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

+ type Pool struct

+ 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 struct

+ Pool string

+ Topic string

+ type Routes struct

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

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

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

+ type Rule struct

+ In string

+ Join string

+ Left string

+ Out string

+ Right string

+ Type string

+ Where string

+ Zed string

+ type Transform struct

+ ETLs []Rule

+ Inputs []Route

+ Output Route

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

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