river

Documentation

Index

• Variables
• func InitStatus(addr string, path string)
• type Config
  • func NewConfig(data string) (*Config, error)
  • func NewConfigWithFile(name string) (*Config, error)
• type River
  • func NewRiver(c *Config) (*River, error)
  • func (r *River) Close()
  • func (r *River) Ctx() context.Context
  • func (r *River) Run() error
• type Rule
  • func (r *Rule) CheckFilter(field string) bool
• type SourceConfig
• type TomlDuration
  • func (d *TomlDuration) UnmarshalText(text []byte) error

Variables

var ErrRuleNotExist = errors.New("rule is not exist")

ErrRuleNotExist is the error if rule is not defined.

Functions

func InitStatus

func InitStatus(addr string, path string)

Types

type Config

type Config struct {
	MyAddr     string `toml:"my_addr"`
	MyUser     string `toml:"my_user"`
	MyPassword string `toml:"my_pass"`
	MyCharset  string `toml:"my_charset"`

	ESHttps    bool   `toml:"es_https"`
	ESAddr     string `toml:"es_addr"`
	ESUser     string `toml:"es_user"`
	ESPassword string `toml:"es_pass"`

	StatAddr string `toml:"stat_addr"`
	StatPath string `toml:"stat_path"`

	ServerID uint32 `toml:"server_id"`
	Flavor   string `toml:"flavor"`
	DataDir  string `toml:"data_dir"`

	DumpExec       string `toml:"mysqldump"`
	SkipMasterData bool   `toml:"skip_master_data"`

	Sources []SourceConfig `toml:"source"`

	Rules []*Rule `toml:"rule"`

	BulkSize int `toml:"bulk_size"`

	FlushBulkTime TomlDuration `toml:"flush_bulk_time"`

	SkipNoPkTable bool `toml:"skip_no_pk_table"`
}

Config is the configuration

func NewConfig

func NewConfig(data string) (*Config, error)

NewConfig creates a Config from data.

func NewConfigWithFile

func NewConfigWithFile(name string) (*Config, error)

NewConfigWithFile creates a Config from file.

type River

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

River is a pluggable service within Elasticsearch pulling data then indexing it into Elasticsearch. We use this definition here too, although it may not run within Elasticsearch. Maybe later I can implement a acutal river in Elasticsearch, but I must learn java. :-)

func NewRiver

func NewRiver(c *Config) (*River, error)

NewRiver creates the River from config

func (*River) Close

func (r *River) Close()

Close closes the River

func (*River) Ctx

func (r *River) Ctx() context.Context

Ctx returns the internal context for outside use.

func (*River) Run

func (r *River) Run() error

Run syncs the data from MySQL and inserts to ES.

type Rule

type Rule struct {
	Schema string   `toml:"schema"`
	Table  string   `toml:"table"`
	Index  string   `toml:"index"`
	Type   string   `toml:"type"`
	Parent string   `toml:"parent"`
	ID     []string `toml:"id"`

	// Default, a MySQL table field name is mapped to Elasticsearch field name.
	// Sometimes, you want to use different name, e.g, the MySQL file name is title,
	// but in Elasticsearch, you want to name it my_title.
	FieldMapping map[string]string `toml:"field"`

	// MySQL table information
	TableInfo *schema.Table

	//only MySQL fields in filter will be synced , default sync all fields
	Filter []string `toml:"filter"`

	// Elasticsearch pipeline
	// To pre-process documents before indexing
	Pipeline string `toml:"pipeline"`
}

Rule is the rule for how to sync data from MySQL to ES. If you want to sync MySQL data into elasticsearch, you must set a rule to let use know how to do it. The mapping rule may thi: schema + table <-> index + document type. schema and table is for MySQL, index and document type is for Elasticsearch.

func (*Rule) CheckFilter

func (r *Rule) CheckFilter(field string) bool

CheckFilter checkers whether the field needs to be filtered.

type SourceConfig

type SourceConfig struct {
	Schema string   `toml:"schema"`
	Tables []string `toml:"tables"`
}

SourceConfig is the configs for source

type TomlDuration

type TomlDuration struct {
	time.Duration
}

TomlDuration supports time codec for TOML format.

func (*TomlDuration) UnmarshalText

func (d *TomlDuration) UnmarshalText(text []byte) error

UnmarshalText implementes TOML UnmarshalText