executor

Documentation

Index

• Constants
• Variables
• type Concurrent
  • func NewConcurrent(logger *zap.SugaredLogger, taskTypeMap map[pipeline.AssetType]Config, ...) (*Concurrent, error)
  • 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 Sequential
  • func (s Sequential) RunSingleTask(ctx context.Context, instance scheduler.TaskInstance) error

Constants

const (
	KeyPrinter contextKey = iota
	ContextLogger
)

Variables

var DefaultExecutorsV2 = map[pipeline.AssetType]Config{
	pipeline.AssetTypeBigqueryQuery: {
		scheduler.TaskInstanceTypeMain:        NoOpOperator{},
		scheduler.TaskInstanceTypeColumnCheck: NoOpOperator{},
	},
	pipeline.AssetTypeBigqueryTableSensor: {
		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{},
	},
	"dbt": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	pipeline.AssetTypeEmpty: {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	pipeline.AssetTypePostgresQuery: {
		scheduler.TaskInstanceTypeMain:        NoOpOperator{},
		scheduler.TaskInstanceTypeColumnCheck: NoOpOperator{},
		scheduler.TaskInstanceTypeCustomCheck: NoOpOperator{},
	},
	pipeline.AssetTypeRedshiftQuery: {
		scheduler.TaskInstanceTypeMain:        NoOpOperator{},
		scheduler.TaskInstanceTypeColumnCheck: NoOpOperator{},
		scheduler.TaskInstanceTypeCustomCheck: NoOpOperator{},
	},
	pipeline.AssetTypeMsSQLQuery: {
		scheduler.TaskInstanceTypeMain:        NoOpOperator{},
		scheduler.TaskInstanceTypeColumnCheck: NoOpOperator{},
		scheduler.TaskInstanceTypeCustomCheck: NoOpOperator{},
	},
	pipeline.AssetTypeSynapseQuery: {
		scheduler.TaskInstanceTypeMain:        NoOpOperator{},
		scheduler.TaskInstanceTypeColumnCheck: NoOpOperator{},
		scheduler.TaskInstanceTypeCustomCheck: NoOpOperator{},
	},
	pipeline.AssetTypePython: {
		scheduler.TaskInstanceTypeMain:        NoOpOperator{},
		scheduler.TaskInstanceTypeColumnCheck: NoOpOperator{},
	},
	"python.beta": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"python.legacy": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"s3.sensor.key_sensor": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	pipeline.AssetTypeSnowflakeQuery: {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	pipeline.AssetTypeSnowflakeQuerySensor: {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	"appsflyer.export.bq": {
		scheduler.TaskInstanceTypeMain: NoOpOperator{},
	},
	pipeline.AssetTypeIngestr: {
		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, error)

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 Sequential

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

func (Sequential) RunSingleTask

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