前陣子老板安排了一個(gè)新任務(wù)，要建設(shè)一個(gè)商家商品搜索系統(tǒng)，能夠?yàn)橛脩籼峁┛焖?、?zhǔn)確的搜索能力。

[[319022]]

圖片來(lái)自 Pexels

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

背景很簡(jiǎn)單，現(xiàn)實(shí)蠻復(fù)雜!我們面臨以下幾個(gè)難題：

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

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

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

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

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

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

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

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

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

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

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

項(xiàng)目實(shí)戰(zhàn)

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

操作系統(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 實(shí)現(xiàn)數(shù)據(jù) ETL 到 MySQL8

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

①安裝 canaldeployer

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

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

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

serverMode

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

Kafka MQ 信息

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

如下圖所示：

destinations 實(shí)例配置

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

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

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

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

如下所示：

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

canaldeploy instance kafka topic配置

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

②安裝 canal.adapter

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

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

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

如下所示：

ETL 到 MySQL8 配置

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

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

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

canaladapter 日志

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

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

①配置第二個(gè) Canal 的 canaladapter

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

如下所示：

canaladapter MQ 及 MySQL 配置

canaladapter ES 配置

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

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

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

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

• 主表不能為子查詢語(yǔ)句。
• 只能使用 left outer join 即最左表一定要是主表。
• 關(guān)聯(lián)從表如果是子查詢不能有多張表。
• 主 sql 中不能有 where 查詢條件(從表子查詢中可以有 where 條件但是不推薦, 可能會(huì)造成數(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)條件必須要有一個(gè)字段出現(xiàn)在主查詢語(yǔ)句中比如：on a.role_id=b.id 其中的 a.role_id 或者 b.id 必須出現(xiàn)在主 select 語(yǔ)句中。
• ElasticSearch 的 mapping 屬性與 sql 的查詢值將一一對(duì)應(yīng)(不支持 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)系

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

建立 index 和父子文檔映射

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

⑤啟動(dòng) canal adapter

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

正確配置 adapter 日志示例

運(yùn)行結(jié)果

現(xiàn)在，我們可以通過(guò) Kibana 來(lái)執(zhí)行 DSL 語(yǔ)句來(lái)查詢看看。

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

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

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

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

總結(jié)

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

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

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

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