使用Logstash

指數ElasticSearch結果我有以下指標：使用Logstash

POST /cars/transactions/_bulk 
{ "index": {}} 
{ "price" : 10000, "color" : "red", "make" : "honda", "sold" : "2014-10-28" } 
{ "index": {}} 
{ "price" : 20000, "color" : "red", "make" : "honda", "sold" : "2014-11-05" } 
{ "index": {}} 
{ "price" : 30000, "color" : "green", "make" : "ford", "sold" : "2014-05-18" } 
{ "index": {}} 
{ "price" : 15000, "color" : "blue", "make" : "toyota", "sold" : "2014-07-02" } 
{ "index": {}} 
{ "price" : 12000, "color" : "green", "make" : "toyota", "sold" : "2014-08-19" } 
{ "index": {}} 
{ "price" : 20000, "color" : "red", "make" : "honda", "sold" : "2014-11-05" } 
{ "index": {}} 
{ "price" : 80000, "color" : "red", "make" : "bmw", "sold" : "2014-01-01" } 
{ "index": {}} 
{ "price" : 25000, "color" : "blue", "make" : "ford", "sold" : "2014-02-12" }

而且我執行以下搜索：

GET /cars/transactions/_search 
{ 
    "size" : 0, 
    "aggs" : { 
     "popular_colors" : { 
      "terms" : { 
       "field" : "color" 
      } 
     } 
    } 
}

的答覆中說，我收到如下：

{ 
    "took": 2, 
    "timed_out": false, 
    "_shards": { 
    "total": 5, 
    "successful": 5, 
    "failed": 0 
    }, 
    "hits": { 
    "total": 8, 
    "max_score": 0, 
    "hits": [] 
    }, 
    "aggregations": { 
    "popular_colors": { 
     "doc_count_error_upper_bound": 0, 
     "sum_other_doc_count": 0, 
     "buckets": [ 
     { 
      "key": "red", 
      "doc_count": 4 
     }, 
     { 
      "key": "blue", 
      "doc_count": 2 
     }, 
     { 
      "key": "green", 
      "doc_count": 2 
     } 
     ] 
    } 
    } 
}

我的問題是，我該如何將該文檔重新索引到不同的索引中？

我想：

input { 
    elasticsearch { 
    hosts => "localhost" 
    index => "cars" 
    query => '{ 
    "size" : 0, 
    "aggs" : { 
     "popular_colors" : { 
      "terms" : { 
       "field" : "color" 
      } 
     } 
    } 
}' 
    size => 500 
    scroll => "5m" 
    docinfo => true 
    } 
}

但它不工作，因爲插件的SEARCH_TYPE是掃描和它不支持聚合。

我也試過：

input { 
file { 
    path => "C:\ELK-STACK\logstash-2.3.4\bin\out.json" 
    start_position => "beginning" 
    codec => json_lines } 
    }

凡out.json的內容是：

{"took":1,"timed_out":false,"_shards":{"total":5,"successful":5,"failed":0},"hits":{"total":8,"max_score":1.0,"hits":[{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7l","_score":1.0,"_source":{ "price" : 10000, "color" : "red", "make" : "honda", "sold" : "2014-10-28" }},{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7m","_score":1.0,"_source":{ "price" : 20000, "color" : "red", "make" : "honda", "sold" : "2014-11-05" }},{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7p","_score":1.0,"_source":{ "price" : 12000, "color" : "green", "make" : "toyota", "sold" : "2014-08-19" }},{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7o","_score":1.0,"_source":{ "price" : 15000, "color" : "blue", "make" : "toyota", "sold" : "2014-07-02" }},{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7n","_score":1.0,"_source":{ "price" : 30000, "color" : "green", "make" : "ford", "sold" : "2014-05-18" }},{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7q","_score":1.0,"_source":{ "price" : 20000, "color" : "red", "make" : "honda", "sold" : "2014-11-05" }},{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7r","_score":1.0,"_source":{ "price" : 80000, "color" : "red", "make" : "bmw", "sold" : "2014-01-01" }},{"_index":"cars","_type":"transactions","_id":"AVexGB7_99OIq3MORm7s","_score":1.0,"_source":{ "price" : 25000, "color" : "blue", "make" : "ford", "sold" : "2014-02-12" }}]}}

但之後

設置沒有產生任何輸出：默認管道工：8

管道主開始

我認爲這是因爲JSON文件沒有爲JSON插件準備，而我需要做一些準備工作（如使用Java API），但我想避免，如果可能的。

謝謝！

來源

2016-10-11 Yeikel

如您所知，elasticsearch輸入插件不支持聚合。可以使用http_poller輸入插件來定期（或每天一次）向Elasticsearch發送聚合查詢。然後使用elasticsearch輸出，可以將生成的聚合再次發送給ES。

配置基本上是這樣的（注意聚合查詢需要使用URL編碼並使用source=... parameter發送給ES）。

input { 
    http_poller { 
    urls => { 
     test1 => 'http://localhost:9200/cars/transactions/_search?source=%7B%22size%22%3A0%2C%22aggs%22%3A%7B%22popular_colors%22%3A%7B%22terms%22%3A%7B%22field%22%3A%22color%22%7D%7D%7D%7D' 
    } 
    # checking once per day 
    interval => 86400 
    codec => "json" 
    } 
} 
filter { 
} 
output { 
    elasticsearch { 
    hosts => ["localhost:9200"] 
    index => "my_aggs" 
    } 
}

來源

2016-10-11 04:04:56 Val

回答

相關問題