一、安裝

參考：http://www.rzrgm.cn/cpy-devops/p/9287531.html

首先保證系統安裝了java 1.8以上版本

tar –zxvf /home/logstash-7.7.0.tar.gz -C /usr/local

進入/usr/local/logstash-7.7.0/bin目錄下

二、配置文件

創建文件 convert_csv.conf

input{
      elasticsearch {
         hosts => ["127.0.0.1:9200"]    #要導出來的es服務器的地址
         index => "es的數據index值"      
     }
}
output{
    file {

     filed => [””,””,””]    #filed字段選擇，如下圖展示_source中的信息，帶有下劃線的字段不需要選擇
#path為指定文件路徑
      path => "/home/csv.csv"
    }

}

三、執行

./logstash -f convert_csv.conf

同步兩套ES數據

只需要將上述output修改對應的es的output即可

output {

    elasticsearch {
        #全文檢索服務
        hosts => "localhost:9200"
        #索引(數據庫)
        index => "zl_dev"
        #類型(數據庫表)
        document_type => "%{type}"
        #主鍵(防止重復)
        document_id => "%{id}"
    }
}

實時同步es數據

官方幫助文檔：https://www.elastic.co/guide/en/logstash/current/plugins-inputs-elasticsearch.html

需要插件 Elasticsearch input plugin

input { # Read all documents from Elasticsearch matching the given query elasticsearch

{ hosts => "localhost"

  query => '{ "query": { "match": { "statuscode": 200 } }, "sort": [ "_doc" ] }' } }

這里query的學習，請參考https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html

我參考了下 http://www.rzrgm.cn/-fengmu/p/13074152.html

可以按照時間 每天進行數據采集

{
    "query": {
        "match": {
            "fileddate":"2020-06-22"   #根據fileddate字段，取得日期2020-06-22數據
        }
    }
}

計劃增加 輸入讀取選項 schedule index 

input { # Read all documents from Elasticsearch matching the given query elasticsearch { hosts => "localhost" index => "按照時間索引傳入" query => '{ "query": { "match": { "statuscode": 200 } }, "sort": [ "_doc" ] }' schedule => "* * 1 * *" #每天凌晨1點執行 #任務調度 (分、時、天、月、年，全部為*默認含義為每分鐘都更新) } }

輸出與上述輸出到es一致即可