如何通過 Docker 部署 Logstash 同步 Mysql 資料庫資料到 ElasticSearch
在開發過程中，我們經常會遇到對業務資料進行模糊搜尋的需求，例如電商網站對於商品的搜尋，以及內容網站對於內容的關鍵字檢索等等。對於這些高階的搜尋功能，顯然資料庫的 Like 是不合適的，通常我們採用 ElasticSearch 來完成資料的搜尋和分析，有了這個利器，我們可以輕鬆應對上述場景，實現關鍵字搜尋等功能。
不過，由於增加了 ElasticSearch 作為搜尋引擎，隨之而來的問題就是，如何將業務中的資料同步到 ElasticSearch 中，主要有兩種方式：
- 業務雙寫（具有侵入性）
- 資料庫同步
由於業務雙寫需要更改業務程式碼，一般不建議採用此種方式，除非有強一致性要求，或者對業務侵入不敏感的系統可以採取此種方式：
- 強一致性：同步通過HTTP請求寫入 ElasticSearch
- 最終一致性：
- 可採取業務寫入日誌，後端通過日誌流資料過濾寫入 ElasticSearch（ELK標準模式，推薦）
- 另一種方案就是同步寫入 MQ，後端通過消費MQ非同步寫入 ElasticSearch
本文主要討論非程式碼侵入的資料庫同步方式，主要採用的是通過 LogStash 定時掃描資料庫來增量同步資料的方案。
資料庫指令碼
資料庫表結構中，需要有一個時間型別的欄位作為增量更新的標識欄位（例如 lastupdatetime），當該條資料更新時，必須同時更新該欄位。
CREATE TABLE user ( `id` int(11) NOT NULL, `name` varchar(50) NOT NULL, `age` int(11) NOT NULL, `createtime` datetime(0) NOT NULL, `lastupdatetime` datetime(0) NOT NULL, PRIMARY KEY (`id`) USING BTREE ) INSERT INTO `user` VALUES(1,"jack",18,Now(),Now()) INSERT INTO `user` VALUES(2,"William",18,Now(),Now()) SELECT * from `user`
查詢結果：
| id | name | age | createtime | lastupdatetime |
|---|---|---|---|---|
| 1 | jack | 18 | 2019-10-24 10:31:14 | 2019-10-24 10:31:14 |
| 2 | William | 18 | 2019-10-24 10:31:49 | 2019-10-24 10:31:49 |
LogStash 配置資訊
logstash docker 安裝指令碼：
mkdir /opt/logstashsync/
mkdir /opt/logstashsync/pipeline
vi /opt/logstashsync/pipeline/logstash.conf
input { jdbc { jdbc_driver_library => "/app/mysql-connector-java-8.0.18.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://192.168.10.102:3306/synctest" jdbc_user => "root" jdbc_password => "123456" tracking_column => "unix_ts_in_secs" use_column_value => true schedule => "*/5 * * * * *" statement => "SELECT *, UNIX_TIMESTAMP(lastupdatetime) AS unix_ts_in_secs FROM user WHERE (UNIX_TIMESTAMP(lastupdatetime) > :sql_last_value AND lastupdatetime < NOW()) ORDER BY lastupdatetime ASC" } } filter { mutate { copy => { "id" => "[@metadata][_id]"} remove_field => ["id", "@version", "unix_ts_in_secs"] } } output { elasticsearch { hosts => "192.168.10.102:9200" index => "syncuser" timeout => 300 document_id => "%{[@metadata][_id]}" } }
上述配置說明：
- jdbc_driver_library：logstash的映象中並不包含 jdbc connector，需要在官方網站中下載下來之後，在容器啟動時對映到容器中，可點此下載。
- tracking_column：用於跟蹤 Logstash從MySQL讀取的最後最後一條資料的 lastupdatetime 的值，並預設持久化到磁碟檔案 .logstash_jdbc_last_run 中。該值用於在下一次迴圈同步時，同步的起始值，從而達到增量同步的作用，儲存在 .logstash_jdbc_last_run 在 SQL 語句中可以以 :sql_last_value 訪問。
- schedule：設定多久迴圈同步一次，以cron語法指定，我們當前設定的是5秒一次迴圈。
- statement：執行同步的SQL語句。值得注意的是where條件中為什麼要這麼寫，可以參考 https://www.elastic.co/blog/how-to-keep-elasticsearch-synchronized-with-a-relational-database-using-logstash 文章中給定的解釋。
- 重要: 關於上述配置中的 [@metadata][_id]，在同步過程中，必須使用資料庫資料id作為 ElasticSearch 中的文件 _id，這樣當資料庫中該條資料有修改時，ElasticSearch 中的文件才會相應的同步修改，否則會以一條新的資料插入 ElasticSearch，導致資料同步錯誤。
有了上述配置，我們把 Logstath 的 docker 容器跑起來：
docker run -d \
-v /opt/logstashsync/config/logstash.yml:/usr/share/logstash/config/logstash.yml \
-v /opt/logstashsync/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf \
-v /opt/logstashsync/mysql-connector-java-8.0.18.jar:/app/mysql-connector-java-8.0.18.jar \
--name=logstash \
logstash:6.7.1注意：上述指令碼可以看到，我們將本地 /opt/logstashsync/ 目錄下的 mysql-connector-java-8.0.18.jar 對映到了容器的 /app 目錄下，對應在上述 logstash.conf 中的配置的 jdbc_driver_library 的值
通過檢視 Logstash 容器執行日誌，我們可以看到如下日誌內容,說明該容易已經按照我們預期的每5s同步一次資料庫：
[2019-10-25T06:27:59,056][INFO ][logstash.inputs.jdbc ] (0.039651s) SELECT *, UNIX_TIMESTAMP(lastupdatetime) AS unix_ts_in_secs FROM user WHERE (UNIX_TIMESTAMP(lastupdatetime) > 0 AND lastupdatetime < NOW()) ORDER BY lastupdatetime ASC
[2019-10-25T06:28:05,154][INFO ][logstash.inputs.jdbc ] (0.004232s) SELECT *, UNIX_TIMESTAMP(lastupdatetime) AS unix_ts_in_secs FROM user WHERE (UNIX_TIMESTAMP(lastupdatetime) > 1571913109 AND lastupdatetime < NOW()) ORDER BY lastupdatetime ASC
[2019-10-25T06:28:10,230][INFO ][logstash.inputs.jdbc ] (0.002832s) SELECT *, UNIX_TIMESTAMP(lastupdatetime) AS unix_ts_in_secs FROM user WHERE (UNIX_TIMESTAMP(lastupdatetime) > 1571913109 AND lastupdatetime < NOW()) ORDER BY lastupdatetime ASC
通過Kibana查詢同步結果
在 Kibana 中建立 syncuser index，即可以檢視到已經同步的資料：
嘗試對資料庫資料做更新操作,將名為 William 的使用者年齡修改為100（記得同時要更新lastupdatetime 欄位）：
UPDATE `user` SET age=100, lastupdatetime=NOW() WHERE `name`='William';
SELECT * from `user`
再次檢視 Kibana 中的資料，可以看到該資料已經成功同步：
結語
根據上述過程，我們完成了簡單的單表資料定時同步至 ElasticSearch 過程，但是在實際使用過程中，需要注意以下問題：
- sql語句需要考慮每次同步最大條數。大多數情況下，資料庫可能已經存在大量資料，如果不做控制，可能會導致 Logstash 剛啟動時一次同步的資料量過大，發生異常，採取的方式可以在 SQL 語句中增加每次獲取最大條數限制。
- 增量更新的標識欄位，既然是通過>號方式判斷，那麼如果id是自增主鍵，也可以採用 int 型別的主鍵欄位，這樣可以減少在資料庫中建立 lastupdatetime 索引。但如果不是主鍵，則需要謹慎使用，具體原因請仔細參考上述配置說明中 statement 給出的連結。
- 由於增量同步機制所致，所有資料庫中的刪除操作應該以軟刪除的方式進行，即增加 is_delete 欄位，否則如果硬刪除會導致該條資料狀態無法同步至 ElasticSearch，當然在查詢 ElasticSearch 時，也應該增加該條件，排除已經刪除的資料。