MySQL异构数据到Es，一直是一个比较热门的话题。现成可用的框架也不在少数。如：Canal的ES适配器，ELK离线同步方案，还有很多其他开源相关方案。但是，基本都数据单表同步方案，实时性也比较受考验。如果遇到复杂的数据异构时，很多时候是通过硬编码方式处理。

考虑到降低开发成本，避免重复劳动，基于硬编码基础上开发出一套适用于企业应用接入的，通用的mysql2es数据异构平台。为了减少企业运维，方案上也考虑兼容ALI云的DTS数据同步和订阅方案。

数据流向

由RDS异构索引数据到ALI的ES服务。主表数据使用DTS数据同步服务，其中旧数据使用全量同步，新数据使用增量同步。副表数据，使用DTS数据订阅服务，新建notify工程根据条件模板更新主表数据。

数据流向设计

详细设计

详细流程设计

1、数据库表通过DTS订阅方式，同步到对应的表。
2、全量同步通过DTS数据同步方案。
3、配置管理，主要用来管理同步数据相应模板。
4、具体方案，要考虑到同步数据的复杂性。如果单表同步，使用DTS即可满足；如果比较复杂的同步数据，使用DTS替代方案实现。

数据处理流程设计

DTS订阅替代方案

使用canal+kafka/rocketmq+notify架构对DTS订阅服务进行替换。
1、可以降低订阅成本。
2、通过配置，实现更灵活的同步方案。

DTS订阅服务替代方案

功能设计

配置管理

首先，配置表信息，保存到缓存。
第二，监听DTS订阅服务，当配置有更新，同时更新数据。
第三，读取配置的时候，没有缓存，则认为无需处理。
第四，缓存使用schema_name.sub_table_name作为cache_key，缓存配置数据。缓存数据结构为<String，List<c_data_sync_config>>

配置信息管理

数据更新功能

1、消费到订阅信息后，检查是否存在配置，存在则认为需要处理数据。
2、然后与配置中的es_index,es_type,es_query字段一起组装ES查询语句。并查出ES记录（也可能不存在）。
3、通过ES记录获得_id，没有则新增。然后与配置中的更新字段，有用数据组装ES更新语句。并更新ES记录。

数据更新流程

数据查询功能

原销售明细分页查询列表的Dao层，将由直接查询db库改为查询elasitcsearch索引，得到主键id列表后，再次查询db库。

查询功能流程

数据库设计

一张表搞定。

数据同步配置表设计

核心流程回顾：

1、通过DTS订阅或者CANAL获得数据
2、通过配置校验是否需要数据同步处理
3、需要则通过db_sql拼接sql查询数据
4、通过es_sql，es_index，es_id 组装ES查询，获取对应的ES记录。
5、存在则更新数据，不存在则新增数据。

补偿方案

1、实时补偿。按条件查询时，对比数据库和es的总数，不匹配，则查询数据库，然后补偿遗漏ES。

2、晚上统一补偿，先补漏单，再补一致性。漏单可以根据时间分片，而一致性，则可以通过es入库的更新更新时间进行校验补偿。

3、快速补偿方案。将同步数据校验库，间隔半个小时进行漏单和一致性校验。校验完后，清理校验库数据。

全量同步方案

由于通过CANAL监听，到MQ，到NOTIFY消费处理并入库ES。整个流程保证实时性，但是链路较长，且没有使用ES批量更新的有事，经测试TPS在100左右。因此，实行全量同步时，使用任务或前端触发的方式，直接在NOTIFY端使用_bulk方式入库ES，TPS可以提升到1000。

The End ! 整体结果比较满意。 Mark The Thoughts！