使用canal实现MySQL 8 增量同步数据到 ElasticSearch 7.15.2中 linux

文章目录

一、清空控制台
- 1. 清空控制台
- 2. 修改数据
- 3. 监控数据
- 4. 数据变化
- 5. 索引查询
- 6. 预期性能评估
- 7. 增量同步分析
二、验证方案
- 2.1. 把shop索引删除
- 2.2. 重建shop索引
- 2.3. 修改id=取得数据
- 2.4. 查看shop索引数据
- 2.4. 修改分类表数据
- 2.6. 监控
- 2.7. 查看shop索引数据
- 2.8. 总结
- 2.9. 解决方案

一、清空控制台

1. 清空控制台

清空canal.adapter控制台

cd /app/canal/canal.adapter
tail -f logs/adapter/adapter.log

2. 修改数据

修改mysql的id=1的name的数据，将陕西22面馆(北京亦庄
改为gblfy陕西面馆(北京亦庄)

3. 监控数据

监控canal.adapter控制台

4. 数据变化

{"data": [{"id": 1,"created_at": 1637308432000,"updated_at": 1640159632000,"name": "gblfy陕西面馆(北京亦庄)","remark_score": 4.9,"price_per_man": 156,"latitude": 31.195341,"longitude": 120.915855,"category_id": 1,"tags": "新开业 人气爆棚","start_time": "10:00","end_time": "22:00","address": "船厂路36号","seller_id": 1,"icon_url": "/static/image/shopcover/xchg.jpg"}],"database": "dianpingdb","destination": "example","es": 1637644424000,"groupId": "g1","isDdl": false,"old": [{"name": "陕西22面馆(北京亦庄)"}],"pkNames": ["id"],"sql": "","table": "shop","ts": 1637644425043,"type": "UPDATE"
}
Affected indexes: shop

从上面可以看出，canal.adapter监控到了我的数据变化，并进行增量的数据的同步到ES7中。

5. 索引查询

# 查询shop索引
GET /shop/_search
{"query": {"match": {"name": "gblfy"}}
}

6. 预期性能评估

我们要求最大性能（理论上）：当用户修改id=1 中name的数据的值，其他索引的内容都不应该发生变化，仅仅对我对修改的id=1 中name的数据的值发生变化。
我们在配置canal.adapter的shop.xml中的sql时，是一个全量的sql。

SELECTa.id,a.NAME,a.tags,CONCAT( a.latitude, ',', a.longitude ) AS location,a.remark_score,a.price_per_man,a.category_id,b.`name` AS category_name,a.seller_id,c.remark_score AS seller_remark_score,c.disabled_flag AS seller_disabled_flag
FROMshop aINNER JOIN category b ON a.category_id = b.idINNER JOIN seller c ON c.id = a.seller_id

7. 增量同步分析

sql中没有指定具体要求的数据的id，那canal.adapter怎么监控到我数据发生变化后，只同步我修改ID=1的这条数据呢？

二种解释：
第一种：
canal.adapter内部做了一个这样的机制；
当我修改ID=1的数据时，canal.adapter通过全量sql就可以把id=1的数据筛选出来，然后进行增量同步到es7的索引中。
第二种：
canal.adapter 内部武这样监控的机制，只要数据发生变化，就执行全量sql，做全量同步，只不过已经存在而索引让es7做更新操作而已。

二、验证方案

2.1. 把shop索引删除

# 删除shop索引
DELETE  /shop

2.2. 重建shop索引

PUT /shop
{"settings": {"number_of_shards": 1,"number_of_replicas": 1}, "mappings": {"properties": {"id":{"type": "integer"},"name":{"type": "text","analyzer": "ik_max_word","search_analyzer": "ik_smart"},"tags":{"type": "text","analyzer": "whitespace","fielddata": true},"location":{"type": "geo_point"},"remark_score":{"type": "double"},"price_per_man":{"type": "integer"},"category_id":{"type": "integer"},"category_name":{"type": "keyword"},"seller_id":{"type": "integer"},"seller_remark_score":{"type": "double"},"seller_disabled_flag":{"type": "integer"}}}
}

2.3. 修改id=取得数据

把id=1的name值由gblfy陕西面馆(北京亦庄)改成com.gblfy陕西面馆(北京亦庄)

2.4. 查看shop索引数据

# 查询shop索引
GET /shop/_search

从上面查询结果来看符合预期，对吧！

2.4. 修改分类表数据

把category表中id=1中的name值美食5修改成美食8

说明：

2.6. 监控

从截图中可以看出也监控到了，对吧，别急！

2.7. 查看shop索引数据

从上面截图中可以看到，数据canal.adapter 虽然监控到了，但是，es7中的数据没更新。

2.8. 总结

官网自带的canal.adapter 组件
canal.adapter 对单表增量同步，是支持的
一对一形式：mysql的一张表对应es的一个index
canal.adapter 对聚合sql（也就是多表查询），不支持

2.9. 解决方案

我们可以通过代码方式，可以自定义我们对于canal.adapter 的行为的能力。使用canal.deployer建立和mysql数据binlog之间的能力，摒弃掉canal.adapter，在点评搜索的应用当中，直接接入canal.deployer，以canal.deployer消费端的身份去消费我们对应的消息。