executor

Documentation

Index

• Constants
• Variables
• type Concurrent
  • func NewConcurrent(logger *zap.SugaredLogger, taskTypeMap map[string]Operator, workerCount int) *Concurrent
  • func (c Concurrent) Start(input chan *scheduler.TaskInstance, ...)
• type NoOpOperator
  • func (e NoOpOperator) RunTask(ctx context.Context, p *pipeline.Pipeline, t *pipeline.Task) error
• type Operator
• type Sequential
  • func (s Sequential) RunSingleTask(ctx context.Context, pipeline *pipeline.Pipeline, task *pipeline.Task) error

Constants

const (
	TaskTypePython         = "python"
	TaskTypeSnowflakeQuery = "sf.sql"
	TaskTypeBigqueryQuery  = "bq.sql"
	TaskTypeEmpty          = "empty"
)

const (
	KeyPrinter contextKey = iota
)

Variables

var DefaultExecutors = map[string]Operator{
	TaskTypeBigqueryQuery:                  NoOpOperator{},
	"bq.sensor.table":                      NoOpOperator{},
	"bq.sensor.query":                      NoOpOperator{},
	"bq.cost_tracker":                      NoOpOperator{},
	"bq.transfer":                          NoOpOperator{},
	"bq.sensor.partition":                  NoOpOperator{},
	"gcs.from.s3":                          NoOpOperator{},
	"gcs.sensor.object_sensor_with_prefix": NoOpOperator{},
	"gcs.sensor.object":                    NoOpOperator{},
	TaskTypeEmpty:                          NoOpOperator{},
	"athena.sql":                           NoOpOperator{},
	"athena.sensor.query":                  NoOpOperator{},
	TaskTypePython:                         NoOpOperator{},
	"python.beta":                          NoOpOperator{},
	"python.legacy":                        NoOpOperator{},
	"s3.sensor.key_sensor":                 NoOpOperator{},
	TaskTypeSnowflakeQuery:                 NoOpOperator{},
	"adjust.export.bq":                     NoOpOperator{},
	"appsflyer.export.bq":                  NoOpOperator{},
}

DefaultExecutors are set to NoOp executors by default. It serves both as a safe default and a list of supported task types.

Types

type Concurrent

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

func NewConcurrent

func NewConcurrent(
	logger *zap.SugaredLogger,
	taskTypeMap map[string]Operator,
	workerCount int,
) *Concurrent

func (Concurrent) Start

func (c Concurrent) Start(input chan *scheduler.TaskInstance, result chan<- *scheduler.TaskExecutionResult)

type NoOpOperator

type NoOpOperator struct{}

func (NoOpOperator) RunTask

func (e NoOpOperator) RunTask(ctx context.Context, p *pipeline.Pipeline, t *pipeline.Task) error

type Operator

type Operator interface {
	RunTask(ctx context.Context, p *pipeline.Pipeline, t *pipeline.Task) error
}

type Sequential

type Sequential struct {
	TaskTypeMap map[string]Operator
}

func (Sequential) RunSingleTask

func (s Sequential) RunSingleTask(ctx context.Context, pipeline *pipeline.Pipeline, task *pipeline.Task) error