如何設置過濾器在聚集值達到特定閾值時返回？

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也被利用來做到這一點。