elasticsearch中的MR

liangwenrong

2019-12-03

elastic search真是个让人既爱又恨的东西，性能强劲，功能强大，但就是在使用中遇到种种问题(多半因为文档太差)。
文章记录一下在es 2.2.0版本中使用Scripted Metric Aggregation（也就是牛X的map-reduce）的方法。
api这是官方文档，但是并不详细，看完并不能干出什么事来．这是java api

下面贴出完整的实践内容和代码(敏感内容已抹去)，目的是根据行为日志得出活跃分值

在elasticsearch.yml文件中添加配置启用groovy脚本

script.engine.groovy.file.aggs: true
script.engine.groovy.file.mapping: true
script.engine.groovy.file.search: true
script.engine.groovy.file.update: true
script.engine.groovy.file.plugin: true
script.engine.groovy.indexed.aggs: true
script.engine.groovy.indexed.mapping: false
script.engine.groovy.indexed.search: true
script.engine.groovy.indexed.update: false
script.engine.groovy.indexed.plugin: false
script.engine.groovy.inline.aggs: true
script.engine.groovy.inline.mapping: true
script.engine.groovy.inline.search: true
script.engine.groovy.inline.update: true
script.engine.groovy.inline.plugin: true

这是es 中保存的数据

curl -XPUT &#34;http://10.1.200.34:9200/behavior-2017.02/candidate/AVoG9St-6pLzqkumYcIr&#34; -d &#39;
{
 &#34;businessLine&#34;: &#34;platform&#34;,
 &#34;createTime&#34;: &#34;2017-02-04T02:30:14.000Z&#34;,
 &#34;latitude&#34;: 0,
 &#34;longitude&#34;: 0,
 &#34;name&#34;: &#34;c_login&#34;,
 &#34;network&#34;: &#34;unknown&#34;,
 &#34;ownerId&#34;: 6403128,
 &#34;ownerType&#34;: &#34;candidate&#34;,
 &#34;params&#34;: {
 &#34;positionId&#34;: &#34;2112620&#34;
 },
 &#34;uuid&#34;: &#34;a36556ed286348aeb970e0ba1cda1447&#34;
}&#39;
curl -XPUT &#34;http://10.1.200.34:9200/behavior-2017.02/candidate/AVoG9SgJP_y-H6mvM9g8&#34; -d &#39;
{
 &#34;businessLine&#34;: &#34;platform&#34;,
 &#34;clientIp&#34;: &#34;*3.1*8.113.*6&#34;,
 &#34;createTime&#34;: &#34;2017-02-04T02:30:13.683Z&#34;,
 &#34;latitude&#34;: 0,
 &#34;longitude&#34;: 0,
 &#34;name&#34;: &#34;c_login&#34;,
 &#34;network&#34;: &#34;unknown&#34;,
 &#34;ownerId&#34;: 6403118,
 &#34;ownerType&#34;: &#34;candidate&#34;,
 &#34;terminal&#34;: &#34;pc&#34;,
 &#34;userAgent&#34;: &#34;Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36&#34;,
 &#34;uuid&#34;: &#34;0d9b007e3d624180aefd57e7df0b656c&#34;,
}&#39;
curl -XPUT &#34;http://10.1.200.34:9200/behavior-2017.02/candidate/AVoG9SgJP_y-H6mvM9g1&#34; -d &#39;
{
 &#34;businessLine&#34;: &#34;platform&#34;,
 &#34;clientIp&#34;: &#34;*23.*8.*3.1*&#34;,
 &#34;createTime&#34;: &#34;2017-02-04T02:30:13.683Z&#34;,
 &#34;latitude&#34;: 0,
 &#34;longitude&#34;: 0,
 &#34;name&#34;: &#34;c_register&#34;,
 &#34;network&#34;: &#34;unknown&#34;,
 &#34;ownerId&#34;: 6403127,
 &#34;ownerType&#34;: &#34;candidate&#34;,
 &#34;terminal&#34;: &#34;pc&#34;,
 &#34;userAgent&#34;: &#34;Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36&#34;,
 &#34;uuid&#34;: &#34;0d9b007e3d624180aefd57e7df0b656c&#34;,
}&#39;

这是查询用的脚本，以id为分组，统计行为信息。

curl -XGET &#34;http://10.1.200.34:9200/behavior*/_search/?pretty&#34; -d &#39;
{
 &#34;aggs&#34;: {
 &#34;group_by_ownerId&#34; : { 
 &#34;terms&#34; : {
 &#34;field&#34; : &#34;ownerId&#34; 
 },
 &#34;aggs&#34;:{
 &#34;livenessScore&#34;: {
 &#34;scripted_metric&#34;: {
 &#34;init_script&#34; : {&#34;file&#34;: &#34;user-liveness-score-init&#34;},
 &#34;map_script&#34; : {&#34;file&#34;: &#34;user-liveness-score-map&#34;},
 &#34;combine_script&#34; : {&#34;file&#34;: &#34;user-liveness-score-combine&#34;},
 &#34;reduce_script&#34; : {&#34;file&#34;: &#34;user-liveness-score-reduce&#34;},
 &#34;params&#34;: {
 &#34;_agg&#34;: {&#34;resumeScore&#34;:{6403127:60}}
 }
 }
 }
 } 
 }
 },
 &#34;query&#34;: {
 &#34;filtered&#34;: {
 &#34;query&#34;: {
 &#34;match_all&#34;: {}
 }
 }
 },
 &#34;fields&#34;: [
 &#34;ownerId&#34;
 ]
}&#39;

脚本及脚本的存放位置

[root@jiqi001 scripts]# pwd
/apps/elasticsearch/config/scripts
[root@jiqi001 scripts]# ls
user-liveness-score-combine.groovy user-liveness-score-init.groovy user-liveness-score-map.groovy user-liveness-score-reduce.groovy

_agg.loginScoreInWeek=0;
_agg.loginScoreInMonth=0;
_agg.registerScoreInWeek=0;
_agg.registerScoreInMonth=0;
~ 
&#34;user-liveness-score-init.groovy&#34; 15L, 383C

xDaysBefore = Math.round((new Date().getTime() - doc.createTime) / 1000 / 60 / 60 / 24);
behaviorName = doc.name.value;
resumeScore = _agg.resumeScore.get(String.valueOf(doc.ownerId.value));
if (behaviorName.equals(&#34;c_login&#34;)) {
 if (xDaysBefore &lt;= 7) {
 if (_agg.loginScoreInWeek &lt; 4) {
 _agg.loginScoreInWeek += 2;
 }
 } else if (7 &lt; xDaysBefore &amp;&amp; xDaysBefore &lt;= 30) {
 if (_agg.loginScoreInMonth &lt; 2) {
 _agg.loginScoreInMonth += 1;
 }
 }
} else if (behaviorName.equals(&#34;c_register&#34;)) {
 if (resumeScore != null &amp;&amp; resumeScore &gt; 30) {
 if (xDaysBefore &lt;= 7) {
 if (_agg.registerScoreInWeek == 0) {
 _agg.registerScoreInWeek = 5;
 }
 } else if (7 &lt; xDaysBefore &amp;&amp; xDaysBefore &lt;= 30) {
 if (_agg.registerScoreInMonth == 0) {
 _agg.registerScoreInMonth = 3;
 }
 }
 }
};
~ 
&#34;user-liveness-score-map.groovy&#34; 66L, 2657C
```
```shell
_agg
~ 
&#34;user-liveness-score-combine.groovy&#34; 1L, 5C
```
````shell
double score = 0;
loginScoreInWeek=0;
loginScoreInMonth=0;
registerScoreInWeek=0;
registerScoreInMonth=0;
for (a in _aggs) {
 if(loginScoreInWeek&lt;4){
 loginScoreInWeek += a.get(&#34;loginScoreInWeek&#34;);
 };
 if(loginScoreInMonth&lt;2){
 loginScoreInMonth += a.get(&#34;loginScoreInMonth&#34;);
 };
 if(registerScoreInWeek&lt;5){
 registerScoreInWeek += a.get(&#34;registerScoreInWeek&#34;);
 };
 if(registerScoreInMonth&lt;3){
 registerScoreInMonth += a.get(&#34;registerScoreInMonth&#34;);
 };
 };
 if(loginScoreInWeek&gt;4){
 loginScoreInWeek =4;
 };
 if(loginScoreInMonth&gt;2){
 loginScoreInMonth =2;
 };
 if(registerScoreInWeek&gt;5){
 registerScoreInWeek =5;
 };
 if(registerScoreInMonth&gt;3){
 registerScoreInMonth =3;
 };
 score += loginScoreInWeek;
 score += loginScoreInMonth;
 score += registerScoreInWeek;
 score += registerScoreInMonth;
 return score;
&#34;user-liveness-score-reduce.groovy&#34;
```

- Java代码
```java
 private Map&lt;String, Double&gt; getUsersLiveScore(Map&lt;String, Object&gt; userAndResumeScore) throws InterruptedException, ExecutionException {
 Map&lt;String, Double&gt; userAndLivenessScore = new HashMap&lt;&gt;();
 Map&lt;String, Object&gt; param = new HashMap&lt;&gt;();
 param.put(&#34;resumeScore&#34;, userAndResumeScore);
 Map&lt;String, Object&gt; params = new HashMap&lt;&gt;();
 params.put(&#34;_agg&#34;, param);
 Client client = eSClient.getClient();
 AggregationBuilder aggregation = AggregationBuilders.terms(&#34;group_by_ownerId&#34;)
 .field(&#34;ownerId&#34;)
 .subAggregation(
 AggregationBuilders.scriptedMetric(&#34;livenessScore&#34;)
 .params(params)
 .initScript(new Script(&#34;user-liveness-score-init&#34;, ScriptService.ScriptType.FILE, &#34;groovy&#34;, null))
 .mapScript(new Script(&#34;user-liveness-score-map&#34;, ScriptService.ScriptType.FILE, &#34;groovy&#34;, null))
 .combineScript(new Script(&#34;user-liveness-score-combine&#34;, ScriptService.ScriptType.FILE, &#34;groovy&#34;, null))
 .reduceScript(new Script(&#34;user-liveness-score-reduce&#34;, ScriptService.ScriptType.FILE, &#34;groovy&#34;, null))
 );
 TermsQueryBuilder ownerId = QueryBuilders.termsQuery(&#34;ownerId&#34;, userAndResumeScore.keySet());
 BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

 boolQueryBuilder.must(ownerId);
 boolQueryBuilder.must(QueryBuilders.rangeQuery(&#34;createTime&#34;).gte(DateTime.now().plusDays(-30).toDate()));
 SearchResponse response = client.prepareSearch(&#34;behavior-*&#34;)
 .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
 .setQuery(boolQueryBuilder)
 .addAggregation(aggregation)
 .setFrom(0)
 .setSize(3000)
 .addField(&#34;ownerId&#34;)
 .execute()
 .get();
 for (Aggregation agg : response.getAggregations()) {
 List&lt;Terms.Bucket&gt; buckets = ((LongTerms) agg).getBuckets();
 for (Terms.Bucket bucket : buckets) {
 String userId = bucket.getKeyAsString();
 for (Aggregation agg2 : bucket.getAggregations()) {
 double score = (double) ((InternalScriptedMetric) agg2).aggregation();
 userAndLivenessScore.put(String.valueOf(userId), score);
 }
 }
 }
 return userAndLivenessScore;
 }
```
- Tips
1.参数param必须放在_agg变量里。
2.可以用&#34;combine_script&#34;:&#34;_agg;&#34;,&#34;reduce_script&#34;:&#34;_aggs;&#34;来调试脚本。
3.combine组合的结果会以分片为分组，并非整个查询结果的组合．比如查询一个index如果在５个分片上有结果则返回一个长度为５的数组,

elasticsearch中的MR