如何设计百万级商品数据实时同步的秒级搜索系统？

2023-02-27

数据数据库商品

前阵子老板安排了一个新任务，要建设一个商家商品搜索系统，能够为用户提供快速、准确的搜索能力。图片来自Pexels设计要求在用户输入搜索内容时，要能从商家名称和商品名称两个维度去搜索，搜索出来的结果，按照准确率排序，并按商家所属商品的关联关系，来组合数据结构，同时提供API给业务系统调用。背景很简单，

前阵子老板安排了一个新任务，要建设一个商家商品搜索系统，能够为用户提供快速、准确的搜索能力。

图片来自 Pexels

设计要求在用户输入搜索内容时，要能从商家名称和商品名称两个维度去搜索，搜索出来的结果，按照准确率排序，并按商家所属商品的关联关系，来组合数据结构，同时提供 API 给业务系统调用。

背景很简单，现实蛮复杂!我们面临以下几个难题：

商家数据库和商品数据库是多台不同的服务器，并且数据量达百万级，如何才能实现跨数据库的数据同步呢?
商家和商品的数据是有从属关系的，不然就会把肯德基的香辣鸡腿堡挂到麦当劳去，这就尴尬了!
商家商品数据是经常更新的，比如修改价格、库存、上下架等，那搜索服务可不能搜出一堆过时的数据，如果客户明明搜出来的商品，点进去后却已下架了，那么客户就要吐槽了!如何实现搜索数据与源数据库增删改均实时同步呢?

带着以上三个问题，我们开始了搜索服务的整体架构设计。

系统架构设计思路

为了设计出合适的系统架构，我们分析了现状：

首先，商家数据和商品数据分别存储在 2 个独立的 MySQL8 数据库，为满足商家数据和商品数据的关联，我们需要将两个库中所需要的表实时 ETL 到我们的搜索系统数据库。

其次，数据从商家、商品数据库 ETL 到搜索系统数据库后，需要实时的组合成为商家关联商品数据结构，并以父子文档的格式，存储到 ES 中。

最后，商家、商品数据库的增删改操作，需要实时的同步到 ES 中，也就是 ES 中的数据，需要支持实时的增加、删除和修改。

为此，我们设计了 2 个 Canal 组件，第一个 Canal 实现数据 ETL，把商家、商品数据库的某些表及字段，抽取到搜索服务数据库。

再利用第二个 Canal，读取搜索服务 MySQL 数据库的 Binlog，实时传输到 Kafka 消息队列，再由 canal adapter 对数据进行关联、父子文档映射等，将处理好的数据存储到 ElasticSearch 中。

具体系统架构设计如下图所示：

商家商品搜索系统架构设计

项目实战

环境及软件说明

操作系统：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 
1.
2.
3.
4.
5.

用 Canal 实现数据 ETL 到 MySQL8

这个步骤是利用 Canal 从 2 个独立的 MySQL8 数据库中，抽取需要的表到搜索服务的 MySQL 数据库。

①安装 canaldeployer

解压 canal.deployer-1.1.4.tar.gz，并配置 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 实例配置

配置 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 搭建完成。

②安装 canal.adapter

我们需要利用 canal.adapter 将 Kafka Topic 中的 binlog json 数据，经过清洗转换等操作，存储到 MySQL8 中。由于 Canal 原生是不支持 MySQL8 的，故我们需要做一些调整。

增加 MySQL8 连接驱动：解压 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，使数据输出到 MySQL8：进入 canaladapter/conf 目录，编辑 application.yml 文件，主要配置消费 Kafka、源数据库信息和搜索系统数据库信息。

如下所示：

ETL 到 MySQL8 配置

接着，进入 canaladapter/conf/rdb 目录，以官方提供的 mytest_user.yml 为例，配置 Kafka Topic 名、源数据库名、源数据表名，以及目标数据库名和目标数据表名，建议一张表对应一个 yml 文件。

ETL 表结构映射配置

启动 canaladapter：进入 canaladapter/bin 目录，执行 ./startup.sh，启动 canal adapter，观察 logs/adapter/adapter.log 日志文件，手动在搜索系统数据库新增一条记录，看是否会打印如下日志，即有 2 条记录，一条 INFO，一条 DEBUG，则表示配置成功。

canaladapter 日志

至此，数据 ETL 阶段搭建完成，数据可从两个不同的 MySQL8 数据库，实时同步到搜索服务的 MySQL 数据库。

实现数据多表关联、父子文档映射

①配置第二个 Canal 的 canaladapter

进入 canaladapter/conf 目录，编辑 application.yml 文件，主要配置消费 Kafka、搜索系统数据库，和 ES 连接信息。

如下所示：

canaladapter MQ 及 MySQL 配置

canaladapter ES 配置

②配置多表关联

进入 canaladapter/conf/es 目录，vim mytest_user.yml，编辑多表关联配置：

多表关联配置

注意，sql支持多表关联自由组合, 但是有一定的限制：

主表不能为子查询语句。
只能使用 left outer join 即最左表一定要是主表。
关联从表如果是子查询不能有多张表。
主 sql 中不能有 where 查询条件(从表子查询中可以有 where 条件但是不推荐, 可能会造成数据同步的不一致，比如修改了 where 条件中的字段内容)。
关联条件只允许主外键的'='操作不能出现其他常量判断比如：on a.role_id=b.id and b.statues=1。
关联条件必须要有一个字段出现在主查询语句中比如：on a.role_id=b.id 其中的 a.role_id 或者 b.id 必须出现在主 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，配置父子文档映射：