ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > IT基础架构 > > ES中的MR

ES中的MR

作者：飘絮絮絮丶 时间：2022-01-05 15:55:05 0 删除 编辑

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=;
_agg.loginScoreInMonth=;
_agg.registerScoreInWeek=;
_agg.registerScoreInMonth=;
~                      
"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 == ) {
                _agg.registerScoreInWeek = 5;
            }
        } else if (7 < xDaysBefore && xDaysBefore <= 30) {
            if (_agg.registerScoreInMonth == ) {
                _agg.registerScoreInMonth = 3;
            }
        }
    }
};
~                                                                                                                                                                                                                                       
"user-liveness-score-map.groovy" 66L, 2657C
```
```shell
_agg
~                                                                                                             
"user-liveness-score-combine.groovy" 1L, 5C
```
````shell
double score = ;
loginScoreInWeek=;
loginScoreInMonth=;
registerScoreInWeek=;
registerScoreInMonth=;
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()
                .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;
    }
```


作者：chenmingang
链接：https://www.jianshu.com/p/9683e7e0ac44
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/69946831/viewspace-3163972/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： 雷海林_TDSQL异构数据同步与迁移【PPT】
下一篇： 24.3 INFORMATION_SCHEMA InnoDB 表
请登录后发表评论 登录
全部评论
飘絮絮絮丶

注册时间：2019-09-09

  • 博文量
    2646
  • 访问量
    42275

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593