canal全量同步到ES

 参考文档：https://blog.csdn.net/zlt2000/article/details/115291950

一、ETL接口

adapter 的 ETL 接口为：/etl/{type}/{task}

• 默认web端口为 8081
• type 为类型(hbase/es7/rdb)
• task 为任务名对应配置文件名，如sys_user.yml

curl -X POST http://127.0.0.1:8081/etl/es7/sys_user.yml

二、同步错误

示例1

 当同步的数据量比较大时，执行一段时间后会出现上图的错误

原因分析

查看 canal 源码得知当同步的数据量大于1w时，会分批进行同步，每批1w条记录，并使用多线程来并行执行任务，而 adapter 默认的连接池为3，当线程获取数据库连接等待超过1分钟就会抛出该异常。

 解决方式

修改 adapter 的 conf/application.yml 文件中的 srcDataSources 配置项，增加 maxActive 配置数据库的最大连接数为当前服务器cpu的可用线程数

cpu线程数可以下命令查看

grep 'processor' /proc/cpuinfo | sort -u | wc -l

二、es连接超时

原因分析

由于 adapter 的表映射配置文件中的 commitBatch 提交批大小设置过大导致(6000)

解决方式

修改 adapter 的 conf/es7/xxx.yml 映射文件中的 commitBatch 配置项为3000

三、同步慢，丢数据问题

三千万的数据量用时3.5小时左右

原因分析

由于当数据量大于1w时 canal 会对数据进行分批同步，每批1w条通过分页查询实现；所以当数据量较大时会出现深分页的情况导致查询非常慢。

解决方式

预先使用ID、时间或者业务字段等进行数据分批后再进行同步，减少每次同步的数据量。

案例
使用ID进行数据分批，适合增长类型的ID，如自增ID、雪花ID等；

查出 最小ID、最大ID 与 总数据量
根据每批数据量大小计算每批的 ID区间

计算过程：

最小ID = 1333224842416979257
最大ID = 1341698897306914816
总数据量 = 3kw
每次同步量 = 300w

使用分批的ID值进行同步

修改sql映射配置，的 etlCondition 参数：

etlCondition: "where id >= {} and id < {}"

 etlCondition: "where a.topic_msg_id>={} and a.topic_msg_id<{}"

调用etl接口，并增加 params 参数，多个参数之间使用 ; 分割

curl localhost:8083/etl/es7/topic_analysis_msgl.yml -XPOST -d "params=1262193776;1262193777"   # 前面是最小id，最大id减去最小id为一次同步的量
curl localhost:8084/etl/es7/topic_analysis_msgl.yml -X POST -d "params=1773734832;1780338515"