output

Documentation

Index

• func ConsoleLog(prefix string, dfchan chan *datarecord.DataFrame)
• func NewElasticClient(url string) (*elastic.Client, error)
• func RecordMap(d *datarecord.DataFrame)
• func RecordMapWithPrediction(d *datarecord.DataFrame, g *geoipmap.GeoIPCollector)
• func RecordStreamProccesor(dfchan chan *datarecord.DataFrame, fanoutList []chan *datarecord.DataFrame)
• type ElasticBulkProcessor
  • func ElasticBulkImportProcessor(url string, numWorkers int, indexPrefix string, FlushInterval int) (*ElasticBulkProcessor, error)
  • func (b *ElasticBulkProcessor) Close() error
  • func (b *ElasticBulkProcessor) ElasticRecordProcessor(dfchan chan *datarecord.DataFrame)
  • func (b *ElasticBulkProcessor) Run() error

Functions

func ConsoleLog

func ConsoleLog(prefix string, dfchan chan *datarecord.DataFrame)

func NewElasticClient

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

NewElasticClient is a construct function for Client

func RecordMap

func RecordMap(d *datarecord.DataFrame)

func RecordMapWithPrediction

func RecordMapWithPrediction(d *datarecord.DataFrame, g *geoipmap.GeoIPCollector)

func RecordStreamProccesor

func RecordStreamProccesor(dfchan chan *datarecord.DataFrame, fanoutList []chan *datarecord.DataFrame)

RecordStreamProccesor : stream process worker....

Types

type ElasticBulkProcessor

type ElasticBulkProcessor struct {
	C             *elastic.Client
	P             *elastic.BulkProcessor
	Workers       int
	IndexPrefix   string
	FlushInterval int
	StopC         chan struct{} // stop channel for caller
	CreateIdxFlag chan struct{}
}

ElasticBulkProcessor elastic bulk import processor

func ElasticBulkImportProcessor

func ElasticBulkImportProcessor(url string, numWorkers int, indexPrefix string, FlushInterval int) (*ElasticBulkProcessor, error)

ElasticImport :Bulk import record to ElasticSearch

func (*ElasticBulkProcessor) Close

func (b *ElasticBulkProcessor) Close() error

Close the bulker.

func (*ElasticBulkProcessor) ElasticRecordProcessor

func (b *ElasticBulkProcessor) ElasticRecordProcessor(dfchan chan *datarecord.DataFrame)

ElasticRecordProcessor is used bulkupload Record to ElasticSearch Server

func (*ElasticBulkProcessor) Run

func (b *ElasticBulkProcessor) Run() error

Run starts the ElasticBulkProcessor.