elasticsearch + kibana 叢集環境搭建
mysql 作為成熟穩定的資料持久化解決方案,廣泛地應用在各種領域,但是在資料分析方面稍有不足,而 elasticsearch 作為資料分析領域的佼佼者,剛好可以彌補這項不足,而我們要做的只需要將 mysql 中的資料同步到 elasticsearch 中即可,而 logstash 剛好就可以支援,所有你需要做的只是寫一個配置檔案而已
logstash 獲取
獲取 logstash
wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.3.zip
unzip logstash-6.2.3.zip && cd logstash-6.2 .3
安裝 jdbc 和 elasticsearch 外掛
bin/logstash-plugin install logstash-input-jdbc
bin/logstash-plugin install logstash-output-elasticsearch
獲取 jdbc mysql 驅動
wget https://cdn.mysql.com//Downloads/Connector-J/mysql-connector-java-5.1.46.zip
unzip mysql-connector-java-5.1.46.zip
編寫配置檔案
logstash-input-jdbc
使用 logstash-input-jdbc 外掛讀取 mysql 的資料,這個外掛的工作原理比較簡單,就是定時執行一個 sql,然後將 sql 執行的結果寫入到流中,增量獲取的方式沒有通過 binlog 方式同步,而是用一個遞增欄位作為條件去查詢,每次都記錄當前查詢的位置,由於遞增的特性,只需要查詢比當前大的記錄即可獲取這段時間內的全部增量,一般的遞增欄位有兩種,AUTO_INCREMENT
的主鍵 id
和 ON UPDATE CURRENT_TIMESTAMP
的 update_time
欄位,id
欄位只適用於那種只有插入沒有更新的表,update_time
更加通用一些,建議在 mysql 表設計的時候都增加一個 update_time
input {
jdbc {
jdbc_driver_library => "../mysql-connector-java-5.1.46/mysql-connector-java-5.1.46-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://<mysql_host>:3306/rta"
jdbc_user => "<username>"
jdbc_password => "<password>"
schedule => "* * * * *"
statement => "SELECT * FROM table WHERE update_time > :sql_last_value"
use_column_value => true
tracking_column_type => "timestamp"
tracking_column => "update_time"
last_run_metadata_path => "syncpoint_table"
}
}
jdbc_driver_library
: jdbc mysql 驅動的路徑,在上一步中已經下載jdbc_driver_class
: 驅動類的名字,mysql 填com.mysql.jdbc.Driver
就好了jdbc_connection_string
: mysql 地址jdbc_user
: mysql 使用者jdbc_password
: mysql 密碼schedule
: 執行 sql 時機,類似 crontab 的排程statement
: 要執行的 sql,以 “:” 開頭是定義的變數,可以通過 parameters 來設定變數,這裡的sql_last_value
是內建的變數,表示上一次 sql 執行中 update_time 的值use_column_value
: 使用遞增列的值tracking_column_type
: 遞增欄位的型別,numeric
表示數值型別,timestamp
表示時間戳型別tracking_column
: 遞增欄位的名稱,這裡使用 update_time 這一列,這列的型別是timestamp
last_run_metadata_path
: 同步點檔案,這個檔案記錄了上次的同步點,重啟時會讀取這個檔案,這個檔案可以手動修改
logstash-output-elasticsearch
output {
elasticsearch {
hosts => ["172.31.22.165", "172.31.17.241", "172.31.30.84", "172.31.18.178"]
user => "<user>"
password => "<password>"
index => "table"
document_id => "%{id}"
}
}
hosts
: es 叢集地址user
: es 使用者名稱password
: es 密碼index
: 匯入到 es 中的 index 名,這裡我直接設定成了 mysql 表的名字document_id
: 匯入到 es 中的文件 id,這個需要設定成主鍵,否則同一條記錄更新後在 es 中會出現兩條記錄,%{id}
表示引用 mysql 表中id
欄位的值
執行
把上面的程式碼儲存到一個配置檔案裡面 sync_table.cfg
,執行下面命令即可
cd logstash-6.2.3 && bin/logstash -f config/sync_table.cfg
如果成功了會在標準輸出輸出執行的 sql 語句
[2018-04-14T18:12:00,278][INFO ][logstash.inputs.jdbc ] (0.001011s) SELECT version()
[2018-04-14T18:12:00,284][INFO ][logstash.inputs.jdbc ] (0.000723s) SELECT * FROM table WHERE update_time > '2018-04-14 17:55:00'
其他問題
多表同步
一個 logstash 例項可以藉助 pipelines 機制同步多個表,只需要寫多個配置檔案就可以了,假設我們有兩個表 table1 和 table2,對應兩個配置檔案 sync_table1.cfg
和 sync_table2.cfg
在 config/pipelines.yml
中配置
- pipeline.id: table1
path.config: "config/sync_table1.cfg"
- pipeline.id: table2
path.config: "config/sync_table2.cfg"
直接 bin/logstash
啟動即可
@timestamp
欄位
預設情況下 @timestamp
欄位是 logstash-input-jdbc 新增的欄位,預設是當前時間,這個欄位在資料分析的時候非常有用,但是有時候我們希望使用資料中的某些欄位來指定這個欄位,這個時候可以使用 filter.date, 這個外掛是專門用來設定 @timestamp
這個欄位的
比如我有我希望用欄位 timeslice
來表示 @timestamp
,timeslice
是一個字串,格式為 %Y%m%d%H%M
filter {
date {
match => [ "timeslice", "yyyyMMddHHmm" ]
timezone => "Asia/Shanghai"
}
}
把這一段配置加到 sync_table.cfg
中,現在 @timestamp
和 timeslice
一致了