可定制的 elasticsearch 数据导入工具 ——mysql_2_elasticsearch

A customizable importer from mysql to elasticsearch.
可定制的 elasticsearch 数据导入工具 ——基于 elasticsearch 的 JS API
 
 
【github 项目地址】
https://github.com/parksben/mysql_2_elasticsearch 


【主要功能】
1. 完全使用 JS 实现数据从 MySQL 到 elasticsearch 的迁移;
2. 可批量导入多张 MySQL 表;
3. 可自定义的数据迁移规则(数据表/字段关系、字段过滤、使用正则进行数据处理);
4. 可自定义的异步分片导入方式,数据导入效率更高。


【一键安装】
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 退出进程)');
});


【注意事项及参考】
1. elasticsearch数据导入前请先配置好数据的 mapping;
2. "host_config" 更多参数设置详见 [es官方API文档] https://www.elastic.co/guide/e ... .html
3. mysql 表的自增 id 将自动替换为 "表名+_id" 的格式,如:"users_id";
4. 如出现数据缺失情况,请注意查看 elasticsearch 终端进程或日志,找出未成功导入的数据,通过设置 exception_handler 参数处理它。
继续阅读 »
A customizable importer from mysql to elasticsearch.
可定制的 elasticsearch 数据导入工具 ——基于 elasticsearch 的 JS API
 
 
【github 项目地址】
https://github.com/parksben/mysql_2_elasticsearch 


【主要功能】
1. 完全使用 JS 实现数据从 MySQL 到 elasticsearch 的迁移;
2. 可批量导入多张 MySQL 表;
3. 可自定义的数据迁移规则(数据表/字段关系、字段过滤、使用正则进行数据处理);
4. 可自定义的异步分片导入方式,数据导入效率更高。


【一键安装】
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 退出进程)');
});


【注意事项及参考】
1. elasticsearch数据导入前请先配置好数据的 mapping;
2. "host_config" 更多参数设置详见 [es官方API文档] https://www.elastic.co/guide/e ... .html
3. mysql 表的自增 id 将自动替换为 "表名+_id" 的格式,如:"users_id";
4. 如出现数据缺失情况,请注意查看 elasticsearch 终端进程或日志,找出未成功导入的数据,通过设置 exception_handler 参数处理它。 收起阅读 »

Day6:《记一次es性能调优》


一.前言
应medcl写es文章的时候,其实这段时间es研究的不多,感觉没什么新东西可写。
考虑只有这次调优心得可与大家分享,文笔有限,见谅!
二.背景
先交代一下背景,调优的项目是某电商类搜索项目,流量来自于前端的app和h5。
搜索主要是根据用户的地理位置和关键字等条件搜索附近的商家和商品。
商品数据大概在5000w左右,商品更新很频繁,更新量大概是每天2000w条左右,(因商家经常会促销、或者调上下架状态、改价格等)查询也相当频繁。
集群有2个集群,一个主一个备,用于有问题的时候随时切换。主集群有8个节点,配置是32核,
32g内存的docker的机器。给es jvm分配20g内存,jdk 版本是1.7,gc 是使用parnew/cms gc。
这个项目我是后期加入的,来的时候项目已上线。由于参与进来的时候es跑的也还是比较稳定,所以也一直
没调过es的参数。程序,参数基本上也就保持上线的时候那个样子。
es上线的时候是用的1.5版本,后期没升过级。
三.问题
项目大概跑了一年多,时间来到大概16年的9月份。搜索请求响应时间开始出现几秒才完成的情况,
我就被拉过来调优了。通过我们自己内部的调用方法监控,tp99和avg这些值还好,维持在200ms以下。max 最大有5,6s的情况,而且次数有点多。
这里没怎么折腾,很快定位就是es gc导致的。翻了一下es gc日志,就是cms remark这个阶段时间特别长,
而且 这个阶段是stop the world的。
四.解决
为什么remark阶段这么长时间? 直接上结论,就是一次cms 周期内,并发标记后到remark这个期间jvm 堆内存对象
变化很大。说白了对应我们的场景就是一大波 es bulk操作。对应Bigdesk观察,几秒的卡顿基本都出现在一大波 es bulk操作之后。
这里解释一下,引用网上文章的说法:
remark如果耗时较长,通常原因是在cms gc已经结束了concurrent-mark步骤后,旧生代的引用关系仍然发生了很多的变化,旧生代的引用关系发生变化的原因主要是:
* 在这个间隔时间段内,新生代晋升到旧生代的对象比较多;
* 在这个间隔时间段内,新生代没怎么发生ygc,但创建出来的对象又比较多,这种通常就只能是新生代比较大的原因;
原文地址:
http://hellojava.info/?tag=cms-gc-remark

调整一:
加cms gc 的 线程
直接从根源入手,你remark 慢,我就让你跑快点。
因为我们是32 核的cpu ,cpu 利用率用bigdesk观察还是很低的,5%左右。这么低,那就加点线程呗。
-XX:ParallelGCThreads= N
-XX:ParallelCMSThreads= M
调整这2个参数都可以,它们的关系:ParallelCMSThreads = (ParallelGCThreads + 3)/4)
调整后情况缓解了一些,remark还是有3,4秒的情况。
调整二:
关于这点是我们自己的问题。一次bulk 操作就写1条数据。
是的,你没有看错,我们这边的工程师就是这么干的。
一年以前提过这里最好是能合并写,但现在还是这个样子。
这里有一些业务上的原因,合并写会导致一些字段值不准确。
合并写暂时没办法,只能调整es 了。(这里说明一下,其实合并写应该是本次优化比较有效果的办法,可惜这招不让我用。)
通过bigdesk观察,bulk线程池有reject的情况。
但就增加bulk线程池的消费线程,加快数据的消费速度,减少对象驻留在jvm 的时间。
调整后情况没有明显的好转,
但肯定有用,能优化一点是一点嘛。

调整三:
再次从gc入手, -XX:+CMSParallelRemarkEnabled -XX:+CMSScavengeBeforeRemark
这个是网上找的办法:
为了减少第二次暂停的时间,开启并行remark: -XX:+CMSParallelRemarkEnabled。 如果remark还是过长的话,可以开启-XX:+CMSScavengeBeforeRemark选项,强制 remark之前开始一次minor gc,减少remark的暂停时间,但是在remark之后也将立即开始又一次minor gc。调整后情况也没有特别的好转。
以上都是从减小单次cms gc的开销的方向去解决问题,然后我就换了个方向,降低cms gc发生的次数,让它少发生或者不发生。
调整四:
这里调整了一共5个参数,
Xmn10g ==> 8g
CMSInitiatingOccupancyFraction=70 ==>80

index.cache.filter.max_size 2g==>1g
index.cache.filter.expire 2m==>40s
index.refresh_interval 20s==>30s

前2个参数没什么好说的,提高cms gc 被触发的条件,降低cms gc 触发几率。
后3个参数是本文的重点,这里大概讲下es 对于filter cache的管理。
这部分是通过阅读源码分析出来的,涉及代码还挺多,有点复杂,还有很大一部分还是lucene的代码。
这里就不贴大段代码了。
es 对于 filter cache管理是内部维护了一个map的结构(实际是利用com.google.common.cache实现的),关键是这个map 的key 是个它自己定义的类
叫 FilterCacheKey,它override了equals方法
public FilterCacheKey(Object readerKey, Object filterKey) {
    this.readerKey = readerKey;
    this.filterKey = filterKey;
}
...
@Override
public boolean equals(Object o) {
    if (this == o) return true;
//            if (o == null || getClass() != o.getClass()) return false;
    FilterCacheKey that = (FilterCacheKey) o;
    return (readerKey().equals(that.readerKey()) && filterKey.equals(that.filterKey));
}
从这里可以看出,filter cache 能否被再次利用到就跟readerKey和filterKey 有关。
filterkey如果你build 查询语句的时候什么都没设置的话,就是filter对象本身。
举个例子,TermFilter,如果term一样,那前后2次查询filterKey是一致的。
关键是这个readerKey是否一致呢?这个readerKey其实就是lucene 的 indexReader,如果前后2次查询有数据更新并且
index.refresh_interval 这个参数设置的比较小,es 会去刷新indexReader,那么很有可能readerKey不一致。
对应我们的场景,数据更新频繁,而且index.refresh_interval 比较小,那么缓存利用率就不太高。
后一次查询的filter cache 会重新put 到上面提到的map里,然后这个index.cache.filter.max_size 2g 
就迅速占满(我们程序代码里很多地方使用到了filter),配多大都有可能占满。那这个filter cache什么时候被移除呢,index.cache.filter.expire 2m管移除这个事,当然应该还有size满了移除的策略。
这就造成了缓存没什么用,还占这么大地方,还占那么久。
然后这个filter cache就很可能跑到 old gen去了。
那么就让它占少点,不干活就快点走:
index.cache.filter.max_size 2g==>1g
index.cache.filter.expire 2m==>40s
index.refresh_interval 20s==>30s
这些调整完毕,重启es ,通过bigdesk ,gc a线图好看多了,cms gc 基本没了,monitor gc 也顺畅多了。
五.总结和备注
总结:
1.优化这个事,我认为是业务上的优化要比技术上优化有效的多
2.日志和监控是你最好的朋友,仔细的看,你总会看出什么
3.es 缓存要利用好,还是需要好好去设计,回头考虑单独写一篇。
备注:
因为这次优化后,我就离开了原来的公司,没有了原来的环境。所以本文
部分参数和数字可能不准确,我仅凭记忆完成。
 
继续阅读 »

一.前言
应medcl写es文章的时候,其实这段时间es研究的不多,感觉没什么新东西可写。
考虑只有这次调优心得可与大家分享,文笔有限,见谅!
二.背景
先交代一下背景,调优的项目是某电商类搜索项目,流量来自于前端的app和h5。
搜索主要是根据用户的地理位置和关键字等条件搜索附近的商家和商品。
商品数据大概在5000w左右,商品更新很频繁,更新量大概是每天2000w条左右,(因商家经常会促销、或者调上下架状态、改价格等)查询也相当频繁。
集群有2个集群,一个主一个备,用于有问题的时候随时切换。主集群有8个节点,配置是32核,
32g内存的docker的机器。给es jvm分配20g内存,jdk 版本是1.7,gc 是使用parnew/cms gc。
这个项目我是后期加入的,来的时候项目已上线。由于参与进来的时候es跑的也还是比较稳定,所以也一直
没调过es的参数。程序,参数基本上也就保持上线的时候那个样子。
es上线的时候是用的1.5版本,后期没升过级。
三.问题
项目大概跑了一年多,时间来到大概16年的9月份。搜索请求响应时间开始出现几秒才完成的情况,
我就被拉过来调优了。通过我们自己内部的调用方法监控,tp99和avg这些值还好,维持在200ms以下。max 最大有5,6s的情况,而且次数有点多。
这里没怎么折腾,很快定位就是es gc导致的。翻了一下es gc日志,就是cms remark这个阶段时间特别长,
而且 这个阶段是stop the world的。
四.解决
为什么remark阶段这么长时间? 直接上结论,就是一次cms 周期内,并发标记后到remark这个期间jvm 堆内存对象
变化很大。说白了对应我们的场景就是一大波 es bulk操作。对应Bigdesk观察,几秒的卡顿基本都出现在一大波 es bulk操作之后。
这里解释一下,引用网上文章的说法:
remark如果耗时较长,通常原因是在cms gc已经结束了concurrent-mark步骤后,旧生代的引用关系仍然发生了很多的变化,旧生代的引用关系发生变化的原因主要是:
* 在这个间隔时间段内,新生代晋升到旧生代的对象比较多;
* 在这个间隔时间段内,新生代没怎么发生ygc,但创建出来的对象又比较多,这种通常就只能是新生代比较大的原因;
原文地址:
http://hellojava.info/?tag=cms-gc-remark

调整一:
加cms gc 的 线程
直接从根源入手,你remark 慢,我就让你跑快点。
因为我们是32 核的cpu ,cpu 利用率用bigdesk观察还是很低的,5%左右。这么低,那就加点线程呗。
-XX:ParallelGCThreads= N
-XX:ParallelCMSThreads= M
调整这2个参数都可以,它们的关系:ParallelCMSThreads = (ParallelGCThreads + 3)/4)
调整后情况缓解了一些,remark还是有3,4秒的情况。
调整二:
关于这点是我们自己的问题。一次bulk 操作就写1条数据。
是的,你没有看错,我们这边的工程师就是这么干的。
一年以前提过这里最好是能合并写,但现在还是这个样子。
这里有一些业务上的原因,合并写会导致一些字段值不准确。
合并写暂时没办法,只能调整es 了。(这里说明一下,其实合并写应该是本次优化比较有效果的办法,可惜这招不让我用。)
通过bigdesk观察,bulk线程池有reject的情况。
但就增加bulk线程池的消费线程,加快数据的消费速度,减少对象驻留在jvm 的时间。
调整后情况没有明显的好转,
但肯定有用,能优化一点是一点嘛。

调整三:
再次从gc入手, -XX:+CMSParallelRemarkEnabled -XX:+CMSScavengeBeforeRemark
这个是网上找的办法:
为了减少第二次暂停的时间,开启并行remark: -XX:+CMSParallelRemarkEnabled。 如果remark还是过长的话,可以开启-XX:+CMSScavengeBeforeRemark选项,强制 remark之前开始一次minor gc,减少remark的暂停时间,但是在remark之后也将立即开始又一次minor gc。调整后情况也没有特别的好转。
以上都是从减小单次cms gc的开销的方向去解决问题,然后我就换了个方向,降低cms gc发生的次数,让它少发生或者不发生。
调整四:
这里调整了一共5个参数,
Xmn10g ==> 8g
CMSInitiatingOccupancyFraction=70 ==>80

index.cache.filter.max_size 2g==>1g
index.cache.filter.expire 2m==>40s
index.refresh_interval 20s==>30s

前2个参数没什么好说的,提高cms gc 被触发的条件,降低cms gc 触发几率。
后3个参数是本文的重点,这里大概讲下es 对于filter cache的管理。
这部分是通过阅读源码分析出来的,涉及代码还挺多,有点复杂,还有很大一部分还是lucene的代码。
这里就不贴大段代码了。
es 对于 filter cache管理是内部维护了一个map的结构(实际是利用com.google.common.cache实现的),关键是这个map 的key 是个它自己定义的类
叫 FilterCacheKey,它override了equals方法
public FilterCacheKey(Object readerKey, Object filterKey) {
    this.readerKey = readerKey;
    this.filterKey = filterKey;
}
...
@Override
public boolean equals(Object o) {
    if (this == o) return true;
//            if (o == null || getClass() != o.getClass()) return false;
    FilterCacheKey that = (FilterCacheKey) o;
    return (readerKey().equals(that.readerKey()) && filterKey.equals(that.filterKey));
}
从这里可以看出,filter cache 能否被再次利用到就跟readerKey和filterKey 有关。
filterkey如果你build 查询语句的时候什么都没设置的话,就是filter对象本身。
举个例子,TermFilter,如果term一样,那前后2次查询filterKey是一致的。
关键是这个readerKey是否一致呢?这个readerKey其实就是lucene 的 indexReader,如果前后2次查询有数据更新并且
index.refresh_interval 这个参数设置的比较小,es 会去刷新indexReader,那么很有可能readerKey不一致。
对应我们的场景,数据更新频繁,而且index.refresh_interval 比较小,那么缓存利用率就不太高。
后一次查询的filter cache 会重新put 到上面提到的map里,然后这个index.cache.filter.max_size 2g 
就迅速占满(我们程序代码里很多地方使用到了filter),配多大都有可能占满。那这个filter cache什么时候被移除呢,index.cache.filter.expire 2m管移除这个事,当然应该还有size满了移除的策略。
这就造成了缓存没什么用,还占这么大地方,还占那么久。
然后这个filter cache就很可能跑到 old gen去了。
那么就让它占少点,不干活就快点走:
index.cache.filter.max_size 2g==>1g
index.cache.filter.expire 2m==>40s
index.refresh_interval 20s==>30s
这些调整完毕,重启es ,通过bigdesk ,gc a线图好看多了,cms gc 基本没了,monitor gc 也顺畅多了。
五.总结和备注
总结:
1.优化这个事,我认为是业务上的优化要比技术上优化有效的多
2.日志和监控是你最好的朋友,仔细的看,你总会看出什么
3.es 缓存要利用好,还是需要好好去设计,回头考虑单独写一篇。
备注:
因为这次优化后,我就离开了原来的公司,没有了原来的环境。所以本文
部分参数和数字可能不准确,我仅凭记忆完成。
  收起阅读 »

求一个Kibana5能用的高德URL

kibana到5.0后可以直接在kibana.yml上面使用tilemap.url进行配置,
但是之前使用的url貌似都不行,有伙伴能提供个K5能用的url吗?
kibana到5.0后可以直接在kibana.yml上面使用tilemap.url进行配置,
但是之前使用的url貌似都不行,有伙伴能提供个K5能用的url吗?

Day5: 《PacketBeat奇妙的OOM小记》

Beats这个项目的确很好用,几行命令下来,一个成型的Agent就出来了。使用者只需要关注采集什么数据就好,后续的事情libbeat基本都处理完了。不过值得吐槽的是,Beat太散了,管理起来东一个西一个的,产品化的时候对客户说,我们要在机器上放n个Agent不知道客户会是什么样的表情。


d7d0a529244b57acb6ce3796da132df8.jpg



不过轻量级、已部署的特点还是极大的吸引了我,于是就有了后面的事情了。

PacketBeat不明原因的OOM

某天我把PacketBeat放到了我的服务器上面,这台服务器上面有个MongoDB,MongoDB主要是拿来存放ES的元数据的。ES2.x的时候并没有很好的元数据管理,为了能让ES的索引分配的比较均匀,并且有元数据辅助查询,设计好一个元数据管理的仓库是必要的。然后我打开了对MongoDB的抓包功能,恩,一切都很好,接着我打开了日志管理页面,看到了一条一条的MongoDB的包被抓回来,解码,然后塞到了ES。可是第二天一看,咦??Packet跪了?不是吧,ElasticSearch做的产品这么不稳定么。我不信。


06170826_dLgU.png



然后我又启动了第二次,紧接着熟练的top了一下,观察了PacketBeat半个多小时,在被观察的这段时间里面,PacketBeat的表现非常的正常,看不出有什么异样。好吧,那上一次的OOM可能只是个意外,Windows也经常蓝屏嘛,OOM一次也正常。结果第二天我再次打开终端,发现这货居然又OOM了!!


06170909_irst.png



好吧好吧,我感觉我已经踩到Bug了,拿了开源社区这么多东西,总得贡献一下的,好吧,提个Issue去 https://github.com/elastic/beats/issues/2867

真相只有一个

微信群里面聊起这个奇妙的OOM,Medcl大神问是不是因为采集了ES的日志,(我的这台服务器和日志服务器有关系)然后导致滚雪球把PacketBeat给滚死了。咦?说不定真的是这个原因耶!但是看了看PacketBeat,我并没有抓ES的包,而且假如我采集了ES的包,应该一下就OOM掉了,不应该等那么久。不过这么一说,却仿佛打开了新世界的大门


06171040_FVEG.png



我把这台服务器在日志服务器中的角色重新梳理了下,终于发现了这次OOM的原因了。。

由于2.X的ES没有比较好的元数据信息,所以当日志送到LogServer的时候,我做了些额外的操作,让LogServer持久化到ES一定量的时候就会往Mongo写一下元数据信息(当然也有其他服务会往里面做CRUD啦),开始的时候访问Mongo的次数其实是很少的,假设按1W来算。那么问题来了,由于我们的PacketBeat抓了Mongo的包,那么LogServer往ES的CRUD操作都会被PacketBeat给抓走,然后再送回给LogServer


06171248_tcdl.png



那么一个隐藏的滚雪球事件就产生了,刚开始的那段时间,Mongo被抓包的次数只有1W,然后就往LogServer多送了1W条日志,不。。应该多很多,毕竟网络包嘛,然后就导致LogServer因为要管理元数据的频率开始逐渐地提高,逐渐提高CRUD的频率后抓包的内容也越来越多,紧接着到这发生到LogServer的频率也越来越高。。。。。每次PacketBeat崩掉的时候,都送了80W左右的日志量出去,然后它就OOM掉了(因为我那台机器就只剩下2G的空闲内存给它用,被系统给干掉了)。。我居然发现了这样的场景


06171336_PbWI.png



结论

使用PacketBeat的时候,记得要留意一下有没这种反馈型滚雪球的情况,多发生在自己的日志服务器上面。当然那种直接抓ES的就没什么好说了,估计启动了之后没多久就崩溃掉了
继续阅读 »
Beats这个项目的确很好用,几行命令下来,一个成型的Agent就出来了。使用者只需要关注采集什么数据就好,后续的事情libbeat基本都处理完了。不过值得吐槽的是,Beat太散了,管理起来东一个西一个的,产品化的时候对客户说,我们要在机器上放n个Agent不知道客户会是什么样的表情。


d7d0a529244b57acb6ce3796da132df8.jpg



不过轻量级、已部署的特点还是极大的吸引了我,于是就有了后面的事情了。

PacketBeat不明原因的OOM

某天我把PacketBeat放到了我的服务器上面,这台服务器上面有个MongoDB,MongoDB主要是拿来存放ES的元数据的。ES2.x的时候并没有很好的元数据管理,为了能让ES的索引分配的比较均匀,并且有元数据辅助查询,设计好一个元数据管理的仓库是必要的。然后我打开了对MongoDB的抓包功能,恩,一切都很好,接着我打开了日志管理页面,看到了一条一条的MongoDB的包被抓回来,解码,然后塞到了ES。可是第二天一看,咦??Packet跪了?不是吧,ElasticSearch做的产品这么不稳定么。我不信。


06170826_dLgU.png



然后我又启动了第二次,紧接着熟练的top了一下,观察了PacketBeat半个多小时,在被观察的这段时间里面,PacketBeat的表现非常的正常,看不出有什么异样。好吧,那上一次的OOM可能只是个意外,Windows也经常蓝屏嘛,OOM一次也正常。结果第二天我再次打开终端,发现这货居然又OOM了!!


06170909_irst.png



好吧好吧,我感觉我已经踩到Bug了,拿了开源社区这么多东西,总得贡献一下的,好吧,提个Issue去 https://github.com/elastic/beats/issues/2867

真相只有一个

微信群里面聊起这个奇妙的OOM,Medcl大神问是不是因为采集了ES的日志,(我的这台服务器和日志服务器有关系)然后导致滚雪球把PacketBeat给滚死了。咦?说不定真的是这个原因耶!但是看了看PacketBeat,我并没有抓ES的包,而且假如我采集了ES的包,应该一下就OOM掉了,不应该等那么久。不过这么一说,却仿佛打开了新世界的大门


06171040_FVEG.png



我把这台服务器在日志服务器中的角色重新梳理了下,终于发现了这次OOM的原因了。。

由于2.X的ES没有比较好的元数据信息,所以当日志送到LogServer的时候,我做了些额外的操作,让LogServer持久化到ES一定量的时候就会往Mongo写一下元数据信息(当然也有其他服务会往里面做CRUD啦),开始的时候访问Mongo的次数其实是很少的,假设按1W来算。那么问题来了,由于我们的PacketBeat抓了Mongo的包,那么LogServer往ES的CRUD操作都会被PacketBeat给抓走,然后再送回给LogServer


06171248_tcdl.png



那么一个隐藏的滚雪球事件就产生了,刚开始的那段时间,Mongo被抓包的次数只有1W,然后就往LogServer多送了1W条日志,不。。应该多很多,毕竟网络包嘛,然后就导致LogServer因为要管理元数据的频率开始逐渐地提高,逐渐提高CRUD的频率后抓包的内容也越来越多,紧接着到这发生到LogServer的频率也越来越高。。。。。每次PacketBeat崩掉的时候,都送了80W左右的日志量出去,然后它就OOM掉了(因为我那台机器就只剩下2G的空闲内存给它用,被系统给干掉了)。。我居然发现了这样的场景


06171336_PbWI.png



结论

使用PacketBeat的时候,记得要留意一下有没这种反馈型滚雪球的情况,多发生在自己的日志服务器上面。当然那种直接抓ES的就没什么好说了,估计启动了之后没多久就崩溃掉了 收起阅读 »

ES5.0.0 安装记录


创建用户:adduser elasticsearch
可查看创建结果:
##########/etc/passwd
##########/etc/shadow
##########/etc/group
配置环境变量
修改文件:/home/elasticsearch/.profile
追加内容:
export JAVA_HOME=/home/elasticsearch/java/jdk1.8.0_73
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$CLASSPATH
export PATH
配置elasticsearch5.0.0
tar -xf elasticsearch-5.0.0.tar.gz -C /home/elasticsearch/
cd /home/elasticsearch/
ln -sv elasticsearch-5.0.0 elasticsearch
mkdir -pv /esdata/elasticsearch/{data,logs}
chown -R elasticsearch.elasticsearch /esdata/elasticsearch
修改ES配置文件
/home/elasticsearch/elasticsearch-5.0.0/config/elasticsearch.yml
http.cors.enabled: true
http.cors.allow-origin: "*"
path.data: /esdata/elasticsearch/data
path.logs: /esdata/elasticsearch/logs
network.host: 192.168.25.57
http.port: 8201
transport.tcp.port: 8301
bootstrap.memory_lock: true
/home/elasticsearch/elasticsearch-5.0.0/config/jvm.options
-Xms8g
-Xmx8g
修改系统参数
/etc/security/limits.conf
elasticsearch soft nproc 65536
elasticsearch hard nproc 65536
elasticsearch soft nofile 65536
elasticsearch hard nofile 65536
elasticsearch - memlock unlimited
/etc/sysctl.conf
vm.max_map_count = 262144
加载更新:sysctl -p
启动ES服务
su - elasticsearch -c "/home/elasticsearch/elasticsearch/bin/elasticsearch &"
 
继续阅读 »

创建用户:adduser elasticsearch
可查看创建结果:
##########/etc/passwd
##########/etc/shadow
##########/etc/group
配置环境变量
修改文件:/home/elasticsearch/.profile
追加内容:
export JAVA_HOME=/home/elasticsearch/java/jdk1.8.0_73
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$CLASSPATH
export PATH
配置elasticsearch5.0.0
tar -xf elasticsearch-5.0.0.tar.gz -C /home/elasticsearch/
cd /home/elasticsearch/
ln -sv elasticsearch-5.0.0 elasticsearch
mkdir -pv /esdata/elasticsearch/{data,logs}
chown -R elasticsearch.elasticsearch /esdata/elasticsearch
修改ES配置文件
/home/elasticsearch/elasticsearch-5.0.0/config/elasticsearch.yml
http.cors.enabled: true
http.cors.allow-origin: "*"
path.data: /esdata/elasticsearch/data
path.logs: /esdata/elasticsearch/logs
network.host: 192.168.25.57
http.port: 8201
transport.tcp.port: 8301
bootstrap.memory_lock: true
/home/elasticsearch/elasticsearch-5.0.0/config/jvm.options
-Xms8g
-Xmx8g
修改系统参数
/etc/security/limits.conf
elasticsearch soft nproc 65536
elasticsearch hard nproc 65536
elasticsearch soft nofile 65536
elasticsearch hard nofile 65536
elasticsearch - memlock unlimited
/etc/sysctl.conf
vm.max_map_count = 262144
加载更新:sysctl -p
启动ES服务
su - elasticsearch -c "/home/elasticsearch/elasticsearch/bin/elasticsearch &"
  收起阅读 »

Day4: 《将sql转换为es的DSL》

es现在几乎已经是开源搜索引擎的事实标准了,搭建简易,使用方便。不过在很多公司里(包括我司的部分部门),并不是把它当搜索引擎来用,而是当db来用。因为本身查询/搜索原理的区别,使es在千万或者亿级的数据中进行逻辑筛选相对高效。例如一些wms、工单查询系统,单表几十个甚至上百个字段,如果在数据库里为每种类型的查询都建立合适的索引,成本比较高,更不用说索引建多了还会影响到插入速度,后期的索引优化也是比较麻烦的问题。

不过如果把es当db来使的话,始终会有一个绕不过去的坎。就是es的DSL。让所有业务开发去学习dsl的话也不是不可以,但DSL真的有点反人类(不要打我)。简单的a and b或者a or b还比较容易写,如果我要的是a and (b and (c or d) and e)的查询逻辑,那我觉得谁写都会晕。即使是用官方或者第三方提供的client,如果需求多种多样的话,想要灵活地实现`需求=>DSL`的过程还是比较痛苦。

对于业务开发来说,当然是sql更平易近人(毕竟写了这么多年CRUD)。所以还有一种歪门邪道的流派,直接把sql转成DSL。要做sql和DSL转换的工作,需要进行sql的解析,先不要怵,这个年代找一个靠谱的sql parser还是比较容易的。比如阿里开源的druid连接池里的sql模块:
 
https://github.com/alibaba/dru ... d/sql

因为笔者的实现是用的下面这个golang版的parser:

https://github.com/xwb1989/sqlparser

所以用这个来举例吧~

这个是其作者从youtube/vitness里提取并进行改进的一个parser,我们能用到的是一部分子集功能,只需要解析select类的sql。

先举个简单的sql的例子:
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:""
}


sql的select语句在被解析之后生成一个Select的结构体,如果我们不关心使用者需要的字段的话,可以先把SelectExprs/Distinct/Comments/Lock里的内容忽略掉。如果不是分组统计类的需求,也可以先把GroupBy/Having忽略掉。这里我们关心的就剩下From、Where、OrderBy和Limit。

From对应的TableExprs实际上可以认为是简单的字符串,这里的值其实就是`x_order`。

OrderBy实际上是一个元素为
type Order struct {
Expr ValExpr
Direction string
}\

的数组。

Limit也很简单,
type Limit struct {
Offset, Rowcount ValExpr
}

其实就是俩数字。

那么剩下的就是这个Where结构了。where会被解析为AST(`https://en.wikipedia.org/wiki/Abstract_syntax_tree`),中文是抽象语法树。在不说子查询之类的情况下,这个AST也不会太复杂,毕竟where后面的情况比起编译原理里的程序语言来说情况还是要少得多的。以上述的sql为例,这里解析出来的Where结构是这样的:
&sqlparser.Where{
Type:"where",
Expr:(*sqlparser.ComparisonExpr)(0xc420016a50)
}


只有一个节点,一个ComparisonExpr表达式,这个ComparisonExpr,中文比较表达式,指代的就是我们sql里的`user_id = 1`。实际上我们可以认为这个"比较表达式"即是所有复杂AST的叶子节点。叶子结点在AST遍历的时候一般也就是递归的终点。因为这里的查询比较简单,整棵AST只有一个节点,即根节点和叶子节点都是这个ComparisonExpr。

再来一个复杂点的例子。
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}
}


稍微有一些二叉树的样子了吧。把这棵简单的树画出来:


Untitled1.png



回到文章开头的那个复杂的例子:
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)


看着真够麻烦的,我们把这棵树画出来:


Untitled.png



这样看着就直观多了。我们有了AST的结构,那要怎么对应到es的查询DSL呢?少安毋躁。

我们知道es的bool query是可以进行嵌套的,所以实际上我们可以同样可以构造出树形结构的bool query。这里把bool嵌套must和bool嵌套should简化一下,写成boolmust和boolshould:

例如a and (b and c)
query {
boolmust {
a,
boolmust {
b,
c
}
}
}


我们把query内部的第一个boolmust当作根节点,内部嵌套的a和另一个boolmust当作它的两个子节点,然后b和c又是这个boolmust的子节点。可以看出来,实际上这棵树和AST的节点可以一一对应。

再回到文章开头的例子,a and (b and (c or d) and e):
query {
boolmust {
a,
boolmust {
b,
boolshould {
c,
d
},
e
}
}
}

和前文中ast来做个简单的结构对比~


dsl和ast对比.png




和前文中sql的where解析后的AST树也是完全匹配的。思路来了,只要对sql解析生成的AST进行递归,即可得到这棵树。当然了,这里还可以进行一些优化,如果子节点的类型和父
节点的类型一致,例如都是and表达式或者都是or表达式,我们可以在生成dsl的时候将其作为并列的节点进行合并,这里不再赘述。


在递归中有这么几种情况:
AndExpr => bool must [{left}, {right}]
OrExpr => bool should [{left}, {right}]
ComparisonExpr => 一般是叶子节点
ParenBoolExpr => 指代括号表达式,其实内部是上述三种节点的某一种,所以直接取出内部节点按上述方法来处理


这样问题就变成了如何处理AST的叶子节点。前面提到了叶子节点实际上就是Comparison Expression。只要简单进行一些对应即可,下面是我们的项目里的一些对应关系,仅供参考:


convert.png

最后再附上demo
 
https://github.com/cch123/elasticsql
继续阅读 »
es现在几乎已经是开源搜索引擎的事实标准了,搭建简易,使用方便。不过在很多公司里(包括我司的部分部门),并不是把它当搜索引擎来用,而是当db来用。因为本身查询/搜索原理的区别,使es在千万或者亿级的数据中进行逻辑筛选相对高效。例如一些wms、工单查询系统,单表几十个甚至上百个字段,如果在数据库里为每种类型的查询都建立合适的索引,成本比较高,更不用说索引建多了还会影响到插入速度,后期的索引优化也是比较麻烦的问题。

不过如果把es当db来使的话,始终会有一个绕不过去的坎。就是es的DSL。让所有业务开发去学习dsl的话也不是不可以,但DSL真的有点反人类(不要打我)。简单的a and b或者a or b还比较容易写,如果我要的是a and (b and (c or d) and e)的查询逻辑,那我觉得谁写都会晕。即使是用官方或者第三方提供的client,如果需求多种多样的话,想要灵活地实现`需求=>DSL`的过程还是比较痛苦。

对于业务开发来说,当然是sql更平易近人(毕竟写了这么多年CRUD)。所以还有一种歪门邪道的流派,直接把sql转成DSL。要做sql和DSL转换的工作,需要进行sql的解析,先不要怵,这个年代找一个靠谱的sql parser还是比较容易的。比如阿里开源的druid连接池里的sql模块:
 
https://github.com/alibaba/dru ... d/sql

因为笔者的实现是用的下面这个golang版的parser:

https://github.com/xwb1989/sqlparser

所以用这个来举例吧~

这个是其作者从youtube/vitness里提取并进行改进的一个parser,我们能用到的是一部分子集功能,只需要解析select类的sql。

先举个简单的sql的例子:
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:""
}


sql的select语句在被解析之后生成一个Select的结构体,如果我们不关心使用者需要的字段的话,可以先把SelectExprs/Distinct/Comments/Lock里的内容忽略掉。如果不是分组统计类的需求,也可以先把GroupBy/Having忽略掉。这里我们关心的就剩下From、Where、OrderBy和Limit。

From对应的TableExprs实际上可以认为是简单的字符串,这里的值其实就是`x_order`。

OrderBy实际上是一个元素为
type Order struct {
Expr ValExpr
Direction string
}\

的数组。

Limit也很简单,
type Limit struct {
Offset, Rowcount ValExpr
}

其实就是俩数字。

那么剩下的就是这个Where结构了。where会被解析为AST(`https://en.wikipedia.org/wiki/Abstract_syntax_tree`),中文是抽象语法树。在不说子查询之类的情况下,这个AST也不会太复杂,毕竟where后面的情况比起编译原理里的程序语言来说情况还是要少得多的。以上述的sql为例,这里解析出来的Where结构是这样的:
&sqlparser.Where{
Type:"where",
Expr:(*sqlparser.ComparisonExpr)(0xc420016a50)
}


只有一个节点,一个ComparisonExpr表达式,这个ComparisonExpr,中文比较表达式,指代的就是我们sql里的`user_id = 1`。实际上我们可以认为这个"比较表达式"即是所有复杂AST的叶子节点。叶子结点在AST遍历的时候一般也就是递归的终点。因为这里的查询比较简单,整棵AST只有一个节点,即根节点和叶子节点都是这个ComparisonExpr。

再来一个复杂点的例子。
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}
}


稍微有一些二叉树的样子了吧。把这棵简单的树画出来:


Untitled1.png



回到文章开头的那个复杂的例子:
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)


看着真够麻烦的,我们把这棵树画出来:


Untitled.png



这样看着就直观多了。我们有了AST的结构,那要怎么对应到es的查询DSL呢?少安毋躁。

我们知道es的bool query是可以进行嵌套的,所以实际上我们可以同样可以构造出树形结构的bool query。这里把bool嵌套must和bool嵌套should简化一下,写成boolmust和boolshould:

例如a and (b and c)
query {
boolmust {
a,
boolmust {
b,
c
}
}
}


我们把query内部的第一个boolmust当作根节点,内部嵌套的a和另一个boolmust当作它的两个子节点,然后b和c又是这个boolmust的子节点。可以看出来,实际上这棵树和AST的节点可以一一对应。

再回到文章开头的例子,a and (b and (c or d) and e):
query {
boolmust {
a,
boolmust {
b,
boolshould {
c,
d
},
e
}
}
}

和前文中ast来做个简单的结构对比~


dsl和ast对比.png




和前文中sql的where解析后的AST树也是完全匹配的。思路来了,只要对sql解析生成的AST进行递归,即可得到这棵树。当然了,这里还可以进行一些优化,如果子节点的类型和父
节点的类型一致,例如都是and表达式或者都是or表达式,我们可以在生成dsl的时候将其作为并列的节点进行合并,这里不再赘述。


在递归中有这么几种情况:
AndExpr => bool must [{left}, {right}]
OrExpr => bool should [{left}, {right}]
ComparisonExpr => 一般是叶子节点
ParenBoolExpr => 指代括号表达式,其实内部是上述三种节点的某一种,所以直接取出内部节点按上述方法来处理


这样问题就变成了如何处理AST的叶子节点。前面提到了叶子节点实际上就是Comparison Expression。只要简单进行一些对应即可,下面是我们的项目里的一些对应关系,仅供参考:


convert.png

最后再附上demo
 
https://github.com/cch123/elasticsql 收起阅读 »

Day3: 《创建一个你自己的 Beat》

Elastic Advent 第三篇, 手头上事情实在太多,这两天正在进行权威指南翻译的冲刺阶段,临时填下坑,翻译官网的一篇文章吧(原文:https://www.elastic.co/blog/build-your-own-beat),Advent 规则很自由的,没说不能翻译文章啊,嘿嘿嘿,另外号召大家踊跃报名,大家一起玩才有意思。
 
活动地址:http://elasticsearch.cn/article/107
 
言归正传!
 Beat 是一个开源的用来构建轻量级数据汇集的平台,可用于将各种类型的数据发送至Elasticsearch 与 Logstash。我们有 Packetbeat 用于监控局域网内服务器之间的网络流量信息,有 Filebeat 收集服务器上的日志信息,还有新推出的 Metricbeat 可以定期获取外部系统的监控指标信息,除此以外,你还可以非常方便的基于 libbeat 框架来构建你属于自己的专属 Beat,目前 beas 社区已经有超过25个 Community Beats 了。

Elastic 还提供一个 Beat generator(Beat 生成器)来帮你快速构建属于你自己的 Beat。通过这篇博客你将会看到如何通过 Beat 生成器来快速创建一个你自己的 Beat。今天我们创建的是一个叫做 lsbeat 的 Beat,lsbeat 非常类似 Unix 系统下的命令行 ls,我们用 lsbeat 来索引目录和文件信息。本篇文章环境基于 Unix 系统,如果你是 Windows 或是其它系统,相关操作可能需要根据实际情况进行调整。

第一步 – 配置 Golang 环境

Beats 是用 Golang写的,显然,要创建和开发一个 beat,Golang 环境必不可少,关于这方面的文章很多,建议查看这篇 Golang 的安装向导: install Golang。当前 Beats 需要的最低版本是 Golang 1.6。另外请确保正确设置了你的 $GOPATH 环境变量。

另外 Golang Glide 被用来进行包的依赖管理,所以也需要确保正确安装,最低版本是 Glide 0.10.0,安装说明点这里。

让我们先来看看 lsbeat 将会用到的一段代码吧,这是一个简单的 golang 程序,通过命令行接收一个目录参数,然后列出该目录下的文件和子目录信息。
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())
}
}
}

后面我们将使用到这段代码和 listDir 函数。

第二步 – 生成项目

要生成一个你自己的 Beat,就要用到 beat-generator 了,首先你必须安装 cookiecutter。安装的详细说明看这里。安装好 cookiecutter 之后,我们要给自己的 Beat 起一个好听的名字,最好是小写的英文字母,我们今天这个例子就叫 lsbeat 吧。

生成项目模板之前,我们需要下载 Beats generator 包文件,就在 beats 仓库。安装好 GoLang 之后,你就可以很方便的使用 go get 命令来下载 Beats generator 包文件了。 当你执行下面的这个命令后,所有的源码文件都会下载到 $GOPATH/src 目录。
$ go get github.com/elastic/beats

在 GOPATH 下创建一个以你自己github账号名称命名的目录,并切换过去,然后执行 cookiecutter 命令并指向 Beat Generator 源码路径。
$ cd $GOPATH/src/github.com/{user}
$ cookiecutter $GOPATH/src/github.com/elastic/beats/generate/beat

Cookiecutter 接下来会问你几个问题,比如项目名称,我们输入:lsbeat;你的 github 用户名,输入你自己的 github 账户;还有两个关于beat和beat_path应该会自动识别,默认回车就好;最后的问题,你可以输入你的姓和名。
project_name [Examplebeat]: lsbeat
github_name [your-github-name]: {username}
beat [lsbeat]:
beat_path [github.com/{github id}]:
full_name [Firstname Lastname]: {Full Name}


现在应该已经创建好了一个名为 lsbeat 的目录,并且里面应该会生成一些文件,让我们一起来看一下吧,结构如下:
$ 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


我们刚刚已经生成好了一个原始的 Beat 模板了,但是你还需要获取相关的依赖和设置好 git 仓库。

首先,你需要拉取依赖的相关包信息,我们的这个例子是 lsbeat,我们先做一些的基本的配置,回头再看看详细看看其它的模板和配置文件,只需要执行 make setup 就可以自动获取依赖。
$ make setup


当你创建好了自己的 Beat 之后,记得上传到 github 仓库,并和社区进行分享哦,如下:


beats.png



要 push lsbeat 到你的 git 仓库,只需要执行如下命令:
$ git remote add origin git@github.com:{username}/lsbeat.git
$ git push -u origin master


恭喜你,现在你已经完成了一个 Beat ,并且发布了第一个版本到了 Github,不过里面还没有什么具体内容,现在让我们进一步看看里面的代码吧。

第四步 – 配置

执行过上面一系列命令之后,项目里将会自动创建名为 lsbeat.yml 和 lsbeat.template.json 的配置文件。所有的基本配置项都已经生成在了里面。
lsbeat.yml
lsbeat:
# Defines how often an event is sent to the output
period: 1s


Period 参数包含在每一个生成的 Beats 里面,它表示 lsbeat 将会每隔 1 秒钟轮询一次,这里我们修改 period 时间间隔为 10 秒。还可以在修改 etc/ 目录下面的 beat.yml 文件,这里新增一个 path 参数表示我们具体要监听哪个目录。
lsbeat:
# Defines how often an event is sent to the output
period: 10s
path: "."


参数添加好了之后,我们只需要运行 make update 命令就能让这些修改应用到配置文件lsbeat.yml。
$ make update
$ cat lsbeat.yml

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

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

lsbeat:
# Defines how often an event is sent to the output
period: 10s
path: "."
###############################################################################


修改完 yml 文件,记得修改 config/config.go文件,添加一个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: ".",
}


同时我们修改 period 默认时间间隔为 10 秒,默认监听的是当前目录 (.) 。.

第五步 – 添加代码

每一个 Beat 需要实现 Beater 接口,里面定义了 Run() 和 Stop() 函数。. 

我们可以定义一个名为 Lsbeat 的结构体,然后用这个对象实现 Beater 接口。然后添加字段 lastIndexTime 来保存最后运行的时间戳信息。
type Lsbeat struct {
done chan struct{}
config config.Config
client publisher.Client

lastIndexTime time.Time
...
}


另外,每个 Beat 还需要实现 New() 方法来接收 Beat 配置信息和返回 Lsbeat 的具体实例。
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
}


在我们的 lsbeat 例子中,我们要做的就是扩展默认的 Run() 函数来导出指定目录的文件和子目录信息。

在修改 Run() 函数之前,我们先在 lsbeat.go 增加 listDir() 函数,就是我们前面最开始测试的那段代码,用于收集目录和文件信息的简单例子稍微修改一下。另外我们还要生成以下字段信息:
"@timestamp": common.Time(time.Now()),
"type": beatname,
"modtime": common.Time(t),
"filename": f.Name(),
"path": dirFile + "/" + f.Name(),
"directory": f.IsDir(),
"filesize": f.Size(),


第一次运行的时候我们将索引所有的文件和目录信息,然后我们再定期检查是否有新文件被创建或者修改,再索引这些新创建的文件和目录。每次定期检查的时间戳都会保存在 lasIndexTime 变量,完整代码如下:
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)
}
}
}


记住在最开始需要导入 “io/ioutil” 包。
import (
"fmt"
"io/ioutil"
"time"
)


现在,让我们看看如何在 Run() 函数里面调用 listDir() 函数,并且保存时间戳到 lasIndexTime 变量。
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++
}

}


函数 Stop() 用来中断 run 的循环执行,保持默认生成的就行。
func (bt *Lsbeat) Stop() {
bt.client.Close()
close(bt.done)
}


到这里,编码部分基本就完成了。我们接下来添加新字段到 mapping 中,修改文件 etc/fields.yml。.
- 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


重新应用新的配置。

$ make update

字段 file_name 将使用 nGram 分词,我们还需要在文件 lsbeat.template.json 的 “settings” 节点添加一个自定义的 analyzer。
{
"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-*"
}


第六步 – 编译和运行

现在我们可以编译和运行了,只需要执行 make 命令就可以编译出可执行文件 lsbeat (lsbeat.exe on windows) 。

$ make

修改 lsbeat.yml 文件,设置需要监听的目录,如: “/Users/ec2-user/go”,记住是全路径。
lsbeat:
# Defines how often an event is sent to the output
period: 10s
path: "/Users/ec2-user/go"

同时确保你的 elasticsearch 和 kibana 正常运行。现在运行一下 lsbeat 命令看看会发生什么事情吧。

$ ./lsbeat

打开Kibana,通过调用 _cat 接口我们看看的索引是不是创建了。

beats-1.png


可以看到创建了一个名为 lsbeat-2016.06.03 的索引,并且看到已经有了一些文档了。现在对 filename 字段查询一下,由于使用的是 nGram 分词,支持模糊匹配,我们使用 lsbe 关键字搜一下。

beats-2.png


大功告成! 恭喜你,你已经完成了第一个属于你自己的 beat。
继续阅读 »
Elastic Advent 第三篇, 手头上事情实在太多,这两天正在进行权威指南翻译的冲刺阶段,临时填下坑,翻译官网的一篇文章吧(原文:https://www.elastic.co/blog/build-your-own-beat),Advent 规则很自由的,没说不能翻译文章啊,嘿嘿嘿,另外号召大家踊跃报名,大家一起玩才有意思。
 
活动地址:http://elasticsearch.cn/article/107
 
言归正传!
 Beat 是一个开源的用来构建轻量级数据汇集的平台,可用于将各种类型的数据发送至Elasticsearch 与 Logstash。我们有 Packetbeat 用于监控局域网内服务器之间的网络流量信息,有 Filebeat 收集服务器上的日志信息,还有新推出的 Metricbeat 可以定期获取外部系统的监控指标信息,除此以外,你还可以非常方便的基于 libbeat 框架来构建你属于自己的专属 Beat,目前 beas 社区已经有超过25个 Community Beats 了。

Elastic 还提供一个 Beat generator(Beat 生成器)来帮你快速构建属于你自己的 Beat。通过这篇博客你将会看到如何通过 Beat 生成器来快速创建一个你自己的 Beat。今天我们创建的是一个叫做 lsbeat 的 Beat,lsbeat 非常类似 Unix 系统下的命令行 ls,我们用 lsbeat 来索引目录和文件信息。本篇文章环境基于 Unix 系统,如果你是 Windows 或是其它系统,相关操作可能需要根据实际情况进行调整。

第一步 – 配置 Golang 环境

Beats 是用 Golang写的,显然,要创建和开发一个 beat,Golang 环境必不可少,关于这方面的文章很多,建议查看这篇 Golang 的安装向导: install Golang。当前 Beats 需要的最低版本是 Golang 1.6。另外请确保正确设置了你的 $GOPATH 环境变量。

另外 Golang Glide 被用来进行包的依赖管理,所以也需要确保正确安装,最低版本是 Glide 0.10.0,安装说明点这里。

让我们先来看看 lsbeat 将会用到的一段代码吧,这是一个简单的 golang 程序,通过命令行接收一个目录参数,然后列出该目录下的文件和子目录信息。
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())
}
}
}

后面我们将使用到这段代码和 listDir 函数。

第二步 – 生成项目

要生成一个你自己的 Beat,就要用到 beat-generator 了,首先你必须安装 cookiecutter。安装的详细说明看这里。安装好 cookiecutter 之后,我们要给自己的 Beat 起一个好听的名字,最好是小写的英文字母,我们今天这个例子就叫 lsbeat 吧。

生成项目模板之前,我们需要下载 Beats generator 包文件,就在 beats 仓库。安装好 GoLang 之后,你就可以很方便的使用 go get 命令来下载 Beats generator 包文件了。 当你执行下面的这个命令后,所有的源码文件都会下载到 $GOPATH/src 目录。
$ go get github.com/elastic/beats

在 GOPATH 下创建一个以你自己github账号名称命名的目录,并切换过去,然后执行 cookiecutter 命令并指向 Beat Generator 源码路径。
$ cd $GOPATH/src/github.com/{user}
$ cookiecutter $GOPATH/src/github.com/elastic/beats/generate/beat

Cookiecutter 接下来会问你几个问题,比如项目名称,我们输入:lsbeat;你的 github 用户名,输入你自己的 github 账户;还有两个关于beat和beat_path应该会自动识别,默认回车就好;最后的问题,你可以输入你的姓和名。
project_name [Examplebeat]: lsbeat
github_name [your-github-name]: {username}
beat [lsbeat]:
beat_path [github.com/{github id}]:
full_name [Firstname Lastname]: {Full Name}


现在应该已经创建好了一个名为 lsbeat 的目录,并且里面应该会生成一些文件,让我们一起来看一下吧,结构如下:
$ 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


我们刚刚已经生成好了一个原始的 Beat 模板了,但是你还需要获取相关的依赖和设置好 git 仓库。

首先,你需要拉取依赖的相关包信息,我们的这个例子是 lsbeat,我们先做一些的基本的配置,回头再看看详细看看其它的模板和配置文件,只需要执行 make setup 就可以自动获取依赖。
$ make setup


当你创建好了自己的 Beat 之后,记得上传到 github 仓库,并和社区进行分享哦,如下:


beats.png



要 push lsbeat 到你的 git 仓库,只需要执行如下命令:
$ git remote add origin git@github.com:{username}/lsbeat.git
$ git push -u origin master


恭喜你,现在你已经完成了一个 Beat ,并且发布了第一个版本到了 Github,不过里面还没有什么具体内容,现在让我们进一步看看里面的代码吧。

第四步 – 配置

执行过上面一系列命令之后,项目里将会自动创建名为 lsbeat.yml 和 lsbeat.template.json 的配置文件。所有的基本配置项都已经生成在了里面。
lsbeat.yml
lsbeat:
# Defines how often an event is sent to the output
period: 1s


Period 参数包含在每一个生成的 Beats 里面,它表示 lsbeat 将会每隔 1 秒钟轮询一次,这里我们修改 period 时间间隔为 10 秒。还可以在修改 etc/ 目录下面的 beat.yml 文件,这里新增一个 path 参数表示我们具体要监听哪个目录。
lsbeat:
# Defines how often an event is sent to the output
period: 10s
path: "."


参数添加好了之后,我们只需要运行 make update 命令就能让这些修改应用到配置文件lsbeat.yml。
$ make update
$ cat lsbeat.yml

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

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

lsbeat:
# Defines how often an event is sent to the output
period: 10s
path: "."
###############################################################################


修改完 yml 文件,记得修改 config/config.go文件,添加一个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: ".",
}


同时我们修改 period 默认时间间隔为 10 秒,默认监听的是当前目录 (.) 。.

第五步 – 添加代码

每一个 Beat 需要实现 Beater 接口,里面定义了 Run() 和 Stop() 函数。. 

我们可以定义一个名为 Lsbeat 的结构体,然后用这个对象实现 Beater 接口。然后添加字段 lastIndexTime 来保存最后运行的时间戳信息。
type Lsbeat struct {
done chan struct{}
config config.Config
client publisher.Client

lastIndexTime time.Time
...
}


另外,每个 Beat 还需要实现 New() 方法来接收 Beat 配置信息和返回 Lsbeat 的具体实例。
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
}


在我们的 lsbeat 例子中,我们要做的就是扩展默认的 Run() 函数来导出指定目录的文件和子目录信息。

在修改 Run() 函数之前,我们先在 lsbeat.go 增加 listDir() 函数,就是我们前面最开始测试的那段代码,用于收集目录和文件信息的简单例子稍微修改一下。另外我们还要生成以下字段信息:
"@timestamp": common.Time(time.Now()),
"type": beatname,
"modtime": common.Time(t),
"filename": f.Name(),
"path": dirFile + "/" + f.Name(),
"directory": f.IsDir(),
"filesize": f.Size(),


第一次运行的时候我们将索引所有的文件和目录信息,然后我们再定期检查是否有新文件被创建或者修改,再索引这些新创建的文件和目录。每次定期检查的时间戳都会保存在 lasIndexTime 变量,完整代码如下:
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)
}
}
}


记住在最开始需要导入 “io/ioutil” 包。
import (
"fmt"
"io/ioutil"
"time"
)


现在,让我们看看如何在 Run() 函数里面调用 listDir() 函数,并且保存时间戳到 lasIndexTime 变量。
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++
}

}


函数 Stop() 用来中断 run 的循环执行,保持默认生成的就行。
func (bt *Lsbeat) Stop() {
bt.client.Close()
close(bt.done)
}


到这里,编码部分基本就完成了。我们接下来添加新字段到 mapping 中,修改文件 etc/fields.yml。.
- 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


重新应用新的配置。

$ make update

字段 file_name 将使用 nGram 分词,我们还需要在文件 lsbeat.template.json 的 “settings” 节点添加一个自定义的 analyzer。
{
"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-*"
}


第六步 – 编译和运行

现在我们可以编译和运行了,只需要执行 make 命令就可以编译出可执行文件 lsbeat (lsbeat.exe on windows) 。

$ make

修改 lsbeat.yml 文件,设置需要监听的目录,如: “/Users/ec2-user/go”,记住是全路径。
lsbeat:
# Defines how often an event is sent to the output
period: 10s
path: "/Users/ec2-user/go"

同时确保你的 elasticsearch 和 kibana 正常运行。现在运行一下 lsbeat 命令看看会发生什么事情吧。

$ ./lsbeat

打开Kibana,通过调用 _cat 接口我们看看的索引是不是创建了。

beats-1.png


可以看到创建了一个名为 lsbeat-2016.06.03 的索引,并且看到已经有了一些文档了。现在对 filename 字段查询一下,由于使用的是 nGram 分词,支持模糊匹配,我们使用 lsbe 关键字搜一下。

beats-2.png


大功告成! 恭喜你,你已经完成了第一个属于你自己的 beat。 收起阅读 »

开源一组ES工具包,请大家帮忙宣传下,谢谢

wES
https://github.com/DataSays/wES/ 

wES 是一组开源的Java ElasticSearch客户端和工具; 简洁但是很勥 :)
wES = Java Retrofit2/OkHttp版本的客户端(不依赖Json类库,高度可定制) + 工具包 + spring-boot demo + 常用的ElasticSearch环境Dockerfile

wES 模块
wES 分割成许多模块, 可以按需选择.
wUtil: 一些有用的帮助类和工具类.
wES-client: 一个基于Retrofit2/OkHttp的Java客户端, 她是基于官方的ElasticSearch Rest API规范生成的. 她包含两种访问ElasticSearch的实现: OkHttp3版本的和Retrofit2版本. 而且她只依赖okhttp3/retrofit2,并且可以通过实现org.datasays.wes.core.IConvert接口支持任意一种Java Json类库. 你可以按照你的想法使用她.
wES-toolkit: 一些使用wES-client和ElasticSearch的工具包. 她包含了一套基于Gson的标准IConvert实现及封装代码库.
wES-demo: 一个Spring-boot + Vue.js的web应用, 用于展示一些通用的ElasticSearch使用场景.
wES-docker: 一些构建ElasticSearch开发/生产环境的常用Dockerfiles和shell脚本.

 
继续阅读 »
https://github.com/DataSays/wES/ 

wES 是一组开源的Java ElasticSearch客户端和工具; 简洁但是很勥 :)
wES = Java Retrofit2/OkHttp版本的客户端(不依赖Json类库,高度可定制) + 工具包 + spring-boot demo + 常用的ElasticSearch环境Dockerfile

wES 模块
wES 分割成许多模块, 可以按需选择.
wUtil: 一些有用的帮助类和工具类.
wES-client: 一个基于Retrofit2/OkHttp的Java客户端, 她是基于官方的ElasticSearch Rest API规范生成的. 她包含两种访问ElasticSearch的实现: OkHttp3版本的和Retrofit2版本. 而且她只依赖okhttp3/retrofit2,并且可以通过实现org.datasays.wes.core.IConvert接口支持任意一种Java Json类库. 你可以按照你的想法使用她.
wES-toolkit: 一些使用wES-client和ElasticSearch的工具包. 她包含了一套基于Gson的标准IConvert实现及封装代码库.
wES-demo: 一个Spring-boot + Vue.js的web应用, 用于展示一些通用的ElasticSearch使用场景.
wES-docker: 一些构建ElasticSearch开发/生产环境的常用Dockerfiles和shell脚本.

  收起阅读 »

Day2:《Kibana 系漫游指南》

大家好,欢迎你们来到 ELK 三体星系的第二天。昨天,wood 送给大家一本脚踏实地的生存指南,今天让我们仰望星空,由我给大家介绍一下围绕在 Kibana 身边的诸多行星们~

Kibana Plugin 类型简介

我们最熟悉的 Kibana Plugin,其实就是 Kibana 本身~ Kibana 提供了一整套框架,我们可以在此基础上,开发诸多不同类型的插件,包括:
  • app
  • visTypes
  • spyModes
  • fieldFormatter


列这么几个源码里的名词出来可能大家觉得比较晦涩。其实呢,app 就是同时具有前后端实现的应用,在 Kibana 5 里,默认分发的 app 有四个:实现日志查询和可视化的 kibana app、实现时序指标统计和可视化的 timelion app、实现和 ES 接口交互命令的 console app、在有异常的时候才看得到的状态页面 status_page app。

而 visTypes 则是在 kibana 中具体可用的可视化效果。默认分发的有:kbn_vislib_vis_types、metric、table、markdown。我们常用的那些由 D3.js 完成的饼图线图地图,都是在 kbn_vislib_vis_types 中完成的。

fieldFormatter 则用来定义在 ES 中相同类型的数据,根据其实际含义,可以有不同的展示方式。比如说:URL 肯定是一个字符串,但是可以用 fieldFormatter 把它在页面展示的时候,加上 `<a href></a>` 的样式,让人一键点击;同理,还可以过滤判断一下图片类 URL,加上 `<img src></img>` 的样式,直接在 Kibana 界面上就看图片内容~~

官方的我们会看手册啦~

好啦好啦,我也不会真的去抄一把官方手册假冒《Kibana 系漫游指南》来骗你们流量的。下面给大家介绍一些社区开源的,让你绝对眼前一亮的各种新奇扩展:

1. logtrail
    这是一个 app 插件,创意来自 papertrail 公司的产品。完全的满足了 Geeker 们喜欢黑底白字终端的癖好~不过其实实现非常简单:每隔 10 秒请求一次最近 500 条日志就是啦!

2. vectormap
    这是一个 visType 插件,也就是我们在 Kibana3 里曾经用过的 map panel 效果。这个插件不被官方直接采用的一个原因是版权许可问题。不做商用的情况下,这个插件还是可以极大方便我们做行政区域的访问情况统计和展示的。
    
3. kbn_network
    这也是一个 visType 插件,酷毙了的网状图效果!通过不同的 aggs 数据展示 node 和 relational。
    注意这个跟 Elastic 的 graph 并不是完全一致的东西。该插件要求你本身的数据已经有直接的关联可用。
    
4. sentinl
    这是一个同时带有 spyMode 和 app 双插件的项目。其基础思路是参照 Elastic 的 Watcher 接口,但是将监控告警的进程从 ES 挪到 Kibana 里。同时还可以通过 phantomjs 做到截图报表。
    
    这个项目最大的特点,是通过 spyMode 插件,大大降低了配置告警规则的复杂度。这个扩展让你可以在 Kibana 上配置任意聚合效果之后,就地点击定义当前聚合语句为告警规则!
    
5. kibana-keynote
    这是另一个剑走偏锋的 app 插件,出自 Kibana 作者本人之手。它的作用是:播放 keynote 演讲稿!事实上项目里放的演讲稿就是作者本人在 ELastic{ON} 2016 上用的。让我们猜一猜下周的大会上,他会不会就用这个插件给我们分享呢?

今天就先讲这几颗最闪亮的星了~有兴趣了解更多 Kibana 行星的游客,欢迎阅读全本《Kibana系漫游指南》

也欢迎观看 Kibana 行星的《探索·发现》节目哟~
继续阅读 »
大家好,欢迎你们来到 ELK 三体星系的第二天。昨天,wood 送给大家一本脚踏实地的生存指南,今天让我们仰望星空,由我给大家介绍一下围绕在 Kibana 身边的诸多行星们~

Kibana Plugin 类型简介

我们最熟悉的 Kibana Plugin,其实就是 Kibana 本身~ Kibana 提供了一整套框架,我们可以在此基础上,开发诸多不同类型的插件,包括:
  • app
  • visTypes
  • spyModes
  • fieldFormatter


列这么几个源码里的名词出来可能大家觉得比较晦涩。其实呢,app 就是同时具有前后端实现的应用,在 Kibana 5 里,默认分发的 app 有四个:实现日志查询和可视化的 kibana app、实现时序指标统计和可视化的 timelion app、实现和 ES 接口交互命令的 console app、在有异常的时候才看得到的状态页面 status_page app。

而 visTypes 则是在 kibana 中具体可用的可视化效果。默认分发的有:kbn_vislib_vis_types、metric、table、markdown。我们常用的那些由 D3.js 完成的饼图线图地图,都是在 kbn_vislib_vis_types 中完成的。

fieldFormatter 则用来定义在 ES 中相同类型的数据,根据其实际含义,可以有不同的展示方式。比如说:URL 肯定是一个字符串,但是可以用 fieldFormatter 把它在页面展示的时候,加上 `<a href></a>` 的样式,让人一键点击;同理,还可以过滤判断一下图片类 URL,加上 `<img src></img>` 的样式,直接在 Kibana 界面上就看图片内容~~

官方的我们会看手册啦~

好啦好啦,我也不会真的去抄一把官方手册假冒《Kibana 系漫游指南》来骗你们流量的。下面给大家介绍一些社区开源的,让你绝对眼前一亮的各种新奇扩展:

1. logtrail
    这是一个 app 插件,创意来自 papertrail 公司的产品。完全的满足了 Geeker 们喜欢黑底白字终端的癖好~不过其实实现非常简单:每隔 10 秒请求一次最近 500 条日志就是啦!

2. vectormap
    这是一个 visType 插件,也就是我们在 Kibana3 里曾经用过的 map panel 效果。这个插件不被官方直接采用的一个原因是版权许可问题。不做商用的情况下,这个插件还是可以极大方便我们做行政区域的访问情况统计和展示的。
    
3. kbn_network
    这也是一个 visType 插件,酷毙了的网状图效果!通过不同的 aggs 数据展示 node 和 relational。
    注意这个跟 Elastic 的 graph 并不是完全一致的东西。该插件要求你本身的数据已经有直接的关联可用。
    
4. sentinl
    这是一个同时带有 spyMode 和 app 双插件的项目。其基础思路是参照 Elastic 的 Watcher 接口,但是将监控告警的进程从 ES 挪到 Kibana 里。同时还可以通过 phantomjs 做到截图报表。
    
    这个项目最大的特点,是通过 spyMode 插件,大大降低了配置告警规则的复杂度。这个扩展让你可以在 Kibana 上配置任意聚合效果之后,就地点击定义当前聚合语句为告警规则!
    
5. kibana-keynote
    这是另一个剑走偏锋的 app 插件,出自 Kibana 作者本人之手。它的作用是:播放 keynote 演讲稿!事实上项目里放的演讲稿就是作者本人在 ELastic{ON} 2016 上用的。让我们猜一猜下周的大会上,他会不会就用这个插件给我们分享呢?

今天就先讲这几颗最闪亮的星了~有兴趣了解更多 Kibana 行星的游客,欢迎阅读全本《Kibana系漫游指南》

也欢迎观看 Kibana 行星的《探索·发现》节目哟~ 收起阅读 »

Day1: 大规模Elasticsearch集群管理心得

ElasticSearch目前在互联网公司主要用于两种应用场景,其一是用于构建业务的搜索功能模块且多是垂直领域的搜索,数据量级一般在千万至数十亿这个级别;其二用于大规模数据的实时OLAP,经典的如ELKStack,数据规模可能达到千亿或更多。 这两种场景的数据索引和应用访问模式上差异较大,在硬件选型和集群优化方面侧重点也会有所不同。一般来说后一种场景属于大数据范畴,数据量级和集群规模更大,在管理方面也更有挑战。

应Medcl大大的邀请,为ES中文社区做今年的Advent开篇,分享一下我在管理自家公司用于日志分析的ES集群方面的一点心得,蜻蜓点水,泛泛而谈,希望大方向上能对大家提供一些帮助。

这里的自家,即是携程旅行网。从2013年开始接触ES,我们团队先后实践过0.9.x -> 5.0.0中间各个版本,从最初只用于运维内部IIS日志的分析,到如今支持IT、呼叫中心、安全、测试、业务研发等多个部门超过200种日志型数据的实时检索与分析。 一路走来,愉悦了大家,也死磕了自己。

目前我们最大的日志单集群有120个data node,运行于70台物理服务器上。数据规模如下:
  • 单日索引数据条数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


运维这样大规模的ES集群,有哪些值得注意的地方?

一. 必不可少的工具
工欲善其事必先利其器,从一开始,哪怕就只有几个node,就应该使用分布式配置管理工具来做集群的部署。随着应用的成熟,集群规模的逐步扩大,效率的提升会凸显。 官方提供了ES Puppet Module和Chef Cookbook,熟悉这两个工具的同学可以直接拿过来用。 我们自己则是采用的Ansible,编写了一套Playbook来达到类似的效果。 用熟这类工具,对于集群的初始部署,配置批量更改,集群版本升级,重启故障结点都会快捷和安全许多。
第二个必备利器就是sense插件。通过这个插件直接调用集群的restful API,在做集群和索引的状态查看,索引配置更改的时候非常方便。语法提示和自动补全功能更是实用,减少了翻看文档的频率。在Kibana5里面,sense已经成为一个内置的控制台,无需额外安装。

二. 硬件配置
我们采用的是32vcoreCPU + 128GB RAM的服务器,磁盘配置大部分服务器是12块4TB SATA机械磁盘做的Raid0,少部分机器是刚上了不久的6块800GB SSD raid0,主要目的是想做冷热数据分离,后面谈到集群架构的时候,再进一步解释一下如何利用硬件资源。

三. 集群的管理
  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,可以提升聚合速度。


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


升级第一周,我们的冷数据结点出现间歇性不响应问题,从而刨出3个issue提交给官方:
Issue#21595 Issue#21612 Issue#21611
第一个问题确认为Bug,将在5.0.2修复,其他两个目前还不清楚根源,看起来也只在我们的应用场景里遇到了。所幸问题都找到了了规避措施,实施这些措施以后,最近一周我们的集群重新回到以前2.4版本时期的稳定状态。


五. 监控
不差钱没空折腾的建议还是买官方的xpack省心,有精力折腾的,利用ES各种丰富的stats api,用自己熟悉的监控工具采集数据,可视化出来就好了。 那么多监控指标,最最关键的还是以下几类:
  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全部记录下来。通过分析访问记录,可以在集群出现性能问题时,快速找到问题根源,对于问题排查和性能优化都很有帮助。


最后就是多上手实践,遇到问题多查官方资料,多Google看是否有其他人遇到同类问题,精力充足有编程背景的同学也可以多刨刨源码。
继续阅读 »
ElasticSearch目前在互联网公司主要用于两种应用场景,其一是用于构建业务的搜索功能模块且多是垂直领域的搜索,数据量级一般在千万至数十亿这个级别;其二用于大规模数据的实时OLAP,经典的如ELKStack,数据规模可能达到千亿或更多。 这两种场景的数据索引和应用访问模式上差异较大,在硬件选型和集群优化方面侧重点也会有所不同。一般来说后一种场景属于大数据范畴,数据量级和集群规模更大,在管理方面也更有挑战。

应Medcl大大的邀请,为ES中文社区做今年的Advent开篇,分享一下我在管理自家公司用于日志分析的ES集群方面的一点心得,蜻蜓点水,泛泛而谈,希望大方向上能对大家提供一些帮助。

这里的自家,即是携程旅行网。从2013年开始接触ES,我们团队先后实践过0.9.x -> 5.0.0中间各个版本,从最初只用于运维内部IIS日志的分析,到如今支持IT、呼叫中心、安全、测试、业务研发等多个部门超过200种日志型数据的实时检索与分析。 一路走来,愉悦了大家,也死磕了自己。

目前我们最大的日志单集群有120个data node,运行于70台物理服务器上。数据规模如下:
  • 单日索引数据条数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


运维这样大规模的ES集群,有哪些值得注意的地方?

一. 必不可少的工具
工欲善其事必先利其器,从一开始,哪怕就只有几个node,就应该使用分布式配置管理工具来做集群的部署。随着应用的成熟,集群规模的逐步扩大,效率的提升会凸显。 官方提供了ES Puppet Module和Chef Cookbook,熟悉这两个工具的同学可以直接拿过来用。 我们自己则是采用的Ansible,编写了一套Playbook来达到类似的效果。 用熟这类工具,对于集群的初始部署,配置批量更改,集群版本升级,重启故障结点都会快捷和安全许多。
第二个必备利器就是sense插件。通过这个插件直接调用集群的restful API,在做集群和索引的状态查看,索引配置更改的时候非常方便。语法提示和自动补全功能更是实用,减少了翻看文档的频率。在Kibana5里面,sense已经成为一个内置的控制台,无需额外安装。

二. 硬件配置
我们采用的是32vcoreCPU + 128GB RAM的服务器,磁盘配置大部分服务器是12块4TB SATA机械磁盘做的Raid0,少部分机器是刚上了不久的6块800GB SSD raid0,主要目的是想做冷热数据分离,后面谈到集群架构的时候,再进一步解释一下如何利用硬件资源。

三. 集群的管理
  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,可以提升聚合速度。


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


升级第一周,我们的冷数据结点出现间歇性不响应问题,从而刨出3个issue提交给官方:
Issue#21595 Issue#21612 Issue#21611
第一个问题确认为Bug,将在5.0.2修复,其他两个目前还不清楚根源,看起来也只在我们的应用场景里遇到了。所幸问题都找到了了规避措施,实施这些措施以后,最近一周我们的集群重新回到以前2.4版本时期的稳定状态。


五. 监控
不差钱没空折腾的建议还是买官方的xpack省心,有精力折腾的,利用ES各种丰富的stats api,用自己熟悉的监控工具采集数据,可视化出来就好了。 那么多监控指标,最最关键的还是以下几类:
  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全部记录下来。通过分析访问记录,可以在集群出现性能问题时,快速找到问题根源,对于问题排查和性能优化都很有帮助。


最后就是多上手实践,遇到问题多查官方资料,多Google看是否有其他人遇到同类问题,精力充足有编程背景的同学也可以多刨刨源码。 收起阅读 »