使用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

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"
}

从上面可以看出，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。

SELECT
  a.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 
FROM
  shop a
  INNER JOIN category b ON a.category_id = b.id
  INNER JOIN seller c ON c.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索引

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索引

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

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消费端的身份去消费我们对应的消息。