worker

Documentation

Index

• Variables
• type DatabendWorkerStatsData
• type DatabendWorkerStatsRecorder
  • func NewDatabendWorkerStatsRecorder() *DatabendWorkerStatsRecorder
  • func (stats *DatabendWorkerStatsRecorder) RecordMetric(bytes int, rows int)
  • func (stats *DatabendWorkerStatsRecorder) Stats(statsWindow time.Duration) DatabendWorkerStatsData
• type Worker
  • func NewWorker(cfg *config.Config, name string, ig ingester.DatabendIngester, ...) *Worker
  • func (w *Worker) IsSplitAccordingMaxGoRoutine(minSplitKey, maxSplitKey, batchSize int64) bool
  • func (w *Worker) IsWorkerCorrect() (int, int, bool)
  • func (w *Worker) Run(ctx context.Context)
  • func (w *Worker) StepBatchByTimeSplitKey() error

Variables

var (
	AlreadyIngestRows  = 0
	AlreadyIngestBytes = 0
)

Types

type DatabendWorkerStatsData

type DatabendWorkerStatsData struct {
	BytesPerSecond float64
	RowsPerSecond  float64
}

type DatabendWorkerStatsRecorder

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

func NewDatabendWorkerStatsRecorder

func NewDatabendWorkerStatsRecorder() *DatabendWorkerStatsRecorder

func (*DatabendWorkerStatsRecorder) RecordMetric

func (stats *DatabendWorkerStatsRecorder) RecordMetric(bytes int, rows int)

func (*DatabendWorkerStatsRecorder) Stats

func (stats *DatabendWorkerStatsRecorder) Stats(statsWindow time.Duration) DatabendWorkerStatsData

type Worker

type Worker struct {
	Name string
	Cfg  *config.Config
	Ig   ingester.DatabendIngester
	Src  source.Sourcer
	// contains filtered or unexported fields
}

func NewWorker

func NewWorker(cfg *config.Config, name string, ig ingester.DatabendIngester, src source.Sourcer) *Worker

func (*Worker) IsSplitAccordingMaxGoRoutine

func (w *Worker) IsSplitAccordingMaxGoRoutine(minSplitKey, maxSplitKey, batchSize int64) bool

IsSplitAccordingMaxGoRoutine checks if the split key is according to the max go routine

func (*Worker) IsWorkerCorrect

func (w *Worker) IsWorkerCorrect() (int, int, bool)

func (*Worker) Run

func (w *Worker) Run(ctx context.Context)

func (*Worker) StepBatchByTimeSplitKey

func (w *Worker) StepBatchByTimeSplitKey() error