pipeline

Documentation

Index

• func NewBuilder(config BuilderConfig, yamlTaskCreator TaskCreator, ...) *builder
• type Asset
  • func ConvertYamlToTask(content []byte) (*Asset, error)
• type AssetType
• type BuilderConfig
• type Column
• type ColumnCheck
• type ColumnCheckValue
• type DefinitionFile
• type ExecutableFile
• type Materialization
• type MaterializationStrategy
• type MaterializationType
• type Notifications
• type Pipeline
  • func (p Pipeline) HasTaskType(taskType AssetType) bool
  • func (p *Pipeline) RelativeTaskPath(t *Asset) string
• type SlackNotification
• type TaskCreator
  • func CreateTaskFromFileComments(fs afero.Fs) TaskCreator
  • func CreateTaskFromYamlDefinition(fs afero.Fs) TaskCreator
• type TaskDefinitionFile
• type TaskDefinitionType
• type TaskSchedule

Functions

func NewBuilder

func NewBuilder(config BuilderConfig, yamlTaskCreator TaskCreator, commentTaskCreator TaskCreator, fs afero.Fs) *builder

Types

type Asset

type Asset struct {
	Name            string
	Description     string
	Type            AssetType
	ExecutableFile  ExecutableFile
	DefinitionFile  TaskDefinitionFile
	Parameters      map[string]string
	Connections     map[string]string
	DependsOn       []string
	Pipeline        *Pipeline
	Schedule        TaskSchedule
	Materialization Materialization
	Columns         map[string]Column
}

func ConvertYamlToTask

func ConvertYamlToTask(content []byte) (*Asset, error)

type AssetType

type AssetType string

type BuilderConfig

type BuilderConfig struct {
	PipelineFileName    string
	TasksDirectoryName  string
	TasksDirectoryNames []string
	TasksFileSuffixes   []string
}

type Column

type Column struct {
	Name        string
	Description string        `yaml:"description"`
	Checks      []ColumnCheck `yaml:"checks"`
}

type ColumnCheck

type ColumnCheck struct {
	Name  string `yaml:"name"`
	Value ColumnCheckValue
}

type ColumnCheckValue

type ColumnCheckValue struct {
	IntArray    *[]int
	Int         *int
	Float       *float64
	StringArray *[]string
	String      *string
}

type DefinitionFile

type DefinitionFile struct {
	Name string
	Path string
}

type ExecutableFile

type ExecutableFile struct {
	Name    string
	Path    string
	Content string
}

type Materialization

type Materialization struct {
	Type           MaterializationType
	Strategy       MaterializationStrategy
	PartitionBy    string
	ClusterBy      []string
	IncrementalKey string
}

type MaterializationStrategy

type MaterializationStrategy string

const (
	MaterializationStrategyNone          MaterializationStrategy = ""
	MaterializationStrategyCreateReplace MaterializationStrategy = "create+replace"
	MaterializationStrategyDeleteInsert  MaterializationStrategy = "delete+insert"
	MaterializationStrategyAppend        MaterializationStrategy = "append"
)

type MaterializationType

type MaterializationType string

const (
	MaterializationTypeNone  MaterializationType = ""
	MaterializationTypeView  MaterializationType = "view"
	MaterializationTypeTable MaterializationType = "table"
)

type Notifications

type Notifications struct {
	Slack []SlackNotification
}

type Pipeline

type Pipeline struct {
	LegacyID           string   `yaml:"id"`
	Name               string   `yaml:"name"`
	Schedule           schedule `yaml:"schedule"`
	StartDate          string   `yaml:"start_date"`
	DefinitionFile     DefinitionFile
	DefaultParameters  map[string]string `yaml:"default_parameters"`
	DefaultConnections map[string]string `yaml:"default_connections"`
	Tasks              []*Asset
	Notifications      Notifications `yaml:"notifications"`

	TasksByType map[AssetType][]*Asset
	// contains filtered or unexported fields
}

func (Pipeline) HasTaskType

func (p Pipeline) HasTaskType(taskType AssetType) bool

func (*Pipeline) RelativeTaskPath

func (p *Pipeline) RelativeTaskPath(t *Asset) string

type SlackNotification

type SlackNotification struct {
	Name       string
	Connection string
	Success    string
	Failure    string
}

type TaskCreator

type TaskCreator func(path string) (*Asset, error)

func CreateTaskFromFileComments

func CreateTaskFromFileComments(fs afero.Fs) TaskCreator

func CreateTaskFromYamlDefinition

func CreateTaskFromYamlDefinition(fs afero.Fs) TaskCreator

type TaskDefinitionFile

type TaskDefinitionFile struct {
	Name string
	Path string
	Type TaskDefinitionType
}

type TaskDefinitionType

type TaskDefinitionType string

const (
	CommentTask TaskDefinitionType = "comment"
	YamlTask    TaskDefinitionType = "yaml"
)

type TaskSchedule

type TaskSchedule struct {
	Days []string
}