sink

Documentation

Index

• func InitElasticClient(url string) (*elastic.Client, error)
• type CfgElasticSearchSink
• type DataStream
  • func NewDataStream(id uint32, typearg string, src string) *DataStream
  • func (d *DataStream) Map(fn MapCallbackFunc) *DataStream
• type ElasticBulkProcessor
  • func NewElasticSearchSink(ec *CfgElasticSearchSink) (*ElasticBulkProcessor, error)
  • func (b *ElasticBulkProcessor) Close() error
  • func (b *ElasticBulkProcessor) Init() error
  • func (b *ElasticBulkProcessor) Run()
• type MapCallbackFunc

Functions

func InitElasticClient

func InitElasticClient(url string) (*elastic.Client, error)

InitElasticClient is a construct function for Client

Types

type CfgElasticSearchSink

type CfgElasticSearchSink struct {
	ID             uint32
	Name           string
	Input          chan *DataStream
	Uri            string
	IndexPrefix    string
	PrefixTypeList []string
	Mapping        string
	FlushInterval  int
	Parallelism    int
}

type DataStream

type DataStream struct {
	ID        uint32
	TimeStamp int64
	Type      string
	Source    string
	RecordMap map[string]interface{}
}

DataStream is the basic structure for streaming data processinging.

func NewDataStream

func NewDataStream(id uint32, typearg string, src string) *DataStream

func (*DataStream) Map

func (d *DataStream) Map(fn MapCallbackFunc) *DataStream

type ElasticBulkProcessor

type ElasticBulkProcessor struct {
	ID             uint32
	Name           string
	Input          <-chan *DataStream
	Parallelism    int
	C              *elastic.Client
	P              *elastic.BulkProcessor
	IndexPrefix    string
	Mapping        string
	PrefixTypeList []string
	FlushInterval  int
	StopChanList   chan struct{}
	CreateIdxFlag  chan struct{}
}

ElasticBulkProcessor elastic bulk import processor

func NewElasticSearchSink

func NewElasticSearchSink(ec *CfgElasticSearchSink) (*ElasticBulkProcessor, error)

NewElasticSearchSink :Bulk import record to ElasticSearch

func (*ElasticBulkProcessor) Close

func (b *ElasticBulkProcessor) Close() error

Close the bulker.

func (*ElasticBulkProcessor) Init

func (b *ElasticBulkProcessor) Init() error

Run starts the ElasticBulkProcessor.

func (*ElasticBulkProcessor) Run

func (b *ElasticBulkProcessor) Run()

Run is used bulkupload Record to ElasticSearch Server

type MapCallbackFunc

type MapCallbackFunc func(*DataStream) *DataStream