elasticsearch中的MR

elastic search真是個讓人既愛又恨的東西，性能強勁，功能強大，但就是在使用中遇到種種問題(多半因為文檔太差)。

下面貼出完整的實踐內容和代碼(敏感內容已抹去)，目的是根據行為日誌得出活躍分值

• 在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/>/apps/elasticsearch/config/scripts
[root@jiqi001/>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> getUsersLiveScore(Map<string> userAndResumeScore) throws InterruptedException, ExecutionException {
 Map<string> userAndLivenessScore = new HashMap<>();
 Map<string> param = new HashMap<>();
 param.put("resumeScore", userAndResumeScore);
 Map<string> 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如果在５個分片上有結果則返回一個長度為５的數組,
/<terms.bucket>/<string>/<string>/<string>/<string>/<string>

閱讀更多 戚小柒說IT 的文章

關鍵字: ElasticSearch MapReduce