execution

Documentation

Overview

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2018 StreamSets Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Constants
• func NewPreviewOutput(batchOutputs [][]StageOutput) ([][]StageOutputJson, error)
• type Config
  • func NewConfig() Config
• type PreviewOutput
• type Previewer
• type Runner
• type SourceOffsetTracker
• type StageOutput
  • func NewStageOutput(stageContext api.StageContext, stageOutputJson StageOutputJson) (*StageOutput, error)
• type StageOutputJson
  • func NewStageOutputJson(stageOutput StageOutput) (*StageOutputJson, error)

Constants

const (
	DefaultMaxBatchSize = 1000
)

Functions

func NewPreviewOutput

func NewPreviewOutput(batchOutputs [][]StageOutput) ([][]StageOutputJson, error)

Types

type Config

type Config struct {
	MaxBatchSize int `toml:"max-batch-size"`
}

func NewConfig

func NewConfig() Config

NewConfig returns a new Config with default settings.

type PreviewOutput

type PreviewOutput struct {
	PreviewStatus string              `json:"status"`
	Issues        *validation.Issues  `json:"issues"`
	Output        [][]StageOutputJson `json:"batchesOutput"`
	Message       string              `json:"message"`
}

type Previewer

type Previewer interface {
	GetId() string
	ValidateConfigs(timeoutMillis int64) error
	Start(
		batches int,
		batchSize int,
		skipTargets bool,
		stopStage string,
		stagesOverride []StageOutputJson,
		timeoutMillis int64,
		testOrigin bool,
	) error
	Stop() error
	GetStatus() string
	GetOutput() PreviewOutput
}

type Runner

type Runner interface {
	GetPipelineConfig() common.PipelineConfiguration
	GetStatus() (*common.PipelineState, error)
	GetHistory() ([]*common.PipelineState, error)
	GetMetrics() (metrics.Registry, error)
	StartPipeline(runtimeParameters map[string]interface{}) (*common.PipelineState, error)
	StopPipeline() (*common.PipelineState, error)
	ResetOffset() error
	CommitOffset(sourceOffset common.SourceOffset) error
	GetOffset() (common.SourceOffset, error)
	IsRemotePipeline() bool
	GetErrorRecords(stageInstanceName string, size int) ([]api.Record, error)
	GetErrorMessages(stageInstanceName string, size int) ([]api.ErrorMessage, error)
}

type SourceOffsetTracker

type SourceOffsetTracker interface {
	IsFinished() bool

	SetOffset(newOffset *string)

	CommitOffset() error

	GetOffset() *string

	GetLastBatchTime() time.Time
}

Offset tracker is for committing and retrieving offsets in a store.

IsFinished Return if the source finished processing data. This is more of a historical method as it determines whether source is done reading by checking for special offset value. This method will only work (e.g. return true) for (Pull)Source - it will never return true for PushSource.

CommitOffset Change offset for entity in the tracked offset and commit it to persistent store.

GetOffset Return currently staged offset.

GetLastBatchTime Get time of lastly committed batch.

type StageOutput

type StageOutput struct {
	InstanceName string
	Output       map[string][]api.Record
	EventRecords []api.Record
	ErrorRecords []api.Record
	StageErrors  []api.ErrorMessage
}

func NewStageOutput

func NewStageOutput(stageContext api.StageContext, stageOutputJson StageOutputJson) (*StageOutput, error)

type StageOutputJson

type StageOutputJson struct {
	InstanceName string                           `json:"instanceName"`
	Output       map[string][]sdcrecord.SDCRecord `json:"output"`
	EventRecords []sdcrecord.SDCRecord            `json:"eventRecords"`
	ErrorRecords []sdcrecord.SDCRecord            `json:"errorRecords"`
	StageErrors  []api.ErrorMessage               `json:"stageErrors"`
}

func NewStageOutputJson

func NewStageOutputJson(stageOutput StageOutput) (*StageOutputJson, error)