canal同步mysql数据至es5.5.0
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了canal同步mysql数据至es5.5.0，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含7655字，纯文字阅读大概需要11分钟。
内容图文
系统环境：
- canal-1.1.4
- es 5.5.0
- transport方式连接es
各项配置可以直接参考canal官方文档，由于1.1.4支持的es版本为6.x以上，其他版本需要替换依赖重新编译client-adapter.elasticsearch模块，以下为es5.5.0低版本兼容方案以及个人踩的坑。
依赖修改：
修改client-adapter模块的pom.xml，将es的依赖修改为es版本适配的5.5.0。
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>5.5.0</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>5.5.0</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
<version>6.4.3</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>6.4.3</version>
</dependency>
由于5.5.0版本无rest-client，因此只修改transport相关版本，后续仅测试tcp连接es同步，rest不确定兼容性。
代码兼容：
ESConnection.java:
transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)),
Integer.parseInt(host.substring(i + 1))));
修改为
transportClient.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host.substring(0, i)),
Integer.parseInt(host.substring(i + 1))));
开始编译
mvn clean install -Dmaven.test.skip -Denv=release
rest兼容问题
问题1
[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.7.0:compile (default-compile) on project client-adapter.elasticsearch: Compilation failure: Compilation failure:
[ERROR] canal/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java:[502,75] 未报告的异常错误java.io.IOException; 必须对其进行捕获或声明以便抛出
[ERROR] canal/client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RestHighLevelClientExt.java:[24,13] 方法引用无效
5.x版本的transportclient不兼容rest-client，注释掉rest导致的异常。
RestHighLevelClientExt::getMapping
@Deprecated
public static GetMappingsResponse getMapping(RestHighLevelClient restHighLevelClient,
GetMappingsRequest getMappingsRequest,
RequestOptions options) throws IOException,IllegalAccessException {
throw new IllegalAccessException("es 5.x unsupport this method, use tcp mode");
}
ESConnection::getMapping
...
if (mode == ESClientMode.TRANSPORT) {
...
} else {
try {
GetMappingsRequest request = new GetMappingsRequest();
request.indices(index);
GetMappingsResponse response;
// try {
// response = restHighLevelClient
// .indices()
// .getMapping(request, RequestOptions.DEFAULT);
// // 6.4以下版本直接使用该接口会报错
// } catch (Exception e) {
// logger.warn("Low ElasticSearch version for getMapping");
response = RestHighLevelClientExt.getMapping(restHighLevelClient, request, RequestOptions.DEFAULT);
// }
mappings = response.mappings();
} catch (NullPointerException e) {
throw new IllegalArgumentException("Not found the mapping info of index: " + index);
} catch (IOException | IllegalAccessException e) {//此处增加一个异常捕获
logger.error(e.getMessage(), e);
return null;
}
...
}
问题2
[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.7.0:compile (default-compile) on project client-adapter.elasticsearch: Compilation failure
[ERROR] canal/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java:[502,75] 未报告的异常错误java.io.IOException; 必须对其进行捕获或声明以便抛出
原因如下，getSourceAsMap方法在6.4.3抛出runtimeException（ElasticsearchParseException是子类），而5.5.0版本抛出IOException,需要显示捕获。
//6.4.3抛出的异常时runtimeException
public Map<String, Object> getSourceAsMap() throws ElasticsearchParseException {
return this.sourceAsMap();
}
//5.5.0版本
public Map<String, Object> getSourceAsMap() throws IOException {
return sourceAsMap();
}
修改ESTemplate的getEsType方法捕获异常即可
ESTemplate::getEsType
Map<String, Object> sourceMap = null;
try{
sourceMap = mappingMetaData.getSourceAsMap();
}catch (IOException e){
logger.error(e.getMessage(), e);
return null;
}
编译后，替换canal.adapter-1.1.4\plugin下的 client-adapter.elasticsearch-1.1.4-jar-with-dependencies.jar 文件。
执行deploy和adapter启动脚本即可。
配置问题
启动后报错：
2020-07-07 14:36:08.223 [main] INFO org.elasticsearch.plugins.PluginsService - loaded plugin [org.elasticsearch.transport.Netty4Plugin]
2020-07-07 14:36:08.473 [main] ERROR c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: es failed
java.lang.RuntimeException: java.lang.IllegalArgumentException: unknown setting [mode] please check that any required plugins are installed, or check the breaking changes documentation for removed settings
at com.alibaba.otter.canal.client.adapter.es.ESAdapter.init(ESAdapter.java:137)
at com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterLoader.loadAdapter(CanalAdapterLoader.java:172)
查看canal源码，未发现抛出异常日志的代码，再搜索依赖的包，发现异常是es创建transportClient时抛出的异常，于是猜测是canal-adpapter配置中的某个mode参数被引入创建transportClient的setting中导致创建失败，于是注释掉，并重启。
- name: es
hosts: 127.0.0.1:9300 # 127.0.0.1:9200 for rest mode
properties:
#mode: transport # transport # or rest //注释了这行，是1.1.4的坑，代码中properties下的所有配置都会被传入transportClient的setting中，rest模式则不会，所以transport模式除了cluster.name外的配置会导致es连接创建失败
# security.auth: test:123456 # only used for rest mode
cluster.name: elasticsearch
测试：
重启后，向mysql插入数据后，adapter打印出日志
[pool-2-thread-1] INFO c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":21,"name":"测试用户","time":null}],"database":"canal","destination":"example_instance","es":1594347777000,"groupId":null,"isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"class","ts":1594347777991,"type":"INSERT"}
[pool-2-thread-1] TRACE c.a.otter.canal.client.adapter.es.service.ESSyncService - Destination: example_instance, database:canal, table:class, type:INSERT, affected index count: 1
[pool-2-thread-1] TRACE c.a.otter.canal.client.adapter.es.service.ESSyncService - Prepared to sync index: canal_test, destination: example_instance
[pool-2-thread-1] TRACE c.a.otter.canal.client.adapter.es.service.ESSyncService - Single table insert to es index, destination:example_instance, table: class, index: canal_test, id: 21
[pool-2-thread-1] TRACE c.a.otter.canal.client.adapter.es.service.ESSyncService - Sync elapsed time: 1 ms,destination: example_instance, es index: canal_test
[pool-2-thread-1] TRACE c.a.otter.canal.client.adapter.es.service.ESSyncService - Sync completed: canal_test, destination: example_instance
[pool-2-thread-1] TRACE c.a.otter.canal.client.adapter.es.service.ESSyncService - Sync elapsed time: 2 ms, affected indexes count：1, destination: example_instance
[pool-2-thread-1] DEBUG c.a.otter.canal.client.adapter.es.service.ESSyncService - DML: {"data":[{"id":21,"name":"测试用户","time":null}],"database":"canal","destination":"example_instance","es":1594347777000,"groupId":null,"isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"class","ts":1594347777993,"type":"INSERT"}
Affected indexes: canal_test
查看es数据
curl 127.0.0.1:9200/canal_test/canal/21
{
"_index": "canal_test",
"_type": "canal",
"_id": "21",
"_version": 1,
"found": true,
"_source": {
"name": "测试用户"
}
}
小结：
- canal-adapter不支持索引名，若有频繁全量构建需求则不适用该方案
- 更新时查询不支持非数字类型主键（拼接SQL字符串导致）
- 表的更新都会同步至es，一对多关联时，记录变更可能会触发索引批量更新，索引若存储快照数据则建议监听变更开发带业务逻辑的adapter
相关文档：
MySQL实时同步到Elasticsearch实现方案
canal官方文档
内容总结
以上是互联网集市为您收集整理的canal同步mysql数据至es5.5.0全部内容，希望文章能够帮你解决canal同步mysql数据至es5.5.0所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】