Logstash的安裝

Logstash它是ES下的一款開源軟體，它能夠同時從多個來源采集資料、轉換資料，然後将資料發送到Eleasticsearch中建立索引。

我們在項目中通常使用Logstash将資料庫（如MySQL）中的資料采用到ES索引中。也就是Logstash的工作是從MySQL中讀取資料，向ES中建立索引，這裡需要提前建立mapping的模闆檔案以便logstash使用。

1、下載下傳Logstash

解壓：

這是解壓後的目錄結構。

2、安裝logstash-input-jdbc

為什麼要安裝logstash-input-jdbc呢?

因為版本不同，自身所帶的插件也不同，本文所說的是6.x的版本。而6.x版本本身不帶logstash-input-jdbc插件，需要手動安裝。Logstash5.x以上版本本身自帶有logstash-input-jdbc。

ogstash-input-jdbc 是ruby開發的，先下載下傳ruby并安裝

下載下傳位址: https://rubyinstaller.org/downloads/

下載下傳2.5版本即可。

安裝完成檢視是否安裝成功

安裝成功後我們可以在logstash根目錄下的以下目錄檢視對應的插件版本：

如果想省事的話，可以下載下傳解壓本文提供的logstash-6.2.1.zip,此logstash中已內建了logstash-input-jdbc插件。

3、配置mysql.conf

在logstash的config目錄下配置mysql.conf檔案供logstash使用，logstash會根據mysql.conf檔案的配置的位址從MySQL中讀取資料向ES中寫入索引。（本文講的資料庫是MySQL）

參考 https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html

配置輸入資料源和輸出資料源。

input {
  stdin {
  }
  jdbc {
  jdbc_connection_string => "jdbc:mysql://localhost:3306/jy_brand?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC"
  # the user we wish to excute our statement as
  jdbc_user => "root"
  jdbc_password => 123456
  # the path to our downloaded jdbc driver  
  jdbc_driver_library => "F:/develop/maven/repository3/mysql/mysql-connector-java/5.1.41/mysql-connector-java-5.1.41.jar"
  # the name of the driver class for mysql
  jdbc_driver_class => "com.mysql.jdbc.Driver"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "50000"
  #要執行的sql檔案
  #statement_filepath => "/conf/course.sql"
  statement => "select * from course_pub where timestamp > date_add(:sql_last_value,INTERVAL 8 HOUR)"
  #定時配置
  schedule => "* * * * *"
  record_last_run => true
  last_run_metadata_path => "D:/ElasticSearch/logstash-6.2.1/config/logstash_metadata"
  }
}
​
​
output {
  elasticsearch {
  #ES的ip位址和端口
  hosts => "localhost:6200"
  #hosts => ["localhost:6200","localhost:6202","localhost:6203"]
  #ES索引庫名稱
  index => "jy_brand"
  document_id => "%{id}"
  document_type => "doc"
  template =>"D:/ElasticSearch/logstash-6.2.1/config/jy_brand_template.json"
  template_name =>"jy_brand"
  template_overwrite =>"true"
  }
  stdout {
 #日志輸出
  codec => json_lines
  }
}
           

說明：

1、ES采用UTC時區問題

ES采用UTC 時區，比中原標準時間早8小時，是以ES讀取資料時讓最後更新時間加8小時

where timestamp > date_add(:sql_last_value,INTERVAL 8 HOUR)

2、logstash每個執行完成會在D:/ElasticSearch/logstash-6.2.1/config/logstash_metadata記錄執行時間下次以此時間為基準進行增量同步資料到索引庫。

如果需要轉載，請注明出處，謝謝！本文為部落客原創文章，部落格位址：https://blog.csdn.net/weixin_44299027