MySQL異構(gòu)數(shù)據(jù)到Es，一直是一個(gè)比較熱門的話題?，F(xiàn)成可用的框架也不在少數(shù)。如：Canal的ES適配器，ELK離線同步方案，還有很多其他開源相關(guān)方案。但是，基本都數(shù)據(jù)單表同步方案，實(shí)時(shí)性也比較受考驗(yàn)。如果遇到復(fù)雜的數(shù)據(jù)異構(gòu)時(shí)，很多時(shí)候是通過硬編碼方式處理。

考慮到降低開發(fā)成本，避免重復(fù)勞動(dòng)，基于硬編碼基礎(chǔ)上開發(fā)出一套適用于企業(yè)應(yīng)用接入的，通用的mysql2es數(shù)據(jù)異構(gòu)平臺(tái)。為了減少企業(yè)運(yùn)維，方案上也考慮兼容ALI云的DTS數(shù)據(jù)同步和訂閱方案。

數(shù)據(jù)流向

由RDS異構(gòu)索引數(shù)據(jù)到ALI的ES服務(wù)。主表數(shù)據(jù)使用DTS數(shù)據(jù)同步服務(wù)，其中舊數(shù)據(jù)使用全量同步，新數(shù)據(jù)使用增量同步。副表數(shù)據(jù)，使用DTS數(shù)據(jù)訂閱服務(wù)，新建notify工程根據(jù)條件模板更新主表數(shù)據(jù)。

數(shù)據(jù)流向設(shè)計(jì)

詳細(xì)設(shè)計(jì)

詳細(xì)流程設(shè)計(jì)

1、數(shù)據(jù)庫(kù)表通過DTS訂閱方式，同步到對(duì)應(yīng)的表。
2、全量同步通過DTS數(shù)據(jù)同步方案。
3、配置管理，主要用來管理同步數(shù)據(jù)相應(yīng)模板。
4、具體方案，要考慮到同步數(shù)據(jù)的復(fù)雜性。如果單表同步，使用DTS即可滿足；如果比較復(fù)雜的同步數(shù)據(jù)，使用DTS替代方案實(shí)現(xiàn)。

數(shù)據(jù)處理流程設(shè)計(jì)

DTS訂閱替代方案

使用canal+kafka/rocketmq+notify架構(gòu)對(duì)DTS訂閱服務(wù)進(jìn)行替換。
1、可以降低訂閱成本。
2、通過配置，實(shí)現(xiàn)更靈活的同步方案。

DTS訂閱服務(wù)替代方案

功能設(shè)計(jì)

配置管理

首先，配置表信息，保存到緩存。
第二，監(jiān)聽DTS訂閱服務(wù)，當(dāng)配置有更新，同時(shí)更新數(shù)據(jù)。
第三，讀取配置的時(shí)候，沒有緩存，則認(rèn)為無需處理。
第四，緩存使用schema_name.sub_table_name作為cache_key，緩存配置數(shù)據(jù)。緩存數(shù)據(jù)結(jié)構(gòu)為<String，List<c_data_sync_config>>

配置信息管理

數(shù)據(jù)更新功能

1、消費(fèi)到訂閱信息后，檢查是否存在配置，存在則認(rèn)為需要處理數(shù)據(jù)。
2、然后與配置中的es_index,es_type,es_query字段一起組裝ES查詢語(yǔ)句。并查出ES記錄（也可能不存在）。
3、通過ES記錄獲得_id，沒有則新增。然后與配置中的更新字段，有用數(shù)據(jù)組裝ES更新語(yǔ)句。并更新ES記錄。

數(shù)據(jù)更新流程

數(shù)據(jù)查詢功能

原銷售明細(xì)分頁(yè)查詢列表的Dao層，將由直接查詢db庫(kù)改為查詢elasitcsearch索引，得到主鍵id列表后，再次查詢db庫(kù)。

查詢功能流程

數(shù)據(jù)庫(kù)設(shè)計(jì)

一張表搞定。

數(shù)據(jù)同步配置表設(shè)計(jì)

核心流程回顧：

1、通過DTS訂閱或者CANAL獲得數(shù)據(jù)
2、通過配置校驗(yàn)是否需要數(shù)據(jù)同步處理
3、需要?jiǎng)t通過db_sql拼接sql查詢數(shù)據(jù)
4、通過es_sql，es_index，es_id 組裝ES查詢，獲取對(duì)應(yīng)的ES記錄。
5、存在則更新數(shù)據(jù)，不存在則新增數(shù)據(jù)。

補(bǔ)償方案

1、實(shí)時(shí)補(bǔ)償。按條件查詢時(shí)，對(duì)比數(shù)據(jù)庫(kù)和es的總數(shù)，不匹配，則查詢數(shù)據(jù)庫(kù)，然后補(bǔ)償遺漏ES。

2、晚上統(tǒng)一補(bǔ)償，先補(bǔ)漏單，再補(bǔ)一致性。漏單可以根據(jù)時(shí)間分片，而一致性，則可以通過es入庫(kù)的更新更新時(shí)間進(jìn)行校驗(yàn)補(bǔ)償。

3、快速補(bǔ)償方案。將同步數(shù)據(jù)校驗(yàn)庫(kù)，間隔半個(gè)小時(shí)進(jìn)行漏單和一致性校驗(yàn)。校驗(yàn)完后，清理校驗(yàn)庫(kù)數(shù)據(jù)。

全量同步方案

由于通過CANAL監(jiān)聽，到MQ，到NOTIFY消費(fèi)處理并入庫(kù)ES。整個(gè)流程保證實(shí)時(shí)性，但是鏈路較長(zhǎng)，且沒有使用ES批量更新的有事，經(jīng)測(cè)試TPS在100左右。因此，實(shí)行全量同步時(shí)，使用任務(wù)或前端觸發(fā)的方式，直接在NOTIFY端使用_bulk方式入庫(kù)ES，TPS可以提升到1000。

The End ! 整體結(jié)果比較滿意。 Mark The Thoughts！