如何通過 Docker 部署 Logstash 同步 Mysql 數據庫數據到 ElasticSearch

在開發過程中，我們經常會遇到對業務數據進行模糊搜索的需求，例如電商網站對於商品的搜索，以及內容網站對於內容的關鍵字檢索等等。對於這些高級的搜索功能，顯然 數據庫的 Like 是不合適的，通常我們采用 ElasticSearch 來完成數據的搜索和分析，有了這個利器，我們可以輕松應對上述場景，實現關鍵字搜索等功能。

不過，由於增加了 ElasticSearch 作為搜索引擎，隨之而來的問題就是，如何將業務中的數據同步到 ElasticSearch 中，主要有兩種方式：

1. 業務雙寫（具有侵入性）
2. 數據庫同步

由於業務雙寫需要更改業務代碼，一般不建議采用此種方式，除非有強一致性要求，或者對業務侵入不敏感的系統可以采取此種方式：

• 強一致性：同步通過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 過程，但是在實際使用過程中，需要注意以下問題：

1. sql語句需要考慮每次同步最大條數。大多數情況下，數據庫可能已經存在大量數據，如果不做控制，可能會導致 Logstash 剛啟動時一次同步的數據量過大，發生異常，采取的方式可以在 SQL 語句中增加每次獲取最大條數限制。
2. 增量更新的標識字段，既然是通過>號方式判斷，那么如果id是自增主鍵，也可以采用 int 類型的主鍵字段，這樣可以減少在數據庫中創建 lastupdatetime 索引。但如果不是主鍵，則需要謹慎使用，具體原因請仔細參考上述配置說明中 statement 給出的鏈接。
3. 由於增量同步機制所致，所有數據庫中的刪除操作應該以軟刪除的方式進行，即增加 is_delete 字段，否則如果硬刪除會導致該條數據狀態無法同步至 ElasticSearch，當然在查詢 ElasticSearch 時，也應該增加該條件，排除已經刪除的數據。