Elasticsearch、Kibana 數據導出實戰

摘要：問題2： elasticsearch數據導出。而數據的導出，一部分是業務場景需要，如：業務系統中支持檢索結果導出爲CSV、Json格式等。

1、問題引出

以下兩個導出問題來自Elastic中文社區。

問題1、kibana怎麼導出查詢數據？

問題2： elasticsearch數據導出

就像數據庫數據導出一樣，elasticsearch可以麼?

或者找到它磁盤上存放數據的位置，拷貝出來，放到另一個es服務器上或者轉成自己要的數據格式？

實際業務實戰中，大家或多或少的都會遇到導入、導出問題。

根據數據源的不同，基本可以藉助：

1、程序寫入
2、數據同步

logstash/flume/cana/es_hadoopl等來實現關係型數據庫（如：Oracle、mysql）、非關係型數據庫（如：Mongo、Redis）、大數據（Hadoop、Spark、Hive）到Elasticsearch的寫入。

而數據的導出，一部分是業務場景需要，如：業務系統中支持檢索結果導出爲CSV、Json格式等。

還有一部分是分析數據的需求：期望藉助Kibana工具將儀表盤聚合結果導出、不需要藉助程序儘快將滿足給定條件的結果數據導出等。

這些快速導出的需求，最好藉助插件或者第三方工具實現。

本文將重點介紹Kibana/Elasticsearch高效導出的插件、工具集。

2、期望導出數據格式

一般期望導出：CSV、Json格式。

3、Kibana導出工具

3.1 Kibana 官方導出

步驟1：點擊Kibana；

步驟2：左側選擇數據，篩選字段；

步驟3：右側點擊： share->csv reports。

步驟4：菜單欄：

選擇Management->Reporting->下載。

以上是kibana6.5.4的實操截圖。

其他常見報表數據導出：

在Dashboard的右上角點擊Inspect，再點擊就可以導出對應可視化報表對應的數據。

3.2 數據透視表pivot-kibana

Kibana的數據透視表——使用Kibana UI中的任何其他工具一樣使用數據透視表可以極大地簡化數據工作。

Flexmonster Pivot可以交叉和快速地彙總業務數據並以交叉表格式顯示結果。

地址： https://github.com/flexmonster/pivot-kibana/

篩選數據效果如下：

注意：建議7.X以上版本使用。低版本不支持。

4、Elasticsearch導出工具

4.1 es2csv

1、簡介：用Python編寫的命令行實用程序，用於以Lucene查詢語法或查詢DSL語法查詢Elasticsearch，並將結果作爲文檔導出到CSV文件中。

es2csv 可以查詢多個索引中的批量文檔，並且只獲取選定的字段，這可以縮短查詢執行時間。

2、地址： https://pypi.org/project/es2csv/

3、使用方法：

1 es2csv -u 192.168.1.1:9200 -q '{ "_source" :{ "excludes" :[ "*gxn" ,, "*kex" , "vperxs" , "lpix" ]}, "query" :{ "term" :{ "this_topic" :{ "value" :41}}}}' -r -i sogou_topic -o ~/ export .csv

4、使用效果：

官方最新更新支持5.X版本，實際驗證6.X版本也可以使用，導出效率高。

5、推薦指數：

五星，

Elasticsearch導出CSV首選方案。

4.2 elasticsearch-dump

1、簡介： Elasticsearch導入導出工具。

支持操作包含但不限於：

1）、數據導出

導出索引、檢索結果、別名或模板爲Json
導出索引爲gzip
支持導出大文件切割爲小文件
支持統一集羣不同索引間或者跨索引數據拷貝

2）、數據導入

支持Json數據、S3數據導入Elasticsearch。

2、地址：

https://github.com/taskrabbit/elasticsearch-dump

3、使用方法：

1 elasticdump \

2 --input=http://production.es.com:9200/my_index \

3 --output=query.json \

4 --searchBody='{ "query" :{ "term" :{ "username" : "admin" }}}'

如上，將檢索結果導出爲json文件。

更多導入、導出詳見github介紹。

4、使用效果：

早期1.X版本沒有reindex操作，使用elasticdump解決跨集羣數據備份功能。效果可以。

5、推薦指數：

五星。

Elasticsearch導出json首選方案。

4.3 logstash_output_csv

步驟1：安裝logstash_output_csv工具：

1 D :\l ogstash-6.5.4 \ bin > logstash-plugin.bat install logstash-output-csv

2 Validating logstash-output-csv

3 Installing logstash-output-csv

4 Installation successful

步驟2：配置conf文件

核心的：輸入input，輸出ouput，中間處理filter都在如下的配置文件中。

輸入：指定ES地址，索引，請求query語句；
輸出：csv輸出地址，輸出字段列表。

1 input {

2 elasticsearch {

3 hosts => "127.0.0.1:9200"

4 index => "company_infos"

5 query => '

6 {

7 "query": {

8 "match_all": {}

9 }

10 }

12 }

13 }

15 output {

16 csv {

17 # elastic field name

18 fields => [ "no" , "name" , "age" , "company_name" , "department" , "sex" ]

19 # This is path where we store output.

20 path => "D:\logstash-6.5.4\export\csv-export.csv"

21 }

22 }

步驟3：執行導出

1 D:\\logstash-6.5.4\bin>logstash -f ../config/logstash _ouput_ csv.conf

2 Sending Logstash logs to D:/2.es_install/logstash-6.5.4/logs which is now configured via log4j2.properties

3 [ 2019-08-03T23:45:00,914 ][ WARN ][ logstash.config.source.multilocal ] Ignoring the 'pipelines.yml' file because modules or command line options are specified

4 [ 2019-08-03T23:45:00,934 ][ INFO ][ logstash.runner ] Starting Logstash {"logstash.version"=>"6.5.4"}

5 [ 2019-08-03T23:45:03,473 ][ INFO ][ logstash.pipeline ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>8, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}

6 [ 2019-08-03T23:45:04,241 ][ INFO ][ logstash.pipeline ] Pipeline started successfully {:pipeline_id=>"main", :thread=>"# < Thread:0x34b305d3 sleep > "}

7 [ 2019-08-03T23:45:04,307 ][ INFO ][ logstash.agent ] Pipelines running {:count=>1, :running _pipelines=>[:main], :non_ running_pipelines=>[]}

8 [ 2019-08-03T23:45:04,740 ][ INFO ][ logstash.agent ] Successfully started Logstash API endpoint {:port=>9600}

9 [ 2019-08-03T23:45:05,610 ][ INFO ][ logstash.outputs.csv ] Opening file {:path=>"D:/logstash-6.5.4/export/csv-export.csv"}

10 [ 2019-08-03T23:45:07,558 ][ INFO ][ logstash.pipeline ] Pipeline has terminated {:pipeline_id=>"main", :thread=>"# < Thread:0x34b305d3 run > "}