logstash增量同步mysql數據到es

本篇本章地址:https://www.cnblogs.com/Thehorse/p/11601013.html

今天我們來講一下logstash同步mysql數據到es

 

我認為呢，logstash是眾多同步mysql數據到es的插件中，最穩定並且最容易配置的一個。

 

input {
stdin {
}

jdbc {
type => "xxx"
jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/yinhelaowu"
jdbc_user => "root"
jdbc_password => "root"
jdbc_driver_library => "mysql1/mysql-connector-java-5.1.17-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => true
jdbc_page_size => "500"
record_last_run => true
use_column_value => true
tracking_column => "id"
last_run_metadata_path => "mysql1/jdbc1"
clean_run => "false"

statement_filepath => "mysql1/jdbc.sql"
schedule => "* * * * *"
clean_run => false
}

}

 

我來講一下這些屬性的作用，像type這個 主要是為了聲明這個jdbc{} 這個處理對象的，type可以定義任意值，

jdbc_connection_string ，這個是數據庫地址 還有用戶名密碼

jdbc_driver_library，logstash是需要使用jdbc來連接mysql的，並且使用logstash必須配置java環境，jdk1.8以上

jdbc_driver_class給本次連接起名 statement_filepath 這個是執行的sql語句 

 schedule 是表示多長時間同步一下數據 如果是******表示一分鍾同步一次

last_run_metadata_path這個會記錄你最后一次同步數據的一些值，比如你可以記錄創建時間，和自增的id，

type索引類型

clean_run 是否清楚執行過的數據

 

到了下邊

#這是輸出流

output {
#這個type是上面jdbc的type,有了這個if就可以進行批處理 if [type]=="TInquiryInfo"{ elasticsearch {
　　　　#這邊的端口一定是9200，java中使用的端口是9300這邊不要弄混了 hosts => ["localhost:9200"] # 索引名稱 index => "ca-inquiry" # type名稱 document_type => "TInquiryInfo" # 文檔id，inquiryId為sql文件中查詢出的字段名 document_id => "%{inquiryId}" } }  }


下面是執行的sql寫法

SELECT c.*,CONCAT(CONCAT(c.lat,','),c.lng) AS location,u.`praise`,u.`age`,u.`area_id` FROM zsf_carftsmanship c
LEFT JOIN zsf_user u ON c.user_id = u.`id` WHERE c.id > :sql_last_value

像elasticsearch這種大數據量查詢，多表關聯查詢無疑對性能有很大的沖擊，所以這邊我直接做了關聯查詢

這樣的話可以大大提高es的查詢性能，這個:sql_last_value是last_run_metadata_path中記錄的值