batchsplitter

README

Lets imagine we use BatchSplitter middleware and set MaxItemsPerBatch to 1000. The following diagram shows how does this middleware fit into the data transfer process(and how it interacts with bufferer in particular)

sequenceDiagram
    participant Souce
    participant Bufferer
    participant Middleware
    participant Sink
    Souce->>Bufferer: AsyncPush(changeItems [count = 1000])
    Souce->>Bufferer: AsyncPush(changeItems [count = 3000])
    Souce->>Bufferer: AsyncPush(changeItems [count = 700])
    Souce-->Sink: Trigger to push changes is activated
    Bufferer->>Middleware: Push(changeItems [count = 4700])
    Middleware->>Sink: Push(changeItems [count = 1000])
    Middleware->>Sink: Push(changeItems [count = 1000])
    Middleware->>Sink: Push(changeItems [count = 1000])
    Middleware->>Sink: Push(changeItems [count = 1000])
    Middleware->>Sink: Push(changeItems [count = 700])

Documentation

Index

• Constants
• Variables
• func PluggableBatchSplitterTransformer(transfer *model.Transfer, _ metrics.Registry, _ coordinator.Coordinator) func(abstract.Sinker) abstract.Sinker
• type BatchSplitterTransformer
  • func (t *BatchSplitterTransformer) Apply(input []abstract.ChangeItem) abstract.TransformerResult
  • func (t *BatchSplitterTransformer) Description() string
  • func (t *BatchSplitterTransformer) ResultSchema(original *abstract.TableSchema) (*abstract.TableSchema, error)
  • func (t *BatchSplitterTransformer) Suitable(abstract.TableID, *abstract.TableSchema) bool
  • func (t *BatchSplitterTransformer) Type() abstract.TransformerType
• type Config

Constants

const Type = abstract.TransformerType("batch_splitter")

Variables

var IdentityMiddleware = func(s abstract.Sinker) abstract.Sinker { return s }

Functions

func PluggableBatchSplitterTransformer

func PluggableBatchSplitterTransformer(transfer *model.Transfer, _ metrics.Registry, _ coordinator.Coordinator) func(abstract.Sinker) abstract.Sinker

Types

type BatchSplitterTransformer

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

func (*BatchSplitterTransformer) Apply

func (t *BatchSplitterTransformer) Apply(input []abstract.ChangeItem) abstract.TransformerResult

func (*BatchSplitterTransformer) Description

func (t *BatchSplitterTransformer) Description() string

func (*BatchSplitterTransformer) ResultSchema

func (t *BatchSplitterTransformer) ResultSchema(original *abstract.TableSchema) (*abstract.TableSchema, error)

func (*BatchSplitterTransformer) Suitable

func (t *BatchSplitterTransformer) Suitable(abstract.TableID, *abstract.TableSchema) bool

func (*BatchSplitterTransformer) Type

func (t *BatchSplitterTransformer) Type() abstract.TransformerType

type Config

type Config struct {
	MaxItemsPerBatch int `json:"max_items_per_batch"`
}