如何通過 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`

查詢結果：

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 =https://www.cnblogs.com/xboo/p/> 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 時，也應該增加該條件，排除已經洗掉的資料，

標籤：.NET Core

上一篇：（22）ASP.NET Core2.2 EF創建模型（索引、備用鍵、繼承、支持欄位）

下一篇：輕量級ORM《sqlcommon》第一個版本發布了！！！