您可以为此使用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"
}
}
}
}
}
}
}
更新
总结一下:
- Watcher 是一个商业产品
- ElastAlert(尚)不支持它,需要一些努力才能使其工作
还有另一种更简单、更便宜的方法可以使用 Logstash 实现这一目标。即使elasticsearch
输入插件不支持聚合,也可以使用http_poller
输入插件定期向 Elasticsearch 发送聚合查询。然后使用过滤器,您可以检查是否达到了所需的阈值,最后,如果是这种情况,使用email
输出插件通过电子邮件提醒某人。
source=...
配置基本上是这样的(请注意,您上面的聚合查询需要进行 URL 编码并使用参数发送到 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
也可以利用另一个名为elasticwatch的node.js 工具来执行此操作。