百万级商品数据实时同步，查询结果秒出
商家商品搜索系统架构设计
项目实战
1、环境及软件说明
操作系统：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
2、利用Canal实现数据ETL到MySQL8
这个步骤是利用canal从2个独立的MySQL8数据库中，抽取需要的表到搜索服务的MySQL数据库。
2.1 安装canaldeployer
（1）解压canal.deployer-1.1.4.tar.gz
（2）配置canal deployer
进入canaldeployer/conf目录，修改canal.properties文件，主要配置serverMode、MQ和destination三部分。
首先，我们serverMode修改为kafka模式，增加系统缓冲能力以及提高系统稳定性：
serverMode
接着，配置kafka的MQ信息（kafka请自行安装）：
kafka MQ信息
最后，配置需要实例化的instance，这里配置了3个，表示canal deploy会启动这3个实例，同步MySQL的binlog到kafka的topic内。如下图所示：
destinations实例配置
（3）配置canal deployer instance
进入canaldeployer/conf/example目录，发现有一个
instance.properties文件，这是canal给的示例，我们可以参考其配置。
①我们拷贝整个example目录，并重命名为上个步骤配置的destination之一，如xxxsearch；
②进入xxxsearch目录，编辑instance.properties文件，主要配置源数据库信息、所需数据表及字段，以及指定kafka的topic名，这样源数据库的binlog就会转换为json数据，并实时的通过canal deployer传输到kafka该topic中。如下所示：
canaldeploy instance 源数据库配置
canaldeploy instance kafka topic配置
③进入canaldeployer/bin目录，执行./startup.sh，启动canal deployer及所属实例。
至此canal deployer搭建完成。
2.2 安装canal.adapter
我们需要利用canal.adapter将kafka topic中的binlog json数据，经过清洗转换等操作，存储到MySQL8中。由于canal原生是不支持MySQL8的，故我们需要做一些调整。
（1）增加MySQL8连接驱动
解压canal.adapter-1.1.4.tar.gz，进入canaladapter/lib目录，移除mysql-connector-java-5.1.40.jar，导入mysql-connector-java-8.0.18.jar
（2）配置canal adapter，使数据输出到MySQL8。
进入canaladapter/conf目录，编辑application.yml文件，主要配置消费kafka、源数据库信息和搜索系统数据库信息，如下所示：
ETL到MySQL8配置
接着，进入canaladapter/conf/rdb目录，以官方提供的mytest_user.yml为例，配置kafka topic名、源数据库名、源数据表名，以及目标数据库名和目标数据表名，
建议一张表对应一个yml文件。
ETL表结构映射配置
（3）启动canaladapter
进入canaladapter/bin目录，执行./startup.sh，启动canal adapter，观察logs/adapter/adapter.log日志文件，手动在搜索系统数据库新增一条记录，看是否会打印如下日志，即有2条记录，一条INFO，一条DEBUG，则表示配置成功。
canaladapter日志
至此，数据ETL阶段搭建完成，数据可从两个不同的MySQL8数据库，实时同步到搜索服务的MySQL数据库。
3、实现数据多表关联、父子文档映射
（1）配置第二个canal的canaladapter
进入canaladapter/conf目录，编辑application.yml文件，主要配置消费kafka、搜索系统数据库，和ES连接信息，如下所示：
canaladapter MQ及mysql配置
canaladapter ES配置
（2）配置多表关联
进入canaladapter/conf/es目录，vim mytest_user.yml，编辑多表关联配置：
多表关联配置
注意，sql支持多表关联自由组合, 但是有一定的限制:
（a）主表不能为子查询语句
（b）只能使用left outer join即最左表一定要是主表
（c）关联从表如果是子查询不能有多张表
（d）主sql中不能有where查询条件(从表子查询中可以有where条件但是不推荐, 可能会造成数据同步的不一致, 比如修改了where条件中的字段内容)
（e）关联条件只允许主外键的"="操作不能出现其他常量判断比如: on a.role_id=b.id and b.statues=1
（f）关联条件必须要有一个字段出现在主查询语句中比如: on a.role_id=b.id 其中的 a.role_id 或者 b.id 必须出现在主select语句中
（g）Elastic Search的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映射.
（3）配置父子文档
以官方的biz_order.yml为例，vim biz_order.yml，配置父子文档映射：
配置父子文档映射
（4）在ElasticSearch6中，建立index和父子文档映射关系
进入kibana页面，点击Dev Tools，执行如下命令，即可建立索引及父子文档映射：
建立index和父子文档映射
其中，ES6和kibana的安装，在此无特别配置，不做赘述。
（5）启动canal adapter
进入canaladapter/bin目录，执行./startup.sh，启动canal adapter，观察logs/adapter/adapter.log日志文件，手动在搜索系统数据库新增一条记录，看是否会打印如下日志，如打印则表示配置成功。
正确配置adapter日志示例
4、运行结果
现在，我们可以通过kibana来执行DSL语句来查询看看。
我们事先已在商家系统中增加了一个“肯德基”商店，然后在商品系统中添加了“西红柿”和”新鲜西红柿“2个商品，并将商品关联到“肯德基”上。接着我们查询”肯德基“或者“西红柿”，得到以下是查询的结果（去除了ES默认字段）：
通过DSL查询的结果
由图可见，我们可以通过商家名查询商品，也可通过商品名查询商店和商品，并且canal支持数据的实时增删改，所以ES的数据也会与商家系统和商品系统保持一致，同时数据结构包含商家及对应的商品，满足业务需求。
5、总结
至此，基于Canal、kafka、MySQL8、ElasticSearch6技术的商家商品搜索系统基础框架搭建完成。我们采用canal deployer实时读取商家、商品系统的MySQL数据库binlog，并发送至kafka，接着由canal adapter消费kafka，并将binlog json数据进行多表关联、父子文档映射，最后存储到ES6中，供上层搜索服务调用。
搜索服务系统最终成功上线，为公司百万级商家商品提供实时数据同步，秒级搜索结果展示，达到业务要求，老板说了，给研发团队每人加个鸡腿！想想还有点小激动，嘿嘿~~
本文作者：Kevin，微微科技资深Java工程师，关注“微微科技公司”，获取更多技术干货
以上是 百万级商品数据实时同步，查询结果秒出 的全部内容， 来源链接： utcz.com/z/532613.html
