ES中的MR
2017-02-09 本文已影响349人
chenmingang
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 "http://10.1.200.34:9200/behavior-2017.02/candidate/AVoG9St-6pLzqkumYcIr" -d '
{
"businessLine": "platform",
"createTime": "2017-02-04T02:30:14.000Z",
"latitude": 0,
"longitude": 0,
"name": "c_login",
"network": "unknown",
"ownerId": 6403128,
"ownerType": "candidate",
"params": {
"positionId": "2112620"
},
"uuid": "a36556ed286348aeb970e0ba1cda1447"
}'
curl -XPUT "http://10.1.200.34:9200/behavior-2017.02/candidate/AVoG9SgJP_y-H6mvM9g8" -d '
{
"businessLine": "platform",
"clientIp": "*3.1*8.113.*6",
"createTime": "2017-02-04T02:30:13.683Z",
"latitude": 0,
"longitude": 0,
"name": "c_login",
"network": "unknown",
"ownerId": 6403118,
"ownerType": "candidate",
"terminal": "pc",
"userAgent": "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36",
"uuid": "0d9b007e3d624180aefd57e7df0b656c",
}'
curl -XPUT "http://10.1.200.34:9200/behavior-2017.02/candidate/AVoG9SgJP_y-H6mvM9g1" -d '
{
"businessLine": "platform",
"clientIp": "*23.*8.*3.1*",
"createTime": "2017-02-04T02:30:13.683Z",
"latitude": 0,
"longitude": 0,
"name": "c_register",
"network": "unknown",
"ownerId": 6403127,
"ownerType": "candidate",
"terminal": "pc",
"userAgent": "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36",
"uuid": "0d9b007e3d624180aefd57e7df0b656c",
}'
- 这是查询用的脚本，以id为分组，统计行为信息。
curl -XGET "http://10.1.200.34:9200/behavior*/_search/?pretty" -d '
{
"aggs": {
"group_by_ownerId" : {
"terms" : {
"field" : "ownerId"
},
"aggs":{
"livenessScore": {
"scripted_metric": {
"init_script" : {"file": "user-liveness-score-init"},
"map_script" : {"file": "user-liveness-score-map"},
"combine_script" : {"file": "user-liveness-score-combine"},
"reduce_script" : {"file": "user-liveness-score-reduce"},
"params": {
"_agg": {"resumeScore":{6403127:60}}
}
}
}
}
}
},
"query": {
"filtered": {
"query": {
"match_all": {}
}
}
},
"fields": [
"ownerId"
]
}'
- 脚本及脚本的存放位置
[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;
~
"user-liveness-score-init.groovy" 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("c_login")) {
if (xDaysBefore <= 7) {
if (_agg.loginScoreInWeek < 4) {
_agg.loginScoreInWeek += 2;
}
} else if (7 < xDaysBefore && xDaysBefore <= 30) {
if (_agg.loginScoreInMonth < 2) {
_agg.loginScoreInMonth += 1;
}
}
} else if (behaviorName.equals("c_register")) {
if (resumeScore != null && resumeScore > 30) {
if (xDaysBefore <= 7) {
if (_agg.registerScoreInWeek == 0) {
_agg.registerScoreInWeek = 5;
}
} else if (7 < xDaysBefore && xDaysBefore <= 30) {
if (_agg.registerScoreInMonth == 0) {
_agg.registerScoreInMonth = 3;
}
}
}
};
~
"user-liveness-score-map.groovy" 66L, 2657C
```
```shell
_agg
~
"user-liveness-score-combine.groovy" 1L, 5C
```
````shell
double score = 0;
loginScoreInWeek=0;
loginScoreInMonth=0;
registerScoreInWeek=0;
registerScoreInMonth=0;
for (a in _aggs) {
if(loginScoreInWeek<4){
loginScoreInWeek += a.get("loginScoreInWeek");
};
if(loginScoreInMonth<2){
loginScoreInMonth += a.get("loginScoreInMonth");
};
if(registerScoreInWeek<5){
registerScoreInWeek += a.get("registerScoreInWeek");
};
if(registerScoreInMonth<3){
registerScoreInMonth += a.get("registerScoreInMonth");
};
};
if(loginScoreInWeek>4){
loginScoreInWeek =4;
};
if(loginScoreInMonth>2){
loginScoreInMonth =2;
};
if(registerScoreInWeek>5){
registerScoreInWeek =5;
};
if(registerScoreInMonth>3){
registerScoreInMonth =3;
};
score += loginScoreInWeek;
score += loginScoreInMonth;
score += registerScoreInWeek;
score += registerScoreInMonth;
return score;
"user-liveness-score-reduce.groovy"
```
- Java代码
```java
private Map<String, Double> getUsersLiveScore(Map<String, Object> userAndResumeScore) throws InterruptedException, ExecutionException {
Map<String, Double> userAndLivenessScore = new HashMap<>();
Map<String, Object> param = new HashMap<>();
param.put("resumeScore", userAndResumeScore);
Map<String, Object> params = new HashMap<>();
params.put("_agg", param);
Client client = eSClient.getClient();
AggregationBuilder aggregation = AggregationBuilders.terms("group_by_ownerId")
.field("ownerId")
.subAggregation(
AggregationBuilders.scriptedMetric("livenessScore")
.params(params)
.initScript(new Script("user-liveness-score-init", ScriptService.ScriptType.FILE, "groovy", null))
.mapScript(new Script("user-liveness-score-map", ScriptService.ScriptType.FILE, "groovy", null))
.combineScript(new Script("user-liveness-score-combine", ScriptService.ScriptType.FILE, "groovy", null))
.reduceScript(new Script("user-liveness-score-reduce", ScriptService.ScriptType.FILE, "groovy", null))
);
TermsQueryBuilder ownerId = QueryBuilders.termsQuery("ownerId", userAndResumeScore.keySet());
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must(ownerId);
boolQueryBuilder.must(QueryBuilders.rangeQuery("createTime").gte(DateTime.now().plusDays(-30).toDate()));
SearchResponse response = client.prepareSearch("behavior-*")
.setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
.setQuery(boolQueryBuilder)
.addAggregation(aggregation)
.setFrom(0)
.setSize(3000)
.addField("ownerId")
.execute()
.get();
for (Aggregation agg : response.getAggregations()) {
List<Terms.Bucket> 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.可以用"combine_script":"_agg;","reduce_script":"_aggs;"来调试脚本。
3.combine组合的结果会以分片为分组，并非整个查询结果的组合．比如查询一个index如果在５个分片上有结果则返回一个长度为５的数组,
