如何设置过滤器在聚集值达到特定阈值时返回？

Question

以下面的聚集查询为例：

{ 
    "query": { 
    "match_all": {} 
    }, 
    "aggs": { 
    "groupBy": { 
     "terms": { 
     "field": "CustomerName" 
     }, 
     "aggs": { 
     "points_sum": { 
      "stats": { 
      "field": "TransactionAmount" 
      } 
     } 
     } 
    } 
    }, 
    "size": 0 
} 

我只要我想知道当任何客户名称的平均TransactionAmount（stats.avg）是上述所有客户的采购，一些门槛我编制了一份文件，使得我的平均值高于该阈值。似乎渗滤器是为了匹配文档或规则而设计的，但我找不到任何使用渗滤器匹配基于聚合结果的规则的好例子。

这可能吗？过滤器是最好的解决方案吗？还有其他更好的解决方案吗？在此先感谢

Answer 1

可以使用Watcher商业产品为并定义如下表：

PUT _watcher/watch/transaction_alert 
{ 
    "trigger": { 
    "schedule": { 
     "interval": "1m" 
    } 
    }, 
    "input": { 
    "search": { 
     "request": { 
     "indices": "transactions", 
     "types": "transaction", 
     "body": { 
      "query": { 
      "match_all": {} 
      }, 
      "size": 0, 
      "aggs": { 
      "groupBy": { 
       "terms": { 
       "field": "CustomerName" 
       }, 
       "aggs": { 
       "points_sum": { 
        "stats": { 
        "field": "TransactionAmount" 
        } 
       } 
       } 
      } 
      } 
     } 
     } 
    } 
    }, 
    "condition": { 
    "script": { 
     "inline": "return ctx.payload.aggregations.groupBy.buckets.findAll{ cust -> cust.points_sum.avg >= 200}" 
    } 
    }, 
    "actions": { 
    "send_email": { 
     "email": { 
     "to": "<username>@<domainname>", 
     "subject": "Customer Notification - Transaction > 200", 
     "body": "The attached customers have a transaction average above $200" 
     "attachments" : { 
      "data.yml" : { 
       "data" : { 
       "format" : "yaml" 
       } 
      } 
     } 
     } 
    } 
    } 
} 

UPDATE

综上所述：

• Watcher是一个商业产品
• ElastAlert不支持它（尚）和requires some effort，使其工作

还有另一种更简单和更便宜的方式来实现这个使用Logstash。即使elasticsearch输入插件不支持聚合，也可以使用输入插件http_poller以定期向Elasticsearch发送聚合查询。然后使用过滤器，您可以检查是否达到了所需的阈值，最后，如果使用email输出插件，则通过电子邮件通知某人。

配置基本上是这样的（请注意，您的上述汇总查询需要使用URL编码并使用source=... parameter发送给ES）。另外请注意，我已经修改您的查询按照points_sum.avg（DESC）

input { 
    http_poller { 
    urls => { 
     test1 => 'http://localhost:9200/your-index/_search?source=%7B%22query%22%3A%7B%22match_all%22%3A%7B%7D%7D%2C%22aggs%22%3A%7B%22groupBy%22%3A%7B%22terms%22%3A%7B%22field%22%3A%22CustomerName%22%2C%22order%22%3A%7B%22points_sum.avg%22%3A%22desc%22%7D%7D%2C%22aggs%22%3A%7B%22points_sum%22%3A%7B%22stats%22%3A%7B%22field%22%3A%22TransactionAmount%22%7D%7D%7D%7D%7D%2C%22size%22%3A0%7D' 
    } 
    # checking every 10 seconds 
    interval => 10 
    codec => "json" 
    } 
} 
filter { 
    split { 
    field => "[aggregations][groupBy][buckets]" 
    } 
} 
output { 
    if [aggregations][groupBy][buckets][points_sum][avg] > 200 { 
    email { 
     to => "<username>@<domainname>" 
     subject => "Customer Notification - Transaction > 200", 
     body => "The customer %{[aggregations][groupBy][buckets][key]} has a transaction average above $200" 
    } 
    } 
} 

同意桶进行排序，这是一个非常简单的实现，但它应该是工作，你可以建立在它，使其更聪明与Logstash和你的想象力的极限是天空;-)

更新2

另一个node.js的工具调用elasticwatch也被利用来做到这一点。