elasticsearch使用river同步mysql数据,schedule设置问题
Elasticsearch • znketophawk 回复了问题 • 1 人关注 • 2 个回复 • 4925 次浏览 • 2015-12-29 16:46
我用RPM包安装，但凡安装插件后，ES就连接不上
Elasticsearch • medcl 回复了问题 • 2 人关注 • 1 个回复 • 4384 次浏览 • 2015-12-29 14:22
ES查询增加排序效率低下
Elasticsearch • zttech 回复了问题 • 5 人关注 • 4 个回复 • 10502 次浏览 • 2016-01-04 21:00
Day24: Elasticsearch添加Shield后TransportClient如何连接？
Advent • medcl 发表了文章 • 6 个评论 • 7995 次浏览 • 2015-12-28 12:13
Shield是Elasticsearch一个安全防护插件,提供了权限访问控制和日志审计功能,企业可以很方便的和LDAP或是ActiveDirectory进行集成,重用现有的安全认证体系.
Elasticsearch使用了Shield后,Elasticsearch就需要权限才能访问了,和默认的调用方式有些不同,下面简单介绍一下HTTP和TCP两种方式的连接.
关于Shield的安装和配置我这里不就具体介绍,创建了一个用户名和密码都是tribe_user的用户,权限是admin.
1.HTTP方式
现在直接访问es的http接口就会报错
curl http://localhost:9200
{"error":{"root_cause":[{"type":"security_exception","reason":"missing authentication token for REST request [/]","header":{"WWW-Authenticate":"Basic realm=\"shield\""}}],"type":"security_exception","reason":"missing authentication token for REST request [/]","header":{"WWW-Authenticate":"Basic realm=\"shield\""}},"status":401}
shield支持HttpBasic验证,所以正确的访问姿势是:
curl -u tribe_user:tribe_user http://localhost:9200 { "name" : "Melter", "cluster_name" : "elasticsearch", "version" : { "number" : "2.1.1", "build_hash" : "805c528f3167980046f224310f9147fa745e5371", "build_timestamp" : "2015-12-09T20:23:16Z", "build_snapshot" : false, "lucene_version" : "5.3.1" }, "tagline" : "You Know, for Search" }
如果是浏览器访问的话,第一次访问会弹出验证窗口,后续只要不关闭这个浏览器保持这个session就能一直访问.
注意http basic是不安全的认证方式,仅供开发调试使用,生产环境还需要结合HTTPS的加密通道使用.
2.TransportClient方式的访问Shield加防的Elasticsearch,稍微麻烦点,需要依赖Shield的包,步骤如下:
2.1 如果你是maven管理的项目,在pom.xml文件里添加Elasticsearch的maven仓库源,如下:
<repositories>
<repository>
<id>elasticsearch-releases</id>
<url>https://maven.elasticsearch.or ... gt%3B
<releases> <enabled>true</enabled> </releases>
<snapshots> <enabled>false</enabled> </snapshots>
</repository>
</repositories>
2.2 添加依赖的配置
<dependency>
<groupId>org.elasticsearch.plugin</groupId>
<artifactId>shield</artifactId>
<version>2.1.1</version>
</dependency
2.3 构建TransportClient的地方增加访问用户的配置
import org.elasticsearch.shield.ShieldPlugin; import org.elasticsearch.shield.authc.support.SecuredString; import static org.elasticsearch.shield.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
String clusterName="elasticsearch"; String ip= "127.0.0.1";
Settings settings = Settings.settingsBuilder()
.put("cluster.name", clusterName)
.put("shield.user", "tribe_user:tribe_user")
.build();
try { client = TransportClient.builder()
.addPlugin(ShieldPlugin.class)
.settings(settings).build()
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip),9300));
String token = basicAuthHeaderValue("tribe_user", new SecuredString("tribe_user".toCharArray())); client.prepareSearch()
.putHeader("Authorization", token).get(); }
catch (UnknownHostException e)
{ logger.error("es",e); }
现在的编辑器贴代码有点恶心,可以看这里:
http://log.medcl.net/item/2015 ... -1252
Elasticsearch使用了Shield后,Elasticsearch就需要权限才能访问了,和默认的调用方式有些不同,下面简单介绍一下HTTP和TCP两种方式的连接.
关于Shield的安装和配置我这里不就具体介绍,创建了一个用户名和密码都是tribe_user的用户,权限是admin.
1.HTTP方式
现在直接访问es的http接口就会报错
curl http://localhost:9200
{"error":{"root_cause":[{"type":"security_exception","reason":"missing authentication token for REST request [/]","header":{"WWW-Authenticate":"Basic realm=\"shield\""}}],"type":"security_exception","reason":"missing authentication token for REST request [/]","header":{"WWW-Authenticate":"Basic realm=\"shield\""}},"status":401}
shield支持HttpBasic验证,所以正确的访问姿势是:
curl -u tribe_user:tribe_user http://localhost:9200 { "name" : "Melter", "cluster_name" : "elasticsearch", "version" : { "number" : "2.1.1", "build_hash" : "805c528f3167980046f224310f9147fa745e5371", "build_timestamp" : "2015-12-09T20:23:16Z", "build_snapshot" : false, "lucene_version" : "5.3.1" }, "tagline" : "You Know, for Search" }
如果是浏览器访问的话,第一次访问会弹出验证窗口,后续只要不关闭这个浏览器保持这个session就能一直访问.
注意http basic是不安全的认证方式,仅供开发调试使用,生产环境还需要结合HTTPS的加密通道使用.
2.TransportClient方式的访问Shield加防的Elasticsearch,稍微麻烦点,需要依赖Shield的包,步骤如下:
2.1 如果你是maven管理的项目,在pom.xml文件里添加Elasticsearch的maven仓库源,如下:
<repositories>
<repository>
<id>elasticsearch-releases</id>
<url>https://maven.elasticsearch.or ... gt%3B
<releases> <enabled>true</enabled> </releases>
<snapshots> <enabled>false</enabled> </snapshots>
</repository>
</repositories>
2.2 添加依赖的配置
<dependency>
<groupId>org.elasticsearch.plugin</groupId>
<artifactId>shield</artifactId>
<version>2.1.1</version>
</dependency
2.3 构建TransportClient的地方增加访问用户的配置
import org.elasticsearch.shield.ShieldPlugin; import org.elasticsearch.shield.authc.support.SecuredString; import static org.elasticsearch.shield.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
String clusterName="elasticsearch"; String ip= "127.0.0.1";
Settings settings = Settings.settingsBuilder()
.put("cluster.name", clusterName)
.put("shield.user", "tribe_user:tribe_user")
.build();
try { client = TransportClient.builder()
.addPlugin(ShieldPlugin.class)
.settings(settings).build()
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip),9300));
String token = basicAuthHeaderValue("tribe_user", new SecuredString("tribe_user".toCharArray())); client.prepareSearch()
.putHeader("Authorization", token).get(); }
catch (UnknownHostException e)
{ logger.error("es",e); }
现在的编辑器贴代码有点恶心,可以看这里:
http://log.medcl.net/item/2015 ... -1252
感谢elastic送来的圣诞礼物！
默认分类 • atomyliu 发表了文章 • 2 个评论 • 4066 次浏览 • 2015-12-28 10:01
圣诞节收到了elastic圣诞老人medcl送来的圣诞礼物，在此特别感谢！
发个赞吧！
晒图@！
居然有个瓶起子，看来不喝一台是不行了。。。
sugru据说是什么都能粘的‘硅胶’。。。
elastic的周边越来越强大了。
特别感谢medcl大神对中国地区elastic用户的关照！
发个赞吧！
晒图@！
居然有个瓶起子，看来不喝一台是不行了。。。
sugru据说是什么都能粘的‘硅胶’。。。
elastic的周边越来越强大了。
特别感谢medcl大神对中国地区elastic用户的关照！
大家好我用elasticsearch mapper-attachment然后查询很慢，我把返回的字段都限制了还是很慢，有没得人知道啊
Elasticsearch • paopao 回复了问题 • 3 人关注 • 2 个回复 • 4557 次浏览 • 2015-12-28 10:46
elasticsearch安装插件异常
Elasticsearch • suwensen 回复了问题 • 4 人关注 • 3 个回复 • 9221 次浏览 • 2016-01-28 16:06
Sql 语法转换es node版本
Elasticsearch • wwfalcon 回复了问题 • 6 人关注 • 3 个回复 • 7788 次浏览 • 2016-03-17 15:31
elasticsearch使用bucket_script查询的结果如何排序
Elasticsearch • zpzkit 回复了问题 • 4 人关注 • 6 个回复 • 11637 次浏览 • 2019-09-17 15:15
特殊名称的type如何删除
Elasticsearch • 三斗室 回复了问题 • 2 人关注 • 1 个回复 • 4860 次浏览 • 2015-12-25 17:46
Day 23 谈谈ES 的Recovery
Advent • kennywu76 发表了文章 • 12 个评论 • 17071 次浏览 • 2015-12-25 16:45
Note: 本文针对ES2.x
Recovery是指将一个索引的未分配shard分配到一个结点的过程。 在快照恢复，更改索引复制片数量，结点故障或者结点启动时发生。由于master持有整个集群的状态信息，因此可以判断出哪些shard需要做再分配，以及分配到哪个结点。例如:
通过CAT health API，我们可以查看集群的状态，从而获知数据的完整性情况:
可能的状态及含义：
Recovery过程要消耗额外的资源，CPU、内存、结点之间的网络带宽等等。 这些额外的资源消耗，有可能会导致集群的服务能力降级，或者一部分功能暂时不可用。了解一些Recovery的过程和相关的配置参数，对于减小recovery带来的资源消耗，加快集群恢复过程都是很有帮助的。
减少集群Full Restart造成的数据来回拷贝
集群可能会有整体重启的需要，比如需要升级硬件、升级操作系统或者升级ES大版本。重启所有结点可能带来的一个问题: 某些结点可能先于其他结点加入集群。 先加入集群的结点可能已经可以选举好master，并立即启动了recovery的过程，由于这个时候整个集群数据还不完整，master会指示一些结点之间相互开始复制数据。 那些晚到的结点，一旦发现本地的数据已经被复制到其他结点，则直接删除掉本地“失效”的数据。 当整个集群恢复完毕后，数据分布不均衡显然是不均衡的，master会触发rebalance过程，将数据在结点之间挪动。整个过程无谓消耗了大量的网络流量。 合理设置recovery相关参数则可以防范这种问题的发生。
以上三个参数是说集群里一旦有多少个结点就立即开始recovery过程。 不同之处在于，第一个参数指的是master或者data都算在内，而后面两个参数则分指master和data node。
在期待的节点数条件满足之前, recovery过程会等待gateway.recover_after_time (默认5分钟) 这么长时间，一旦等待超时，则会根据以下条件判断是否启动:
举例来说，对于一个有10个data node的集群，如果有以下的设置:
那么集群5分钟以内10个data node都加入了，或者5分钟以后8个以上的data node加入了，都会立即启动recovery过程。
减少主副本之间的数据复制
如果不是full restart，而是重启单个data node，仍然会造成数据在不同结点之间来回复制。为避免这个问题，可以在重启之前，先关闭集群的shard allocation:
然后在结点重启完成加入集群后，再重新打开:
这样在结点重启完成后，尽量多的从本地直接恢复数据。
但是在ES1.6版本之前，即使做了以上措施，仍然会发现有大量主副本之间的数据拷贝。从表面去看，这点很让人不能理解。 主副本数据完全一致，ES应该直接从副本本地恢复数据就好了，为什么要重新从主片再复制一遍呢？ 原因在于Recovery是简单对比主副本的segment file来判断哪些数据一致可以本地恢复，哪些不一致需要远端拷贝的。而不同结点的segment merge是完全独立运行的，可能导致主副本merge的深度不完全一样，从而造成即使文档集完全一样，产生的segment file却不完全一样。
为了解决这个问题，ES1.6版本以后加入了synced flush的新特性。 对于5分钟没有更新过的shard，会自动synced flush一下，实质是为对应的shard加了一个synced flush ID。这样当重启结点的时候，先对比一下shard的synced flush ID，就可以知道两个shard是否完全相同，避免了不必要的segment file拷贝，极大加快了冷索引的恢复速度。
需要注意的是synced flush只对冷索引有效，对于热索引（5分钟内有更新的索引）没有作用。 如果重启的结点包含有热索引，那么还是免不了大量的文件拷贝。因此在重启一个结点之前，最好按照以下步骤执行，recovery几乎可以瞬间完成:
(特别大的)热索引为何恢复慢
对于冷索引，由于数据不再更新，利用synced flush特性，可以快速直接从本地恢复数据。 而对于热索引，特别是shard很大的热索引，除了synced flush派不上用场需要大量跨结点拷贝segment file以外，translog recovery是导致慢的更重要的原因。
从主片恢复数据到副片需要经历3个阶段:
可见，在recovery完成之前，translog是不能够被清除掉的（禁用掉正常运作期间后台的flush操作）。如果shard比较大，第一阶段耗时很长，会导致此阶段产生的translog很大。重放translog比起简单的文件拷贝耗时要长得多，因此第二阶段的translog耗时也会显著增加。等到第三阶段，需要重放的translog可能会比第二阶段还要多。 而第三阶段是会阻塞新索引写入的，在对写入实时性要求很高的场合，就会非常影响用户体验。 因此，要加快大的热索引恢复速度，最好的方式是遵从上一节提到的方法: 暂停新数据写入，手动sync flush，等待数据恢复完成后，重新开启数据写入，这样可以将数据延迟影响可以降到最低。
万一遇到Recovery慢，想知道进度怎么办呢？ CAT Recovery API可以显示详细的recovery各个阶段的状态。 这个API怎么用就不在这里赘述了，参考: CAT Recovery
其他Recovery相关的专家级设置
还有其他一些专家级的设置（参见： recovery）可以影响recovery的速度，但提升速度的代价是更多的资源消耗，因此在生产集群上调整这些参数需要结合实际情况谨慎调整，一旦影响应用要立即调整回来。 对于搜索并发量要求高，延迟要求低的场合，默认设置一般就不要去动了。 对于日志实时分析类对于搜索延迟要求不高，但对于数据写入延迟期望比较低的场合，可以适当调大indices.recovery.max_bytes_per_sec，提升recovery速度，减少数据写入被阻塞的时长。
最后要说的一点是ES的版本迭代很快，对于Recovery的机制也在不断的优化中。 其中有一些版本甚至引入了一些bug，比如在ES1.4.x有严重的translog recovery bug，导致大的索引trans log recovery几乎无法完成 （issue #9226） 。因此实际使用中如果遇到问题，最好在Github的issue list里搜索一下，看是否使用的版本有其他人反映同样的问题。
Recovery是指将一个索引的未分配shard分配到一个结点的过程。 在快照恢复，更改索引复制片数量，结点故障或者结点启动时发生。由于master持有整个集群的状态信息，因此可以判断出哪些shard需要做再分配，以及分配到哪个结点。例如:
- 如果某个shard主片在，副片所在结点挂了，那么选择另外一个可用结点，将副片分配(allocate)上去，然后进行主从片的复制。
- 如果某个shard的主片所在结点挂了，副片还在，那么将副片升级为主片，然后做主副复制。
- 如果某个shard的主副片所在结点都挂了，则暂时无法恢复，等待持有相关数据的结点重新加入集群后，从结点上恢复主分片，再选择某个结点分配复制片，并从主分片同步数据。
通过CAT health API，我们可以查看集群的状态，从而获知数据的完整性情况:
可能的状态及含义：
Green: 所有的shard主副片都完好的
Yellow: 所有shard的主片都完好，部分副片没有了，数据完整性依然完好。
Red: 某些shard的主副片都没有了，对应的索引数据不完整
Recovery过程要消耗额外的资源，CPU、内存、结点之间的网络带宽等等。 这些额外的资源消耗，有可能会导致集群的服务能力降级，或者一部分功能暂时不可用。了解一些Recovery的过程和相关的配置参数，对于减小recovery带来的资源消耗，加快集群恢复过程都是很有帮助的。
减少集群Full Restart造成的数据来回拷贝
集群可能会有整体重启的需要，比如需要升级硬件、升级操作系统或者升级ES大版本。重启所有结点可能带来的一个问题: 某些结点可能先于其他结点加入集群。 先加入集群的结点可能已经可以选举好master，并立即启动了recovery的过程，由于这个时候整个集群数据还不完整，master会指示一些结点之间相互开始复制数据。 那些晚到的结点，一旦发现本地的数据已经被复制到其他结点，则直接删除掉本地“失效”的数据。 当整个集群恢复完毕后，数据分布不均衡显然是不均衡的，master会触发rebalance过程，将数据在结点之间挪动。整个过程无谓消耗了大量的网络流量。 合理设置recovery相关参数则可以防范这种问题的发生。
gateway.expected_nodes
gateway.expected_master_nodes
gateway.expected_data_nodes
以上三个参数是说集群里一旦有多少个结点就立即开始recovery过程。 不同之处在于，第一个参数指的是master或者data都算在内，而后面两个参数则分指master和data node。
在期待的节点数条件满足之前, recovery过程会等待gateway.recover_after_time (默认5分钟) 这么长时间，一旦等待超时，则会根据以下条件判断是否启动:
gateway.recover_after_nodes
gateway.recover_after_master_nodes
gateway.recover_after_data_nodes
举例来说，对于一个有10个data node的集群，如果有以下的设置:
gateway.expected_data_nodes: 10
gateway.recover_after_time: 5m
gateway.recover_after_data_nodes: 8
那么集群5分钟以内10个data node都加入了，或者5分钟以后8个以上的data node加入了，都会立即启动recovery过程。
减少主副本之间的数据复制
如果不是full restart，而是重启单个data node，仍然会造成数据在不同结点之间来回复制。为避免这个问题，可以在重启之前，先关闭集群的shard allocation:
然后在结点重启完成加入集群后，再重新打开:
这样在结点重启完成后，尽量多的从本地直接恢复数据。
但是在ES1.6版本之前，即使做了以上措施，仍然会发现有大量主副本之间的数据拷贝。从表面去看，这点很让人不能理解。 主副本数据完全一致，ES应该直接从副本本地恢复数据就好了，为什么要重新从主片再复制一遍呢？ 原因在于Recovery是简单对比主副本的segment file来判断哪些数据一致可以本地恢复，哪些不一致需要远端拷贝的。而不同结点的segment merge是完全独立运行的，可能导致主副本merge的深度不完全一样，从而造成即使文档集完全一样，产生的segment file却不完全一样。
为了解决这个问题，ES1.6版本以后加入了synced flush的新特性。 对于5分钟没有更新过的shard，会自动synced flush一下，实质是为对应的shard加了一个synced flush ID。这样当重启结点的时候，先对比一下shard的synced flush ID，就可以知道两个shard是否完全相同，避免了不必要的segment file拷贝，极大加快了冷索引的恢复速度。
需要注意的是synced flush只对冷索引有效，对于热索引（5分钟内有更新的索引）没有作用。 如果重启的结点包含有热索引，那么还是免不了大量的文件拷贝。因此在重启一个结点之前，最好按照以下步骤执行，recovery几乎可以瞬间完成:
- 暂停数据写入程序
- 关闭集群shard allocation
- 手动执行POST /_flush/synced
- 重启结点
- 重新开启集群shard allocation
- 等待recovery完成，集群health status变成green
- 重新开启数据写入程序
(特别大的)热索引为何恢复慢
对于冷索引，由于数据不再更新，利用synced flush特性，可以快速直接从本地恢复数据。 而对于热索引，特别是shard很大的热索引，除了synced flush派不上用场需要大量跨结点拷贝segment file以外，translog recovery是导致慢的更重要的原因。
从主片恢复数据到副片需要经历3个阶段:
- 对主片上的segment file做一个快照，然后拷贝到复制片分配到的结点。数据拷贝期间，不会阻塞索引请求，新增索引操作记录到translog里。
- 对translog做一个快照，此快照包含第一阶段新增的索引请求，然后重放快照里的索引操作。此阶段仍然不阻塞索引请求，新增索引操作记录到translog里。
- 为了能达到主副片完全同步，阻塞掉新索引请求，然后重放阶段二新增的translog操作。
可见，在recovery完成之前，translog是不能够被清除掉的（禁用掉正常运作期间后台的flush操作）。如果shard比较大，第一阶段耗时很长，会导致此阶段产生的translog很大。重放translog比起简单的文件拷贝耗时要长得多，因此第二阶段的translog耗时也会显著增加。等到第三阶段，需要重放的translog可能会比第二阶段还要多。 而第三阶段是会阻塞新索引写入的，在对写入实时性要求很高的场合，就会非常影响用户体验。 因此，要加快大的热索引恢复速度，最好的方式是遵从上一节提到的方法: 暂停新数据写入，手动sync flush，等待数据恢复完成后，重新开启数据写入，这样可以将数据延迟影响可以降到最低。
万一遇到Recovery慢，想知道进度怎么办呢？ CAT Recovery API可以显示详细的recovery各个阶段的状态。 这个API怎么用就不在这里赘述了，参考: CAT Recovery
其他Recovery相关的专家级设置
还有其他一些专家级的设置（参见： recovery）可以影响recovery的速度，但提升速度的代价是更多的资源消耗，因此在生产集群上调整这些参数需要结合实际情况谨慎调整，一旦影响应用要立即调整回来。 对于搜索并发量要求高，延迟要求低的场合，默认设置一般就不要去动了。 对于日志实时分析类对于搜索延迟要求不高，但对于数据写入延迟期望比较低的场合，可以适当调大indices.recovery.max_bytes_per_sec，提升recovery速度，减少数据写入被阻塞的时长。
最后要说的一点是ES的版本迭代很快，对于Recovery的机制也在不断的优化中。 其中有一些版本甚至引入了一些bug，比如在ES1.4.x有严重的translog recovery bug，导致大的索引trans log recovery几乎无法完成 （issue #9226） 。因此实际使用中如果遇到问题，最好在Github的issue list里搜索一下，看是否使用的版本有其他人反映同样的问题。
elasticsearch-rtf更新至2.1.1
Elasticsearch • medcl 发表了文章 • 4 个评论 • 8242 次浏览 • 2015-12-25 16:32
地址：https://github.com/medcl/elasticsearch-rtf
使用git快速签出最新版：
git clone git://github.com/medcl/elasticsearch-rtf.git -b master --depth 1
包含插件：
elasticsearch-analysis-ik-1.6.2 elasticsearch-analysis-pinyin-1.5.2
elasticsearch-analysis-mmseg-1.6.2 elasticsearch-analysis-stconvert-1.6.1
使用：
cd elasticsearch/bin
./elasticsearch
使用git快速签出最新版：
git clone git://github.com/medcl/elasticsearch-rtf.git -b master --depth 1
包含插件：
elasticsearch-analysis-ik-1.6.2 elasticsearch-analysis-pinyin-1.5.2
elasticsearch-analysis-mmseg-1.6.2 elasticsearch-analysis-stconvert-1.6.1
使用：
cd elasticsearch/bin
./elasticsearch
Day22：pipeline aggregation计算日留存率示例
Advent • 三斗室 发表了文章 • 1 个评论 • 12458 次浏览 • 2015-12-25 11:06
网友们多次讨论如何利用 ES 计算用户留存率的问题。这是个比较尴尬的情况，如果多次请求再自己做一下运算，问题很简单。但如果想要一次请求得到最终结果，在没有完整 JOIN 支持的 ES 里又显得比较难以完成。
目前我想到的比较容易达成的做法，是我们在记录用户登录操作日志的时候，把该用户的注册时间也同期输出。也就是说，这个索引的 mapping 是下面这样：
显然以这短短 3 行示例数据，我们口算都知道单日留存率是 50% 了。那么怎么通过一次 ES 请求也算出来呢？下面就要用到 ES 2.0 新增加的 pipeline aggregation 了。
最终这次请求的响应如下：
目前我想到的比较容易达成的做法，是我们在记录用户登录操作日志的时候，把该用户的注册时间也同期输出。也就是说，这个索引的 mapping 是下面这样：
curl -XPUT 'http://127.0.0.1:9200/login-2015.12.23/' -d '{
"settings" : {
"number_of_shards" : 1
},
"mappings" : {
"logs" : {
"properties" : {
"uid" : { "type" : "string", "index" : "not_analyzed" },
"register_time" : { "type" : "date", "index" : "not_analyzed" },
"login_time" : { "type" : "date", "index" : "not_analyzed" }
}
}
}
}'{"index":{"_index":"login-2015.12.23","_type":"logs"}}
{"uid":"1","register_time":"2015-12-23T12:00:00Z","login_time":"2015-12-23T12:00:00Z"}
{"index":{"_index":"login-2015.12.23","_type":"logs"}}
{"uid":"2","register_time":"2015-12-23T12:00:00Z","login_time":"2015-12-23T12:00:00Z"}
{"index":{"_index":"login-2015.12.24","_type":"logs"}}
{"uid":"1","register_time":"2015-12-23T12:00:00Z","login_time":"2015-12-24T12:00:00Z"}显然以这短短 3 行示例数据，我们口算都知道单日留存率是 50% 了。那么怎么通过一次 ES 请求也算出来呢？下面就要用到 ES 2.0 新增加的 pipeline aggregation 了。
curl -XPOST 'http://127.0.0.1:9200/login-2015.12.23,login-2015.12.24/_search' -d'
{
"size" : 0,
"aggs" : {
"new_users" : {
"filters" : {
"filters" : [
{
"range" : {
"register_time" : {
"gte" : "2015-12-23",
"lt" : "2015-12-24"
}
}
}
]
},
"aggs" : {
"register_count" : {
"cardinality" : {
"field" : "uid"
}
},
"today" : {
"filter" : {
"range" : {
"login_time" : {
"gte" : "2015-12-24",
"lt" : "2015-12-25"
}
}
},
"aggs" : {
"login_count" : {
"cardinality" : {
"field" : "uid"
}
}
}
},
"retention" : {
"bucket_script" : {
"buckets_path" : {
"today_count" : "today>login_count",
"yesterday_count" : "register_count"
},
"script" : {
"lang" : "expression",
"inline" : "today_count / yesterday_count"
}
}
}
}
}
}
}'- pipeline agg 的 parent agg 必须是返回数组的 buckets agg 类型。我这里曾经打算使用 filter agg 直接请求register_time:["now-2d" TO "now-1d"]，结果报错说找不到 buckets_path 的 START_OBJECT。所以改用了 filters agg 的数组格式。
- bucket_script agg 同样受 scripting module 的影响。也就是说，官网示例里的"script":"today_count / yesterday_count" 这种写法，是采用了 groovy 引擎的 inline 模式。在 ES 2.0 的默认设置下，是被禁止运行的！所以，应该按照 scripting module 的统一要求，改写成 file 形式存放到 config/scripts下；或者改用 Lucene Expression 运行。考虑到 pipeline aggregation 只支持数值运算，这里使用 groovy 价值不大，所以直接指明 lang 参数即可。
最终这次请求的响应如下：
{
"took" : 3,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"failed" : 0
},
"hits" : {
"total" : 3,
"max_score" : 0.0,
"hits" : [ ]
},
"aggregations" : {
"new_users" : {
"buckets" : [ {
"doc_count" : 3,
"today" : {
"doc_count" : 1,
"login_count" : {
"value" : 1
}
},
"register_count" : {
"value" : 2
},
"retention" : {
"value" : 0.5
}
} ]
}
}
}
