前陣子老板安排了一個新任務(wù)，要建設(shè)一個商家商品搜索系統(tǒng)，能夠為用戶提供快速、準確的搜索能力。

[[319022]]

圖片來自 Pexels

設(shè)計要求在用戶輸入搜索內(nèi)容時，要能從商家名稱和商品名稱兩個維度去搜索，搜索出來的結(jié)果，按照準確率排序，并按商家所屬商品的關(guān)聯(lián)關(guān)系，來組合數(shù)據(jù)結(jié)構(gòu)，同時提供 API 給業(yè)務(wù)系統(tǒng)調(diào)用。

背景很簡單，現(xiàn)實蠻復雜!我們面臨以下幾個難題：

• 商家數(shù)據(jù)庫和商品數(shù)據(jù)庫是多臺不同的服務(wù)器，并且數(shù)據(jù)量達百萬級，如何才能實現(xiàn)跨數(shù)據(jù)庫的數(shù)據(jù)同步呢?
• 商家和商品的數(shù)據(jù)是有從屬關(guān)系的，不然就會把肯德基的香辣雞腿堡掛到麥當勞去，這就尷尬了!
• 商家商品數(shù)據(jù)是經(jīng)常更新的，比如修改價格、庫存、上下架等，那搜索服務(wù)可不能搜出一堆過時的數(shù)據(jù)，如果客戶明明搜出來的商品，點進去后卻已下架了，那么客戶就要吐槽了!如何實現(xiàn)搜索數(shù)據(jù)與源數(shù)據(jù)庫增刪改均實時同步呢?

帶著以上三個問題，我們開始了搜索服務(wù)的整體架構(gòu)設(shè)計。

系統(tǒng)架構(gòu)設(shè)計思路

為了設(shè)計出合適的系統(tǒng)架構(gòu)，我們分析了現(xiàn)狀：

首先，商家數(shù)據(jù)和商品數(shù)據(jù)分別存儲在 2 個獨立的 MySQL8 數(shù)據(jù)庫，為滿足商家數(shù)據(jù)和商品數(shù)據(jù)的關(guān)聯(lián)，我們需要將兩個庫中所需要的表實時 ETL 到我們的搜索系統(tǒng)數(shù)據(jù)庫。

其次，數(shù)據(jù)從商家、商品數(shù)據(jù)庫 ETL 到搜索系統(tǒng)數(shù)據(jù)庫后，需要實時的組合成為商家關(guān)聯(lián)商品數(shù)據(jù)結(jié)構(gòu)，并以父子文檔的格式，存儲到 ES 中。

最后，商家、商品數(shù)據(jù)庫的增刪改操作，需要實時的同步到 ES 中，也就是 ES 中的數(shù)據(jù)，需要支持實時的增加、刪除和修改。

為此，我們設(shè)計了 2 個 Canal 組件，第一個 Canal 實現(xiàn)數(shù)據(jù) ETL，把商家、商品數(shù)據(jù)庫的某些表及字段，抽取到搜索服務(wù)數(shù)據(jù)庫。

再利用第二個 Canal，讀取搜索服務(wù) MySQL 數(shù)據(jù)庫的 Binlog，實時傳輸?shù)?Kafka 消息隊列，再由 canal adapter 對數(shù)據(jù)進行關(guān)聯(lián)、父子文檔映射等，將處理好的數(shù)據(jù)存儲到 ElasticSearch 中。

具體系統(tǒng)架構(gòu)設(shè)計如下圖所示：

商家商品搜索系統(tǒng)架構(gòu)設(shè)計

項目實戰(zhàn)

環(huán)境及軟件說明

操作系統(tǒng)：CentOS 7 
canal：canal.adapter-1.1.4，canal.deployer-1.1.4 
kafka：kafka_2.12-2.3.0 
ElasticSearch：elasticsearch-6.3.2 
kibana：kibana-6.3.2 

用 Canal 實現(xiàn)數(shù)據(jù) ETL 到 MySQL8

這個步驟是利用 Canal 從 2 個獨立的 MySQL8 數(shù)據(jù)庫中，抽取需要的表到搜索服務(wù)的 MySQL 數(shù)據(jù)庫。

①安裝 canaldeployer

解壓 canal.deployer-1.1.4.tar.gz，并配置 canal deployer。

進入 canaldeployer/conf 目錄，修改 canal.properties 文件，主要配置 serverMode、MQ 和 destination 三部分。

首先，我們 serverMode 修改為 Kafka 模式，增加系統(tǒng)緩沖能力以及提高系統(tǒng)穩(wěn)定性：

serverMode

接著，配置 Kafka 的 MQ 信息(Kafka 請自行安裝)：

Kafka MQ 信息

最后，配置需要實例化的 instance，這里配置了 3 個，表示 canal deploy 會啟動這 3 個實例，同步 MySQL 的 Binlog 到 Kafka 的 Topic 內(nèi)。

如下圖所示：

destinations 實例配置

配置 canal deployer instance：進入 canaldeployer/conf/example 目錄，發(fā)現(xiàn)有一個 instance.properties 文件，這是 Canal 給的示例，我們可以參考其配置。

我們拷貝整個 example 目錄，并重命名為上個步驟配置的 destination 之一，如 xxxsearch。

進入 xxxsearch 目錄，編輯 instance.properties 文件，主要配置源數(shù)據(jù)庫信息、所需數(shù)據(jù)表及字段，以及指定 Kafka 的 Topic 名。

這樣源數(shù)據(jù)庫的 Binlog 就會轉(zhuǎn)換為 Json 數(shù)據(jù)，并實時的通過 canal deployer 傳輸?shù)?Kafka 該 Topic 中。

如下所示：

canaldeploy instance 源數(shù)據(jù)庫配置

canaldeploy instance kafka topic配置

進入 canaldeployer/bin 目錄，執(zhí)行 ./startup.sh，啟動 canal deployer 及所屬實例。至此 canal deployer 搭建完成。

②安裝 canal.adapter

我們需要利用 canal.adapter 將 Kafka Topic 中的 binlog json 數(shù)據(jù)，經(jīng)過清洗轉(zhuǎn)換等操作，存儲到 MySQL8 中。由于 Canal 原生是不支持 MySQL8 的，故我們需要做一些調(diào)整。

增加 MySQL8 連接驅(qū)動：解壓 canal.adapter-1.1.4.tar.gz，進入 canaladapter/lib 目錄，移除 mysql-connector-java-5.1.40.jar，導入 mysql-connector-java-8.0.18.jar。

配置 canal adapter，使數(shù)據(jù)輸出到 MySQL8：進入 canaladapter/conf 目錄，編輯 application.yml 文件，主要配置消費 Kafka、源數(shù)據(jù)庫信息和搜索系統(tǒng)數(shù)據(jù)庫信息。

如下所示：

ETL 到 MySQL8 配置

接著，進入 canaladapter/conf/rdb 目錄，以官方提供的 mytest_user.yml 為例，配置 Kafka Topic 名、源數(shù)據(jù)庫名、源數(shù)據(jù)表名，以及目標數(shù)據(jù)庫名和目標數(shù)據(jù)表名，建議一張表對應一個 yml 文件。

ETL 表結(jié)構(gòu)映射配置

啟動 canaladapter：進入 canaladapter/bin 目錄，執(zhí)行 ./startup.sh，啟動 canal adapter，觀察 logs/adapter/adapter.log 日志文件，手動在搜索系統(tǒng)數(shù)據(jù)庫新增一條記錄，看是否會打印如下日志，即有 2 條記錄，一條 INFO，一條 DEBUG，則表示配置成功。

canaladapter 日志

至此，數(shù)據(jù) ETL 階段搭建完成，數(shù)據(jù)可從兩個不同的 MySQL8 數(shù)據(jù)庫，實時同步到搜索服務(wù)的 MySQL 數(shù)據(jù)庫。

實現(xiàn)數(shù)據(jù)多表關(guān)聯(lián)、父子文檔映射

①配置第二個 Canal 的 canaladapter

進入 canaladapter/conf 目錄，編輯 application.yml 文件，主要配置消費 Kafka、搜索系統(tǒng)數(shù)據(jù)庫，和 ES 連接信息。

如下所示：

canaladapter MQ 及 MySQL 配置

canaladapter ES 配置

②配置多表關(guān)聯(lián)

進入 canaladapter/conf/es 目錄，vim mytest_user.yml，編輯多表關(guān)聯(lián)配置：

多表關(guān)聯(lián)配置

注意，sql支持多表關(guān)聯(lián)自由組合, 但是有一定的限制：

• 主表不能為子查詢語句。
• 只能使用 left outer join 即最左表一定要是主表。
• 關(guān)聯(lián)從表如果是子查詢不能有多張表。
• 主 sql 中不能有 where 查詢條件(從表子查詢中可以有 where 條件但是不推薦, 可能會造成數(shù)據(jù)同步的不一致，比如修改了 where 條件中的字段內(nèi)容)。
• 關(guān)聯(lián)條件只允許主外鍵的'='操作不能出現(xiàn)其他常量判斷比如：on a.role_id=b.id and b.statues=1。
• 關(guān)聯(lián)條件必須要有一個字段出現(xiàn)在主查詢語句中比如：on a.role_id=b.id 其中的 a.role_id 或者 b.id 必須出現(xiàn)在主 select 語句中。
• ElasticSearch 的 mapping 屬性與 sql 的查詢值將一一對應(不支持 select *)。
• 比如：select a.id as _id，a.name，a.email as _email from user，其中 name 將映射到 es mapping 的 name field，_email 將映射到 mapping 的 _email field，這里以別名(如果有別名)作為最終的映射字段。這里的 _id 可以填寫到配置文件的 _id: _id 映射。

③配置父子文檔

以官方的 biz_order.yml 為例，vim biz_order.yml，配置父子文檔映射：

配置父子文檔映射

④在 ElasticSearch6 中，建立 index 和父子文檔映射關(guān)系

進入 Kibana 頁面，點擊 Dev Tools，執(zhí)行如下命令，即可建立索引及父子文檔映射：

建立 index 和父子文檔映射

其中，ES6 和 Kibana 的安裝，在此無特別配置，不做贅述。

⑤啟動 canal adapter

進入 canaladapter/bin 目錄，執(zhí)行 ./startup.sh，啟動 canal adapter，觀察 logs/adapter/adapter.log 日志文件，手動在搜索系統(tǒng)數(shù)據(jù)庫新增一條記錄，看是否會打印如下日志，如打印則表示配置成功。

正確配置 adapter 日志示例

運行結(jié)果

現(xiàn)在，我們可以通過 Kibana 來執(zhí)行 DSL 語句來查詢看看。

我們事先已在商家系統(tǒng)中增加了一個“肯德基”商店，然后在商品系統(tǒng)中添加了“西紅柿”和“新鮮西紅柿”2 個商品，并將商品關(guān)聯(lián)到“肯德基”上。

接著我們查詢“肯德基”或者“西紅柿”，得到以下是查詢的結(jié)果(去除了 ES 默認字段)：

通過 DSL 查詢的結(jié)果

由圖可見，我們可以通過商家名查詢商品，也可通過商品名查詢商店和商品，并且 Canal 支持數(shù)據(jù)的實時增刪改，所以 ES 的數(shù)據(jù)也會與商家系統(tǒng)和商品系統(tǒng)保持一致，同時數(shù)據(jù)結(jié)構(gòu)包含商家及對應的商品，滿足業(yè)務(wù)需求。

總結(jié)

至此，基于 Canal、Kafka、MySQL8、ElasticSearch6 技術(shù)的商家商品搜索系統(tǒng)基礎(chǔ)框架搭建完成。

我們采用 canal deployer 實時讀取商家、商品系統(tǒng)的 MySQL 數(shù)據(jù)庫 Binlog，并發(fā)送至 Kafka。

接著由 canal adapter 消費 Kafka，并將 binlog json 數(shù)據(jù)進行多表關(guān)聯(lián)、父子文檔映射，最后存儲到 ES6 中，供上層搜索服務(wù)調(diào)用。

搜索服務(wù)系統(tǒng)最終成功上線，為公司百萬級商家商品提供實時數(shù)據(jù)同步，秒級搜索結(jié)果展示，達到業(yè)務(wù)要求，老板說了，給研發(fā)團隊每人加個雞腿!想想還有點小激動，嘿嘿!