$ npm install -g cnpm --registry=https://registry.npm.taobao.org

$ cnpm install

export NODE_HOME=/usr/local/elastic/node/node-v6.9.2-linux-x64

export PATH=$NODE_HOME/bin:$PATH

$ source /etc/profile

http.cors.enabled: true

http.cors.allow-origin: "*"

./grunt server 

npm install mysql_2_elasticsearch

var esMysqlRiver = require('mysql_2_elasticsearch');



var river_config = {

  mysql: {

    host: '127.0.0.1',

    user: 'root',

    password: 'root',

    database: 'users',

    port: 3306

  },

  elasticsearch: {

    host_config: {               // es客户端的配置参数

      host: 'localhost:9200',

      // log: 'trace'

    },

    index: 'myIndex'

  },

  riverMap: {

    'users => users': {}         // 将数据表 users 导入到 es 类型: /myIndex/users

  }

};





/*

** 以下代码内容：

** 通过 esMysqlRiver 方法进行数据传输，方法的回调参数(一个JSON对象) obj 包含此次数据传输的结果

** 其中：

** 1. obj.total    => 需要传输的数据表数量

** 2. obj.success  => 传输成功的数据表数量

** 3. obj.failed   => 传输失败的数据表数量

** 4. obj.result   => 本次数据传输的结论

*/



esMysqlRiver(river_config, function(obj) {

  /* 将传输结果打印到终端 */

  console.log('\n---------------------------------');

  console.log('总传送：' + obj.total + '项');

  console.log('成功：' + obj.success + '项');

  console.log('失败：' + obj.failed + '项');

  if (obj.result == 'success') {

    console.log('\n结论：全部数据传送完成！');

  } else {

    console.log('\n结论：传送未成功...');

  }

  console.log('---------------------------------');

  console.log('\n(使用 Ctrl + C 退出进程)');

  /* 将传输结果打印到终端 */

});

var esMysqlRiver = require('mysql_2_elasticsearch');



/*

** mysql_2_elasticsearch 的相关参数配置(详情见注释)

*/



var river_config = {



  /* [必需] MySQL数据库的相关参数(根据实际情况进行修改) */

  mysql: {

    host: '127.0.0.1',

    user: 'root',

    password: 'root',

    database: 'users',

    port: 3306

  },



  /* [必需] es 相关参数(根据实际情况进行修改) */

  elasticsearch: {

    host_config: {               // [必需] host_config 即 es客户端的配置参数，详细配置参考 es官方文档

      host: 'localhost:9200',

      log: 'trace',

      // Other options...

    },

    index: 'myIndex',            // [必需] es 索引名

    chunkSize: 8000,             // [非必需] 单分片最大数据量，默认为 5000 (条数据)

    timeout: '2m'                // [非必需] 单次分片请求的超时时间，默认为 1m

    //(注意：此 timeout 并非es客户端请求的timeout，后者请在 host_config 中设置)

  },



  /* [必需] 数据传送的规则 */

  riverMap: {

    'users => users': {            // [必需] 'a => b' 表示将 mysql数据库中名为 'a' 的 table 的所有数据 输送到 es中名为 'b' 的 type 中去

      filter_out: [                // [非必需] 需要过滤的字段名，即 filter_out 中的设置的所有字段将不会被导入 elasticsearch 的数据中

        'password',

        'age'

      ],

      exception_handler: {           // [非必需] 异常处理器，使用JS正则表达式处理异常数据，避免 es 入库时由于类型不合法造成数据缺失

        'birthday': [                // [示例] 对 users 表的 birthday 字段的异常数据进行处理

          {

            match: /NaN/gi,          // [示例] 正则条件(此例匹配字段值为 "NaN" 的情况)

            writeAs: null            // [示例] 将 "NaN" 重写为 null

          },

          {

            match: /(\d{4})年/gi,    // [示例] 正则表达式(此例匹配字段值为形如 "2016年" 的情况)

            writeAs: '$1.1'          // [示例] 将 "2015年" 样式的数据重写为 "2016.1" 样式的数据

          }

        ]

      }

    },

    // Other fields' options...

  }



};





/*

** 将传输结果打印到终端

*/



esMysqlRiver(river_config, function(obj) {

  console.log('\n---------------------------------');

  console.log('总传送：' + obj.total + '项');

  console.log('成功：' + obj.success + '项');

  console.log('失败：' + obj.failed + '项');

  if (obj.result == 'success') {

    console.log('\n结论：全部数据传送完成！');

  } else {

    console.log('\n结论：传送未成功...');

  }

  console.log('---------------------------------');

  console.log('\n(使用 Ctrl + C 退出进程)');

});

select * from x_order where userId = 1 order by id desc limit 10,1;



解析之后会变成golang的一个struct，来看看具体的定义：



&sqlparser.Select{

    Comments:sqlparser.Comments(nil),

    Distinct:"",

    SelectExprs:sqlparser.SelectExprs{(*sqlparser.StarExpr)(0xc42000aee0)},

    From:sqlparser.TableExprs{(*sqlparser.AliasedTableExpr)(0xc420016930)},

    Where:(*sqlparser.Where)(0xc42000afa0),

    GroupBy:sqlparser.GroupBy(nil),

    Having:(*sqlparser.Where)(nil),

    OrderBy:sqlparser.OrderBy{(*sqlparser.Order)(0xc42000af20)},

    Limit:(*sqlparser.Limit)(0xc42000af80),

    Lock:""

}

type Order struct {

    Expr      ValExpr

    Direction string

}\

type Limit struct {

    Offset, Rowcount ValExpr

}

&sqlparser.Where{

    Type:"where",

    Expr:(*sqlparser.ComparisonExpr)(0xc420016a50)

}

select * from users where user_id = 1 and product_id =2



=>



&sqlparser.Where{

    Type:"where",

    Expr:(*sqlparser.AndExpr)(0xc42000b020)

}



AndExpr有Left和Right两个成员，分别是：



Left:

&sqlparser.ComparisonExpr{

    Operator:"=",

    Left:(*sqlparser.ColName)(0xc4200709c0),

    Right:sqlparser.NumVal{0x31}

}



Right:

&sqlparser.ComparisonExpr{

    Operator:"=",

    Left:(*sqlparser.ColName)(0xc420070a50),

    Right:sqlparser.NumVal{0x32}

}

a and (b and (c or d) and e)



=>



select * from user_history where user_id = 1 and (product_id = 2 and (star_num = 4 or star_num = 5) and banned = 1)

query {

    boolmust {

        a,

        boolmust {

            b,

            c

        }

    }

}

query {

    boolmust {

        a,

        boolmust {

            b,

            boolshould {

                c,

                d

            },

            e

        }

    }

}

AndExpr => bool must [{left}, {right}]

OrExpr => bool should [{left}, {right}]

ComparisonExpr => 一般是叶子节点

ParenBoolExpr => 指代括号表达式，其实内部是上述三种节点的某一种，所以直接取出内部节点按上述方法来处理

package main

 

import (

    "fmt"

    "io/ioutil"

    "os"

)

 

func main() {

    //apply run path "." without argument.

    if len(os.Args) == 1 {

        listDir(".")

    } else {

        listDir(os.Args[1])

    }

}

 

func listDir(dirFile string) {

    files, _ := ioutil.ReadDir(dirFile)

    for _, f := range files {

        t := f.ModTime()

        fmt.Println(f.Name(), dirFile+"/"+f.Name(), f.IsDir(), t, f.Size())

 

        if f.IsDir() {

            listDir(dirFile + "/" + f.Name())

        }

    }

}

$ go get github.com/elastic/beats

$ cd $GOPATH/src/github.com/{user}

$ cookiecutter $GOPATH/src/github.com/elastic/beats/generate/beat

project_name [Examplebeat]: lsbeat

github_name [your-github-name]: {username}

beat [lsbeat]:

beat_path [github.com/{github id}]:

full_name [Firstname Lastname]: {Full Name}

$ cd lsbeat

$ tree

.

├── CONTRIBUTING.md

├── LICENSE

├── Makefile

├── README.md

├── beater

│   └── lsbeat.go

├── config

│   ├── config.go

│   └── config_test.go

├── dev-tools

│   └── packer

│       ├── Makefile

│       ├── beats

│       │   └── lsbeat.yml

│       └── version.yml

├── docs

│   └── index.asciidoc

├── etc

│   ├── beat.yml

│   └── fields.yml

├── glide.yaml

├── lsbeat.template.json

├── main.go

├── main_test.go

└── tests

    └── system

        ├── config

        │   └── lsbeat.yml.j2

        ├── lsbeat.py

        ├── requirements.txt

        └── test_base.py

$ make setup

$ git remote add origin git@github.com:{username}/lsbeat.git

$ git push -u origin master

lsbeat.yml

lsbeat:

# Defines how often an event is sent to the output

period: 1s

lsbeat:

# Defines how often an event is sent to the output

period: 10s

path: "."

$ make update

$ cat lsbeat.yml

 

################### Lsbeat Configuration Example #########################

 

############################# Lsbeat ######################################

 

lsbeat:

  # Defines how often an event is sent to the output

  period: 10s

  path: "."

###############################################################################

package config

 

import "time"

 

type Config struct {

Period time.Duration `config:"period"`

Path   string        `config:"path"`

}

 

var DefaultConfig = Config{

Period: 10 * time.Second,

Path:   ".",

}

type Lsbeat struct {

done   chan struct{}

config config.Config

client publisher.Client

 

lastIndexTime time.Time

...

}

func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {

config := config.DefaultConfig

if err := cfg.Unpack(&config); err != nil {

return nil, fmt.Errorf("Error reading config file: %v", err)

}

 

ls := &Lsbeat{

done:   make(chan struct{}),

config: config,

}

return ls, nil

}

"@timestamp": common.Time(time.Now()),

"type": beatname,

"modtime": common.Time(t),

"filename": f.Name(),

"path": dirFile + "/" + f.Name(),

"directory": f.IsDir(),

"filesize": f.Size(),

func (bt *Lsbeat) listDir(dirFile string, beatname string, init bool) {

files, _ := ioutil.ReadDir(dirFile)

for _, f := range files {

t := f.ModTime()

 

event := common.MapStr{

"@timestamp": common.Time(time.Now()),

"type":       beatname,

"modtime":    common.Time(t),

"filename":   f.Name(),

"path":       dirFile + "/" + f.Name(),

"directory":  f.IsDir(),

"filesize":   f.Size(),

}

if init {

// index all files and directories on init

bt.client.PublishEvent(event)

} else {

// Index only changed files since last run.

if t.After(bt.lastIndexTime) {

bt.client.PublishEvent(event)

}

}

 

if f.IsDir() {

bt.listDir(dirFile+"/"+f.Name(), beatname, init)

}

}

}

import (

"fmt"

"io/ioutil"

"time"

)

func (bt *Lsbeat) Run(b *beat.Beat) error {

logp.Info("lsbeat is running! Hit CTRL-C to stop it.")

 

bt.client = b.Publisher.Connect()

ticker := time.NewTicker(bt.config.Period)

counter := 1

for {

 

select {

case <-bt.done:

return nil

case <-ticker.C:

}

 

bt.listDir(bt.config.Path, b.Name, true)   // call lsDir

bt.lastIndexTime = time.Now()               // mark Timestamp

 

logp.Info("Event sent")

counter++

}

 

}

func (bt *Lsbeat) Stop() {

bt.client.Close()

close(bt.done)

}

- key: lsbeat

  title: LS Beat

  description:

  fields:

    - name: counter

      type: integer

      required: true

      description: >

        PLEASE UPDATE DOCUMENTATION

    #new fiels added lsbeat

    - name: modtime

      type: date

    - name: filename

      type: text

    - name: path

    - name: directory

      type: boolean

    - name: filesize

      type: long

{

  "mappings": {

        ...

  },

  "order": 0,

  "settings": {

    "index.refresh_interval": "5s",

    "analysis": {

      "analyzer": {

        "ls_ngram_analyzer": {

          "tokenizer": "ls_ngram_tokenizer"

        }

      },

      "tokenizer": {

        "ls_ngram_tokenizer": {

          "type": "ngram",

          "min_gram": "2",

          "token_chars": [

            "letter",

            "digit"

          ]

        }

      }

    }

  },

  "template": "lsbeat-*"

}

lsbeat:

  # Defines how often an event is sent to the output

  period: 10s

  path: "/Users/ec2-user/go"

• app
• visTypes
• spyModes
• fieldFormatter

• 单日索引数据条数600亿，新增索引文件25TB (含一个复制片则为50TB)
• 业务高峰期峰值索引速率维持在百万条/秒
• 历史数据保留时长根据业务需求制定，从10天 - 90天不等
• 集群共3441个索引、17000个分片、数据总量约9300亿, 磁盘总消耗1PB
• Kibana用户600多人, 每日来自Kibana和第三方的API调用共63万次
• 查询响应时间百分位 75%:0.160s  90%:1.640s 95%:6.691s 99%:14.0039s

1. 首先很有必要对ES的结点做角色划分和隔离。大家知道ES的data node除了放数据以外，也可以兼任master和client的角色，多数同学会将这些角色混入到data node。然而对于一个规模较大，用户较多的集群，master和client在一些极端使用情况下可能会有性能瓶颈甚至内存溢出，从而使得共存的data node故障。data node的故障恢复涉及到数据的迁移，对集群资源有一定消耗，容易造成数据写入延迟或者查询减慢。如果将master和client独立出来，一旦出现问题，重启后几乎是瞬间就恢复的，对用户几乎没有任何影响。另外将这些角色独立出来的以后，也将对应的计算资源消耗从data node剥离出来，更容易掌握data node资源消耗与写入量和查询量之间的联系，便于做容量管理和规划。
2. 避免过高的并发，包括控制shard数量和threadpool的数量。在写入量和查询性能能够满足的前提下，为索引分配尽量少的分片。分片过多会带来诸多负面影响，例如：每次查询后需要汇总排序的数据更多；过多的并发带来的线程切换造成过多的CPU损耗；索引的删除和配置更新更慢Issue#18776; 过多的shard也带来更多小的segment，而过多的小segment会带来非常显著的heap内存消耗，特别是如果查询线程配置得很多的情况下。 配置过大的threadpool更是会产生很多诡异的性能问题Issue#18161里所描述的问题就是我们所经历过的。 默认的Theadpool大小一般来说工作得很不错了。
3. 冷热数据最好做分离。对于日志型应用来说，一般是每天建立一个新索引，当天的热索引在写入的同时也会有较多的查询。如果上面还存有比较长时间之前的冷数据，那么当用户做大跨度的历史数据查询的时候，过多的磁盘IO和CPU消耗很容易拖慢写入，造成数据的延迟。所以我们用了一部分机器来做冷数据的存储，利用ES可以给结点配置自定义属性的功能，为冷结点加上"boxtype":"weak"的标识，每晚通过维护脚本更新冷数据的索引路由设置index.routing.allocation.{require|include|exclude}，让数据自动向冷结点迁移。 冷数据的特性是不再写入，用户查的频率较低，但量级可能很大。比如我们有个索引每天2TB，并且用户要求保持过去90天数据随时可查。保持这么大量的索引为open状态，并非只消耗磁盘空间。ES为了快速访问磁盘上的索引文件，需要在内存里驻留一些数据(索引文件的索引)，也就是所谓的segment memory。稍微熟悉ES的同学知道，JVM heap分配不能超过32GB，对于我们128GB RAM, 48TB磁盘空间的机器而言，如果只跑一个ES实例，只能利用到32GB不到的heap，当heap快用饱和的时候，磁盘上保存的索引文件还不到10TB，这样显然是不经济的。 因此我们决定在冷结点上跑3个ES实例，每个分配31GB heap空间，从而可以在一台物理服务器上存储30多TB的索引数据并保持open状态，供用户随时搜索。 实际使用下来，由于冷数据搜索频率不高，也没有写入，即时只剩余35GB内存给os做文件系统缓存，查询性能还是可以满足需求的。
4. 不同数据量级的shard最好隔离到不同组别的结点。 大家知道ES会自己平衡shard在集群的分布，这个自动平衡的逻辑主要考量三个因素。其一同一索引下的shard尽量分散到不同的结点;其二每个结点上的shard数量尽量接近;其三结点的磁盘有足够的剩余空间。这个策略只能保证shard数量分布均匀，而并不能保证数据大小分布均匀。 实际应用中，我们有200多种索引，数据量级差别很大，大的一天几个TB，小的一个月才几个GB，并且每种类型的数据保留时长又千差万别。抛出的问题，就是如何能比较平衡并充分的利用所有节点的资源。 针对这个问题，我们还是通过对结点添加属性标签来做分组，结合index routing控制的方式来做一些精细化的控制。尽量让不同量级的数据使用不同组别的结点，使得每个组内结点上的数据量比较容易自动平衡。
5. 定期做索引的force merge，并且最好是每个shard merge成一个segment。前面提到过，heap消耗与segment数量也有关系，force merge可以显著降低这种消耗。 如果merge成一个segment还有一个好处，就是对于terms aggregation，搜索时无需构造Global Ordinals，可以提升聚合速度。

• 结点启动的Bootstrap过程加入了很多关键系统参数设置的核验，比如Max File Descriptors, Memory Lock, Virtual Memory设置等等，如果设置不正确会拒绝启动并抛出异常。 与其带着错误的系统参数启动，并在日后造成性能问题，不如启动失败告知用户问题，是个很好的设计！
• 索引性能提升。升级后在同样索引速率下，我们看到cpu消耗下降非常明显，除了对索引速率提升有帮助，也会一定程度提升搜索速率。
• 新的数值型数据结构，存储空间更小，Range和地理位置计算更快速
• Instant Aggregation对于类似now-7d to now这样的范围查询聚合能够做cache了，实际使用下来，效果明显，用户在Kibana上跑个过去一周数据的聚合，头2次刷新慢点，之后有cache了几乎就瞬间刷出！
• 更多的保护措施保证集群的稳定，比如对一次搜索hit的shard数量做了限制，增强了circuit breaker的特性，更好的防护集群资源被坏查询耗尽。

1. 各类Thread pool的使用情况，active/queue/reject可视化出来。 判断集群是否有性能瓶颈了，看看业务高峰期各类queue是不是很高，reject是不是经常发生，基本可以做到心里有数。
2. JVM的heap used%以及old GC的频率，如果old GC频率很高，并且多次GC过后heap used%几乎下不来，说明heap压力太大，要考虑扩容了。（也有可能是有问题的查询或者聚合造成的，需要结合用户访问记录来判断)。
3. Segment memory大小和Segment的数量。节点上存放的索引较多的时候，这两个指标就值得关注，要知道segment memory是常驻heap不会被GC回收的，因此当heap压力太大的时候，可以结合这个指标判断是否是因为节点上存放的数据过多，需要扩容。Segement的数量也是比较关键的，如果小的segment非常多，比如有几千，即使segment memory本身不多，但是在搜索线程很多的情况下，依然会吃掉相当多的heap，原因是lucene为每个segment会在thread local里记录状态信息，这块的heap内存开销和(segment数量* thread数量)相关。
4. 很有必要记录用户的访问记录。我们只开放了http api给用户，前置了一个nginx做http代理，将用户第三方api的访问记录通过access log全部记录下来。通过分析访问记录，可以在集群出现性能问题时，快速找到问题根源，对于问题排查和性能优化都很有帮助。

Settings settings = Settings.builder().put("cluster.name", "xxx")

                    .put("xpack.security.transport.ssl.enabled", false)

                    .put("xpack.security.user", "xxx:xxx")

                    .put("client.transport.sniff", true).build();

try {

    client = new PreBuiltXPackTransportClient(settings)

            .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("xxx.xxx.xxx.xxx"), 9300))

            .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("xxx.xxx.xxx.xxx"), 9300));

} catch (UnknownHostException e) {

    e.printStackTrace();

}

# Create a DataFrame object

from pandasticsearch import DataFrame

df = DataFrame.from_es('http://localhost:9200', index='people')



# Print the schema(mapping) of the index

df.print_schema()

# company

# |-- employee

#   |-- name: {'index': 'not_analyzed', 'type': 'string'}

#   |-- age: {'type': 'integer'}

#   |-- gender: {'index': 'not_analyzed', 'type': 'string'}



# Inspect the columns

df.columns

#['name', 'age', 'gender']



# Get the column

df.name

# Column('name')



# Filter

df.filter(df.age < 13).collect()

# [Row(age=12,gender='female',name='Alice'), Row(age=11,gender='male',name='Bob')]



# Project

df.filter(df.age < 25).select('name', 'age').collect()

# [Row(age=12,name='Alice'), Row(age=11,name='Bob'), Row(age=13,name='Leo')]



# Print the rows into console

df.filter(df.age < 25).select('name').show(3)

# +------+

# | name |

# +------+

# | Alice|

# | Bob  |

# | Leo  |

# +------+



# Sort

df.sort(df.age.asc).select('name', 'age').collect()

#[Row(age=11,name='Bob'), Row(age=12,name='Alice'), Row(age=13,name='Leo')]



# Aggregate

df[df.gender == 'male'].agg(df.age.avg).collect()

# [Row(avg(age)=12)]



# Groupby

df.groupby('gender').collect()

# [Row(doc_count=1), Row(doc_count=2)]



# Groupby and then aggregate

df.groupby('gender').agg(df.age.max).collect()

# [Row(doc_count=1, max(age)=12), Row(doc_count=2, max(age)=13)]



# Convert to Pandas object for subsequent analysis

df[df.gender == 'male'].agg(df.age.avg).to_pandas()

#    avg(age)

# 0        12