bundledImpl

Documentation

Index

• func FullDestinPath(destDir string, fullSourcePath string, extension string) string
• func GetClient(namenode string, user string) (*hdfs.Client, error)
• type ArrayToRust
  • func (atr *ArrayToRust) Process(reader io.Reader, customParams interface{}) bool
  • func (atr *ArrayToRust) Setup(customParams interface{}) bool
• type DummyEvent
  • func (de *DummyEvent) Process(reader io.Reader, customParams interface{}) bool
  • func (de *DummyEvent) Setup(customParams interface{}) bool
• type FetchAndExecEvent
  • func (fae *FetchAndExecEvent) Process(reader io.Reader, customParams interface{}) bool
  • func (fae *FetchAndExecEvent) Setup(customParams interface{}) bool
• type HDFS
  • func NewHDFS() *HDFS
  • func (hdfsStruct *HDFS) FilterListFiles() (map[string]api.FileEntity, error)
  • func (hdfsStruct *HDFS) FindWorkingNamenode()
  • func (hdfsStruct HDFS) GetFileSet() map[string]int64
  • func (hdfsStruct HDFS) GetFilteredFileSet(filter string) map[string]int64
  • func (hdfsStruct HDFS) ListFiles() (map[string]api.FileEntity, error)
  • func (hdfsStruct *HDFS) Watch(eventHandlers []api.IOEvent) (bool, error)
• type KafkaEmitEvent
  • func (kee *KafkaEmitEvent) Process(reader io.Reader, customParams interface{}) bool
  • func (kee *KafkaEmitEvent) Setup(customParams interface{}) bool
• type LocalFilesystem
  • func NewLocalFilesystem() *LocalFilesystem
  • func (lf *LocalFilesystem) ListFiles() map[string]api.FileEntity
  • func (lf *LocalFilesystem) Watch(eventHandlers []api.IOEvent) (bool, error)
• type S3
  • func NewS3() *S3
  • func (s3 S3) GetFileSet(minioclient *minio.Client) map[string]int64
  • func (s3 S3) GetFilteredFileSet(filter string, minioClient *minio.Client) map[string]int64
  • func (s3 S3) ListFiles(minioClient *minio.Client) map[string]api.FileEntity
  • func (s3 *S3) Watch(eventHandlers []api.IOEvent) (bool, error)
• type StreamAndExecEvent
  • func (sae *StreamAndExecEvent) Process(reader io.Reader, customParams interface{}) bool
  • func (sae *StreamAndExecEvent) Setup(customParams interface{}) bool

Functions

func FullDestinPath

func FullDestinPath(destDir string, fullSourcePath string, extension string) string

extension example strconv.Itoa(key)+."parquet"

func GetClient

func GetClient(namenode string, user string) (*hdfs.Client, error)

Code copied from colinmarc/hdfs

Types

type ArrayToRust

type ArrayToRust struct {
}

func (*ArrayToRust) Process

func (atr *ArrayToRust) Process(reader io.Reader, customParams interface{}) bool

func (*ArrayToRust) Setup

func (atr *ArrayToRust) Setup(customParams interface{}) bool

type DummyEvent

type DummyEvent struct {
}

func (*DummyEvent) Process

func (de *DummyEvent) Process(reader io.Reader, customParams interface{}) bool

func (*DummyEvent) Setup

func (de *DummyEvent) Setup(customParams interface{}) bool

type FetchAndExecEvent

type FetchAndExecEvent struct {
	DestinationFolder string
	CmdTemplate       string
	Data              map[string]interface{}
	Tmpl              *template.Template
}

func (*FetchAndExecEvent) Process

func (fae *FetchAndExecEvent) Process(reader io.Reader, customParams interface{}) bool

func (*FetchAndExecEvent) Setup

func (fae *FetchAndExecEvent) Setup(customParams interface{}) bool

type HDFS

type HDFS struct {
	Namenode      string
	WatchFolder   string
	MarkerFolder  string
	ExcludeFolder string
	GraceMilliSec int
	HDFSClient    *hdfs.Client
	Ctx           context.Context
	User          string
}

func NewHDFS

func NewHDFS() *HDFS

func (*HDFS) FilterListFiles

func (hdfsStruct *HDFS) FilterListFiles() (map[string]api.FileEntity, error)

TODO use https://github.com/deckarep/golang-set

func (*HDFS) FindWorkingNamenode

func (hdfsStruct *HDFS) FindWorkingNamenode()

func (HDFS) GetFileSet

func (hdfsStruct HDFS) GetFileSet() map[string]int64

func (HDFS) GetFilteredFileSet

func (hdfsStruct HDFS) GetFilteredFileSet(filter string) map[string]int64

func (HDFS) ListFiles

func (hdfsStruct HDFS) ListFiles() (map[string]api.FileEntity, error)

func (*HDFS) Watch

func (hdfsStruct *HDFS) Watch(eventHandlers []api.IOEvent) (bool, error)

type KafkaEmitEvent

type KafkaEmitEvent struct {
	Data     map[string]interface{}
	Tmpl     *template.Template
	Producer *kafka.Writer
	Topic    string
	// contains filtered or unexported fields
}

func (*KafkaEmitEvent) Process

func (kee *KafkaEmitEvent) Process(reader io.Reader, customParams interface{}) bool

func (*KafkaEmitEvent) Setup

func (kee *KafkaEmitEvent) Setup(customParams interface{}) bool

type LocalFilesystem

type LocalFilesystem struct {
	WatchFolder   string
	ExcludeFolder string
	GraceMilliSec int
	Ctx           context.Context
}

func NewLocalFilesystem

func NewLocalFilesystem() *LocalFilesystem

func (*LocalFilesystem) ListFiles

func (lf *LocalFilesystem) ListFiles() map[string]api.FileEntity

func (*LocalFilesystem) Watch

func (lf *LocalFilesystem) Watch(eventHandlers []api.IOEvent) (bool, error)

type S3

type S3 struct {
	Endpoint      string
	UseSSL        bool
	BucketName    string
	User          string
	Password      string
	Token         string
	WatchFolder   string
	MarkerFolder  string
	ExcludeFolder string
	GraceMilliSec int
	MinioClient   *minio.Client
	Ctx           context.Context
}

func NewS3

func NewS3() *S3

func (S3) GetFileSet

func (s3 S3) GetFileSet(minioclient *minio.Client) map[string]int64

func (S3) GetFilteredFileSet

func (s3 S3) GetFilteredFileSet(filter string, minioClient *minio.Client) map[string]int64

func (S3) ListFiles

func (s3 S3) ListFiles(minioClient *minio.Client) map[string]api.FileEntity

func (*S3) Watch

func (s3 *S3) Watch(eventHandlers []api.IOEvent) (bool, error)

type StreamAndExecEvent

type StreamAndExecEvent struct {
	CmdTemplate string
	Data        map[string]interface{}
	Tmpl        *template.Template
}

func (*StreamAndExecEvent) Process

func (sae *StreamAndExecEvent) Process(reader io.Reader, customParams interface{}) bool

func (*StreamAndExecEvent) Setup

func (sae *StreamAndExecEvent) Setup(customParams interface{}) bool