executor

Documentation

Index

• Constants
• Variables
• type Concurrent
  • func NewConcurrent(logger *zap.SugaredLogger, taskTypeMap map[pipeline.AssetType]Config, ...) *Concurrent
  • func (c Concurrent) Start(input chan scheduler.TaskInstance, ...)
• type Config
• type NoOpOperator
  • func (e NoOpOperator) Run(ctx context.Context, ti scheduler.TaskInstance) error
• type Operator
• type OperatorMap
• type OperatorSecondMap
• type Sequential
  • func (s Sequential) RunSingleTask(ctx context.Context, instance scheduler.TaskInstance) error

Constants

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

const (
	KeyPrinter contextKey = iota
)

Variables

var DefaultExecutorsV2 = map[pipeline.AssetType]Config{
	TaskTypeBigqueryQuery: {
		scheduler.TaskInstanceTypeMain:        NoOpOperator{},
		scheduler.TaskInstanceTypeColumnCheck: NoOpOperator{},
	},
	"bq.sensor.table": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"bq.sensor.query": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"bq.cost_tracker": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"bq.transfer": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"bq.sensor.partition": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"gcs.from.s3": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"gcs.delete": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"gcs.sensor.object_sensor_with_prefix": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"gcs.sensor.object": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	TaskTypeEmpty: {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"athena.sql": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"athena.sensor.query": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	TaskTypePython: {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"python.beta": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"python.legacy": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"s3.sensor.key_sensor": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	TaskTypeSnowflakeQuery: {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"adjust.export.bq": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"appsflyer.export.bq": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
}

Types

type Concurrent

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

func NewConcurrent

func NewConcurrent(
	logger *zap.SugaredLogger,
	taskTypeMap map[pipeline.AssetType]Config,
	workerCount int,
) *Concurrent

func (Concurrent) Start

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

type Config

type Config map[scheduler.TaskInstanceType]Operator

type NoOpOperator

type NoOpOperator struct{}

func (NoOpOperator) Run

func (e NoOpOperator) Run(ctx context.Context, ti scheduler.TaskInstance) error

type Operator

type Operator interface {
	Run(ctx context.Context, ti scheduler.TaskInstance) error
}

type OperatorMap

type OperatorMap map[pipeline.AssetType]Operator

type OperatorSecondMap

type OperatorSecondMap map[string]Operator

type Sequential

type Sequential struct {
	TaskTypeMap map[pipeline.AssetType]Config
}

func (Sequential) RunSingleTask

func (s Sequential) RunSingleTask(ctx context.Context, instance scheduler.TaskInstance) error