file

Documentation

Index

• Constants
• func GetSink() api.Sink
• func GetSource() api.Source
• type FileDirSourceRewindMeta
• type FileType
• type Source
  • func (fs *Source) Close(ctx api.StreamContext) error
  • func (fs *Source) Connect(_ api.StreamContext, sch api.StatusChangeHandler) error
  • func (fs *Source) GetOffset() (any, error)
  • func (fs *Source) Info() (i model.NodeInfo)
  • func (fs *Source) Load(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest)
  • func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error
  • func (fs *Source) Pull(ctx api.StreamContext, _ time.Time, ingest api.TupleIngest, ...)
  • func (fs *Source) ResetOffset(_ map[string]any) error
  • func (fs *Source) Rewind(offset any) error
  • func (fs *Source) SetEofIngest(eof api.EOFIngest)
  • func (fs *Source) TransformType() api.Source
• type SourceConfig
• type WatchWrapper
  • func (f *WatchWrapper) Close(ctx api.StreamContext) error
  • func (f *WatchWrapper) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error
  • func (f *WatchWrapper) Provision(ctx api.StreamContext, configs map[string]any) error
  • func (f *WatchWrapper) SetEofIngest(eof api.EOFIngest)
  • func (f *WatchWrapper) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) error
• type WithTime
• type WithTimeSlice
  • func (f WithTimeSlice) Len() int
  • func (f WithTimeSlice) Less(i, j int) bool
  • func (f WithTimeSlice) Swap(i, j int)

Constants

const (
	GZIP = "gzip"
	ZSTD = "zstd"
)

Functions

func GetSink

func GetSink() api.Sink

func GetSource

func GetSource() api.Source

Types

type FileDirSourceRewindMeta

type FileDirSourceRewindMeta struct {
	LastModifyTime time.Time `json:"lastModifyTime"`
}

type FileType

type FileType string

const (
	JSON_TYPE    FileType = "json"
	CSV_TYPE     FileType = "csv"
	LINES_TYPE   FileType = "lines"
	PARQUET_TYPE FileType = "parquet"
)

type Source

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

Source load data from file system. Depending on file types, it may read line by line like lines, csv. Otherwise, it reads the file as a whole and send to company reader node to read and split. The planner need to plan according to the file type.

func (*Source) Close

func (fs *Source) Close(ctx api.StreamContext) error

func (*Source) Connect

func (fs *Source) Connect(_ api.StreamContext, sch api.StatusChangeHandler) error

func (*Source) GetOffset

func (fs *Source) GetOffset() (any, error)

func (*Source) Info

func (fs *Source) Info() (i model.NodeInfo)

func (*Source) Load

func (fs *Source) Load(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest)

func (*Source) Provision

func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error

func (*Source) Pull

func (fs *Source) Pull(ctx api.StreamContext, _ time.Time, ingest api.TupleIngest, ingestError api.ErrorIngest)

Pull file source may ingest bytes or tuple For stream source, it ingest one line For batch source, it ingest the whole file, thus it need a reader node to coordinate and read the content into lines/array

func (*Source) ResetOffset

func (fs *Source) ResetOffset(_ map[string]any) error

func (*Source) Rewind

func (fs *Source) Rewind(offset any) error

func (*Source) SetEofIngest

func (fs *Source) SetEofIngest(eof api.EOFIngest)

func (*Source) TransformType

func (fs *Source) TransformType() api.Source

TransformType must call after provision

type SourceConfig

type SourceConfig struct {
	FileName         string            `json:"datasource"`
	FileType         string            `json:"fileType"`
	Path             string            `json:"path"`
	Interval         cast.DurationConf `json:"interval"`
	IsTable          bool              `json:"isTable"`
	SendInterval     cast.DurationConf `json:"sendInterval"`
	ActionAfterRead  int               `json:"actionAfterRead"`
	MoveTo           string            `json:"moveTo"`
	IgnoreStartLines int               `json:"ignoreStartLines"`
	IgnoreEndLines   int               `json:"ignoreEndLines"`
	// Only use for planning
	Decompression string `json:"decompression"`
	// contains filtered or unexported fields
}

type WatchWrapper

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

func (*WatchWrapper) Close

func (f *WatchWrapper) Close(ctx api.StreamContext) error

func (*WatchWrapper) Connect

func (f *WatchWrapper) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error

func (*WatchWrapper) Provision

func (f *WatchWrapper) Provision(ctx api.StreamContext, configs map[string]any) error

func (*WatchWrapper) SetEofIngest

func (f *WatchWrapper) SetEofIngest(eof api.EOFIngest)

func (*WatchWrapper) Subscribe

func (f *WatchWrapper) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) error

type WithTime

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

type WithTimeSlice

type WithTimeSlice []WithTime

func (WithTimeSlice) Len

func (f WithTimeSlice) Len() int

func (WithTimeSlice) Less

func (f WithTimeSlice) Less(i, j int) bool

func (WithTimeSlice) Swap

func (f WithTimeSlice) Swap(i, j int)