bulk_indexing

Documentation

Index

• func New(c *config.Config) (pipeline.Processor, error)
• type BulkIndexingProcessor
  • func (processor *BulkIndexingProcessor) HandleQueueConfig(v *queue.QueueConfig, c *pipeline.Context)
  • func (processor *BulkIndexingProcessor) Name() string
  • func (processor *BulkIndexingProcessor) NewBulkWorker(tag string, ctx *pipeline.Context, bulkSizeInByte int, ...)
  • func (processor *BulkIndexingProcessor) Process(c *pipeline.Context) error
• type Config

Functions

func New

func New(c *config.Config) (pipeline.Processor, error)

Types

type BulkIndexingProcessor

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

处理 bulk 格式的数据索引。

func (*BulkIndexingProcessor) HandleQueueConfig

func (processor *BulkIndexingProcessor) HandleQueueConfig(v *queue.QueueConfig, c *pipeline.Context)

func (*BulkIndexingProcessor) Name

func (processor *BulkIndexingProcessor) Name() string

func (*BulkIndexingProcessor) NewBulkWorker

func (processor *BulkIndexingProcessor) NewBulkWorker(tag string, ctx *pipeline.Context, bulkSizeInByte int, qConfig *queue.QueueConfig, host string)

func (*BulkIndexingProcessor) Process

func (processor *BulkIndexingProcessor) Process(c *pipeline.Context) error

type Config

type Config struct {
	NumOfWorkers         int `config:"worker_size"`
	IdleTimeoutInSecond  int `config:"idle_timeout_in_seconds"`
	MaxConnectionPerHost int `config:"max_connection_per_node"`

	Queues map[string]interface{} `config:"queues,omitempty"`

	Consumer queue.ConsumerConfig `config:"consumer"`

	MaxWorkers int `config:"max_worker_size"`

	DetectActiveQueue  bool `config:"detect_active_queue"`
	DetectIntervalInMs int  `config:"detect_interval"`

	ValidateRequest   bool `config:"valid_request"`
	SkipEmptyQueue    bool `config:"skip_empty_queue"`
	SkipOnMissingInfo bool `config:"skip_info_missing"`

	RotateConfig rotate.RotateConfig `config:"rotate"`

	BulkConfig elastic.BulkProcessorConfig `config:"bulk"`

	Elasticsearch string `config:"elasticsearch,omitempty"`

	WaitingAfter []string `config:"waiting_after"`

	PauseWhen *conditions.Config `config:"pause_when,omitempty"`
}