基于 MySQL Binlog 的 Elasticsearch 資料同步實踐 原

一、背景

随着馬蜂窩的逐漸發展，我們的業務資料越來越多，單純使用 MySQL 已經不能滿足我們的資料查詢需求，例如對于商品、訂單等資料的多元度檢索。

使用 Elasticsearch 存儲業務資料可以很好的解決我們業務中的搜尋需求。而資料進行異構存儲後，随之而來的就是資料同步的問題。

二、現有方法及問題

對于資料同步，我們目前的解決方案是建立資料中間表。把需要檢索的業務資料，統一放到一張MySQL 表中，這張中間表對應了業務需要的 Elasticsearch 索引，每一列對應索引中的一個Mapping 字段。通過腳本以 Crontab 的方式，讀取 MySQL 中間表中 UTime 大于上一次讀取時間的所有資料，即該段時間内的增量，寫入Elasticsearch。

是以，一旦業務邏輯中有相應字段的資料變更，需要同時顧及 MySQL 中間表的變更；如果需要 Elasticsearch 中的資料即時性較高，還需要同時寫入 Elasticsearch。

随着業務資料越來越多，MySQL 中間表的資料量越來越大。當需要在 Elasticsearch 的索引中新增 Mapping 字段時，相應的 MySQL 中間表也需要新增列，在資料量龐大的表中，擴充列的耗時是難以忍受的。

而且 Elasticsearch 索引中的 Mapping 字段随着業務發展增多，需要由業務方增加相應的寫入 MySQL 中間表方法，這也帶來一部分開發成本。

三、方案設計

1. 整體思路

現有的一些開源資料同步工具，如阿裡的 DataX 等，主要是基于查詢來擷取資料源，這會存在如何确定增量（比如使用utime字段解決等）和輪詢頻率的問題，而我們一些業務場景對于資料同步的實時性要求比較高。為了解決上述問題，我們提出了一種基于 MySQL Binlog 來進行 MySQL 資料同步到 Elasticsearch 的思路。Binlog 是 MySQL 通過 Replication 協定用來做主從資料同步的資料，是以它有我們需要寫入 Elasticsearch 的資料，并符合對資料同步時效性的要求。

使用 Binlog 資料同步 Elasticsearch，業務方就可以專注于業務邏輯對 MySQL 的操作，不用再關心資料向 Elasticsearch 同步的問題，減少了不必要的同步代碼，避免了擴充中間表列的長耗時問題。

經過調研後，我們采用開源項目 go-mysql-elasticsearch 實作資料同步，并針對馬蜂窩技術棧和實際的業務環境進行了一些定制化開發。

2. 資料同步正确性保證

公司的所有表的 Binlog 資料屬于機密資料，不能直接擷取，為了滿足各業務線的使用需求，采用接入 Kafka 的形式提供給使用方，并且需要使用方申請相應的 Binlog 資料使用權限。擷取使用權限後，使用方以 Consumer Group 的形式讀取。

這種方式保證了 Binglog 資料的安全性，但是對保證資料同步的正确性帶來了挑戰。是以我們設計了一些機制，來保證資料源的擷取有序、完整。

1). 順序性

通過 Kafka 擷取 Binlog 資料，首先需要保證擷取資料的順序性。嚴格說，Kafka 是無法保證全局消息有序的，隻能局部有序，是以無法保證所有 Binlog 資料都可以有序到達 Consumer。

但是每個 Partition 上的資料是有序的。為了可以按順序拿到每一行 MySQL 記錄的 Binglog，我們把每條 Binlog 按照其 Primary Key，Hash 到各個 Partition 上，保證同一條 MySQL 記錄的所有 Binlog 資料都發送到同一個 Partition。

如果是多 Consumer 的情況，一個 Partition 隻會配置設定給一個 Consumer，同樣可以保證 Partition 内的資料可以有序的 Update 到 Elasticsearch 中。

2).  完整性

考慮到同步程式可能面臨各種正常或異常的退出，以及 Consumer 數量變化時的 Rebalance，我們需要保證在任何情況下不能丢失 Binlog 資料。

利用 Kafka 的 Offset 機制，在确認一條 Message 資料成功寫入 Elasticsearch 後，才 Commit 該條 Message 的 Offset，這樣就保證了資料的完整性。而對于資料同步的使用場景，在保證了資料順序性和完整性的情況下，重複消費是不會有影響的。

四、技術實作

1. 功能子產品

配置解析子產品

負責解析配置檔案（toml 或 json 格式），或在配置中心（Skipper）配置的 json 字元串。包括 Kafka 叢集配置、Elasticsearch 位址配置、日志記錄方式配置、MySQL 庫表及字段與 Elasticsearch 的 Index 和 Mapping 對應關系配置等。

規則子產品

規則子產品決定了一條 Binlog 資料應該寫入到哪個 Elasticsearch 索引、文檔_id 對應的 MySQL 字段、Binlog 中的各個 MySQL 字段與索引 Mapping 的對應關系和寫入類型等。

在本地化過程中，根據我們的業務場景，增加了對 MySQL 表各字段的 where 條件判斷，來過濾掉不需要的 Binlog 資料。

Kafka 相關子產品

該子產品負責連接配接 Kafka 叢集，擷取 Binlog 資料。

在本地化過程中，該子產品的大部分功能已經封裝成了一個通用的 Golang Kafka Consumer Client。包括 Dba Binlog 訂閱平台要求的 SASL 認證，以及從指定時間點的 Offset 開始消費資料。

Binlog 資料解析子產品

原項目中的 Binlog 資料解析針對的是原始的 Binlog 資料，包含了解析 Replication 協定的實作。在我們的使用場景中，Binlog 資料已經是由 canal 解析成的 json 字元串，是以對該子產品的功能進行了簡化。

binlog json字元串示例

上面是一個簡化的 binlog json 字元串，通過該條 binlog 的 database 和 table 可以命中一條配置規則，根據該配置規則，把 Data 中的 key-value 構造成一個與對應 Elasticsearch 索引相比對的 key-value map，同時包括一些資料類型的轉換：

Elasticsearch相關子產品

Binlog 資料解析子產品生成的 key-value map，由該子產品拼裝成請求_bulk 接口的 update payload，寫入 Elasticsearch。考慮到 MySQL 頻繁更新時對 Elasticsearch 的寫入壓力，key-value map 會暫存到一個 slice 中，每 200ms 或 slice 長度達到一定長度時（可以通過配置調整），才會調用 Elasticsearch 的_bulk 接口，寫入資料。

2. 定制化開發

1）. 适應業務需求

upsert

業務中使用的索引資料可能是來自多個不同的表，同一個文檔的資料來自不同表的時候，先到的資料是一條 index，後到的資料是一條 update，在我們無法控制先後順序時，需要實作 upsert 功能。在_bulk 參數中加入

{
    "doc_as_upsert" : true
}
           

Filter

實際業務場景中，可能業務需要的資料隻是某張表中的部分資料，比如用 type 字段辨別該條資料來源，隻需要把 type=1或2的資料同步到 Elasticsearch 中。我們擴充了規則配置，可以支援對 Binlog 指定字段的過濾需求，類似：

select * from sometable where type in (1,2)

           

2）快速增量

資料同步一般分為全量和增量。接入一個業務時，首先需要把業務現有的曆史 MySQL 資料導入到 Elasticsearch 中，這部分為全量同步。在全量同步過程中以及後續增加的資料為增量資料。

在全量資料同步完成後，如果從最舊開始消費 Kafka，隊列資料量很大的情況下，需要很長時間增量資料才能追上目前進度。為了更快的拿到所需的增量 Binlog，在 Consumer Group 消費 Kafka 之前，先擷取各個 Topic 的 Partition 在指定時間的 offset 值，并 commit 這些 offset，這樣在 Consumer Group 連接配接 Kafka 叢集時，會從剛才送出的 offset 開始消費，可以立即拿到所需的增量 Binlog。

3）. 微服務和配置中心

項目使用馬蜂窩微服務部署，為新接入業務提供了快速上線支援，并且在業務 Binlog 資料突增時可以友善快速的擴容 Consumer。

馬蜂窩配置中心支援了各個接入業務的配置管理，相比于開源項目中的 toml 格式配置檔案，使用配置中心可以更友善的管理不同業務不同環境的配置。

五、日志與監控

從上圖中可以看出，訂單各個表的資料同步延時平均在 1s 左右。把延時資料接入 ElastAlert，在延時資料過多時發送報警通知。

另一個監控名額是心跳檢測，單獨建立一張獨立于業務的表，crontab 腳本每分鐘修改一次該表，同時檢查上一次修改是否同步到了指定的索引，如果沒有，則發送報警通知。該心跳檢測，監控了整個流程上的 Kafka、微服務和 ES，任何一個會導緻資料不同步的環節出問題，都會第一個接到通知。

六、結語

目前接入的最重要業務方是電商的訂單索引，資料同步延時穩定在 1s 左右。這次的開源項目本地化實踐，希望能為一些有 Elasticsearch 資料同步需求的業務場景提供幫助。

本文作者：張坤，馬蜂窩電商研發團隊度假業務進階研發工程師。

（馬蜂窩技術原創内容，轉載請保留出處及文末二維碼，謝謝）

posted on 2019-07-15 10:45 馬蜂窩技術 閱讀(...) 評論(...) 編輯 收藏