transformer

README

What is transformer

We can apply stateless transformation on our inflight data. This is based on our data model

Each batch of changes can be transformed into new batch of changes:

How to add new transformer

1. Create new package
2. Implemenet abstract.Transformer interface
3. Add proto mesage for transformer config
4. Register implementation

Example:

Implementation of abstract.Transformer

type DummyTransformer struct {
}

func (r *DummyTransformer) Apply(input []abstract.ChangeItem) abstract.TransformerResult {
	return abstract.TransformerResult{
		Transformed: input,
		Errors:      nil,
	}
}

func (r *DummyTransformer) Suitable(table abstract.TableID, schema abstract.TableColumns) bool {
	return true
}

func (r *DummyTransformer) ResultSchema(original abstract.TableColumns) abstract.TableColumns {
	return original
}

func (r *DummyTransformer) Description() string {
	return "this transformer do nothing"
}

Proto interface

message DummyTransformer {
	// here some params can be placed
}

Registration

func init() {
	transformer.Register(new(api.DummyTransformer), func(protoConfig any, lgr log.Logger) (abstract.Transformer, error) {
		return &DummyTransformer{}, nil
	})
}

Prefillment on transfer create

If you want your transformers to prefill on transfer create, you should insert your typeswitch in this function called prefillTransformationDefaults in transfer_service.go: https://a.yandex-team.ru/arcadia/transfer_manager/go/pkg/controlplane/transfer_service.go?rev=r11799915#L127

Documentation

Index

• Constants
• func IsKnownTransformerType(t abstract.TransformerType) bool
• func KnownTransformerNames() []string
• func New(typ abstract.TransformerType, cfg Config, lgr log.Logger, ...) (abstract.Transformer, error)
• func Register[TConfig Config](typ abstract.TransformerType, ...)
• func Sinker(config *Transformers, runtime abstract.TransformationRuntimeOpts, ...) abstract.SinkOption
• type Config
• type ErrorsOutput
• type OutputType
• type Transformer
  • func (t Transformer) Config() Config
  • func (t Transformer) ID() string
  • func (t Transformer) Type() abstract.TransformerType
• type TransformerFactory
• type Transformers
• type WithID

Constants

const (
	SinkErrorsOutput    = OutputType("sink")
	DevnullErrorsOutput = OutputType("devnull")
)

const ID = "transformerId"

Functions

func IsKnownTransformerType

func IsKnownTransformerType(t abstract.TransformerType) bool

func KnownTransformerNames

func KnownTransformerNames() []string

func New

func New(typ abstract.TransformerType, cfg Config, lgr log.Logger, rt abstract.TransformationRuntimeOpts) (abstract.Transformer, error)

func Register

func Register[TConfig Config](typ abstract.TransformerType, f func(cfg TConfig, lgr log.Logger, runtime abstract.TransformationRuntimeOpts) (abstract.Transformer, error))

func Sinker

func Sinker(
	config *Transformers,
	runtime abstract.TransformationRuntimeOpts,
	transformers []abstract.Transformer,
	lgr log.Logger,
	registry metrics.Registry,
) abstract.SinkOption

Types

type Config

type Config any

type ErrorsOutput

type ErrorsOutput struct {
	Type   OutputType
	Config any
}

type OutputType

type OutputType string

type Transformer

type Transformer map[abstract.TransformerType]interface{}

Transformer serializeable one-of wrapper, for backward compatibility will hold both type discriminator and config value itself, config is just an any object on later stages it materializes as real config object, specific to transformer

func (Transformer) Config

func (t Transformer) Config() Config

func (Transformer) ID

func (t Transformer) ID() string

func (Transformer) Type

func (t Transformer) Type() abstract.TransformerType

type TransformerFactory

type TransformerFactory func(protoConfig any, lgr log.Logger, runtime abstract.TransformationRuntimeOpts) (abstract.Transformer, error)

type Transformers

type Transformers struct {
	DebugMode    bool          `json:"debugMode"`
	Transformers []Transformer `json:"transformers"`
	ErrorsOutput *ErrorsOutput `json:"errorsOutput"`
}

type WithID

type WithID interface {
	ID() string
}