https://githubfast.com/alibaba/canal/wiki/QuickStart
https://githubfast.com/alibaba/canal/wiki/Sync-ES
準(zhǔn)備工作
- mysql版本 5.7
- es 版本 7.1.0 （這里用7.X版本都行沒有小版本要求）
- canal-deployer 版本 1.1.7
- cannal-adapter 版本 1.1.5（我本來也是用1.1.7的誰知道最新版有問題）
- mysql表結(jié)構(gòu)
CREATE TABLE `book` (
`id` bigint NOT NULL AUTO_INCREMENT,
`name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
`year` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
`last_updated` bigint DEFAULT NULL,
`name2` varchar(255) DEFAULT NULL,
`name3` varchar(255) DEFAULT NULL,
`name4` varchar(255) DEFAULT NULL,
`name5` varchar(255) DEFAULT NULL,
`name6` tinyint DEFAULT NULL,
`create_time` datetime DEFAULT NULL,
`my_json` json DEFAULT NULL,
PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=18 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci ROW_FORMAT=DYNAMIC
- 數(shù)據(jù)也給一份，到時(shí)候直接測(cè)試
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (1, '1234', '1', 1695636893723, '1', NULL, NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (2, '1', '1', 1695636893724, '1', NULL, NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (3, '世界ol111', '1988', 1695636893734, '2', NULL, NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (4, '三國(guó)演義41', '1987', 1695651267677, '1', NULL, NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (5, '三國(guó)演義41', '1987', 1695651267677, '2', NULL, NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (6, '三國(guó)演義41', '1987', 1695651267677, '4', NULL, NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (7, '三國(guó)演義41', '1987', 1695651267677, '4', '你好', NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (8, '三國(guó)演義411', '1987', 1695651267677, '4', '你好21', NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (9, '三國(guó)演義412', '1987', 1695651267677, '4', '你好2', NULL, NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (10, '三國(guó)演義41', '1987', 1695651267677, '4', '你好2', '你好3 ', NULL, NULL, NULL, NULL);
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (11, '三國(guó)演義411', '1987', 1695651267677, '4', '你好2', NULL, NULL, NULL, '2023-09-27 15:46:33', '{\"age\": 11, \"name\": \"yin1kai\"}');
INSERT INTO `db_example`.`book` (`id`, `name`, `year`, `last_updated`, `name2`, `name3`, `name4`, `name5`, `name6`, `create_time`, `my_json`) VALUES (16, '三國(guó)演義411', '1987', 1695651267677, '4', '你好2', '世界', '哈1', NULL, '2023-09-27 15:46:33', '{\"age\": 11, \"name\": \"yin1kai\"}');
開啟mysql的binlog
[mysqld]
log-bin=mysql-bin # 開啟 binlog
binlog-format=ROW # 選擇 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定義，不要和 canal 的 slaveId 重復(fù)
建立單獨(dú)的mysql用戶（也可以直接使用root）
-- 創(chuàng)建用戶 用戶名：canal 密碼：Canal@123456
create user 'canal'@'%' identified by 'Canal@123456';
-- 授權(quán) *.*表示所有庫(kù)
grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%';
FLUSH PRIVILEGES;
deployer配置
canal.deployer-1.1.7-SNAPSHOT\conf\canal.properties
- canal.serverMode = tcp 表示deployer使用tcp協(xié)議直連adapter
#################################################
######### common argument #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
# canal admin config
#canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
#canal.admin.register.auto = true
#canal.admin.register.cluster =
#canal.admin.register.name =
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024
## meory store gets mode used MEMSIZE or ITEMSIZE
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true
## detecing config
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false
# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
canal.instance.transaction.size = 1024
# mysql fallback connected to new master should fallback times
canal.instance.fallbackIntervalInSeconds = 60
# network config
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30
# binlog filter config
canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false
canal.instance.filter.dml.insert = false
canal.instance.filter.dml.update = false
canal.instance.filter.dml.delete = false
# binlog format/image check
canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
# binlog ddl isolation
canal.instance.get.ddl.isolation = false
# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256
# table meta tsdb info
canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360
#################################################
######### destinations #############
#################################################
canal.destinations = example
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5
# set this value to 'true' means that when binlog pos not found, skip to latest.
# WARN: pls keep 'false' in production env, or if you know what you want.
canal.auto.reset.latest.pos.mode = false
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
canal.instance.global.mode = spring
canal.instance.global.lazy = false
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
canal.instance.global.spring.xml = classpath:spring/file-instance.xml
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml
##################################################
######### MQ Properties #############
##################################################
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
canal.aliyun.uid=
canal.mq.flatMessage = true
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local
canal.mq.database.hash = true
canal.mq.send.thread.size = 30
canal.mq.build.thread.size = 8
##################################################
######### Kafka #############
##################################################
kafka.bootstrap.servers = 127.0.0.1:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0
kafka.kerberos.enable = false
kafka.kerberos.krb5.file = ../conf/kerberos/krb5.conf
kafka.kerberos.jaas.file = ../conf/kerberos/jaas.conf
# sasl demo
# kafka.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \\n username=\"alice\" \\npassword="alice-secret\";
# kafka.sasl.mechanism = SCRAM-SHA-512
# kafka.security.protocol = SASL_PLAINTEXT
##################################################
######### RocketMQ #############
##################################################
rocketmq.producer.group = test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag =
##################################################
######### RabbitMQ #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =
rabbitmq.deliveryMode =
##################################################
######### Pulsar #############
##################################################
pulsarmq.serverUrl =
pulsarmq.roleToken =
pulsarmq.topicTenantPrefix =
conf\example\instance.properties
#################################################
## mysql serverId , v1.0.26+ will autoGen
## mysql serverId，這里的slaveId不能和myql集群中已有的server_id一樣
canal.instance.mysql.slaveId=1234
# enable gtid use true/false
canal.instance.gtidon=false
# position info
canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=
# mysql主庫(kù)鏈接時(shí)起始的binlog偏移量
canal.instance.master.position=
# mysql主庫(kù)鏈接時(shí)起始的binlog的時(shí)間戳
canal.instance.master.timestamp=
canal.instance.master.gtid=
# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=
# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=Canal@123456
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
# table regex
# table regex .*\..*表示監(jiān)聽所有表 也可以寫具體的表名，用，隔開
canal.instance.filter.regex=.*\\..*
# table black regex
# mysql 數(shù)據(jù)解析表的黑名單，多個(gè)表用，隔開
canal.instance.filter.black.regex=
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
# mq config
canal.mq.topic=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.enableDynamicQueuePartition=false
#canal.mq.partitionsNum=3
#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#
# multi stream for polardbx
canal.instance.multi.stream.on=false
#################################################
adapter 配置
adapter的application.yml
server:
port: 8081
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
default-property-inclusion: non_null
canal.conf:
mode: tcp #tcp kafka rocketMQ rabbitMQ
flatMessage: true
zookeeperHosts:
syncBatchSize: 1000
retries: 0
timeout:
accessKey:
secretKey:
consumerProperties:
# canal tcp consumer
canal.tcp.server.host: 127.0.0.1:11111
canal.tcp.zookeeper.hosts:
canal.tcp.batch.size: 500
canal.tcp.username:
canal.tcp.password:
# kafka consumer
kafka.bootstrap.servers:
kafka.enable.auto.commit: false
kafka.auto.commit.interval.ms:
kafka.auto.offset.reset:
kafka.request.timeout.ms:
kafka.session.timeout.ms:
kafka.isolation.level:
kafka.max.poll.records:
# rocketMQ consumer
rocketmq.namespace:
rocketmq.namesrv.addr:
rocketmq.batch.size:
rocketmq.enable.message.trace:
rocketmq.customized.trace.topic:
rocketmq.access.channel:
rocketmq.subscribe.filter:
# rabbitMQ consumer
rabbitmq.host:
rabbitmq.virtual.host:
rabbitmq.username:
rabbitmq.password:
rabbitmq.resource.ownerId:
srcDataSources:
defaultDS:
url: jdbc:mysql://127.0.0.1:3306/db_example?useUnicode=true
username: root
password: 123456
canalAdapters:
- instance: example # canal instance Name or mq topic name
groups:
- groupId: g1
outerAdapters:
- name: logger
# - name: rdb
# key: mysql1
# properties:
# jdbc.driverClassName: com.mysql.jdbc.Driver
# jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true
# jdbc.username: root
# jdbc.password: 121212
# - name: rdb
# key: oracle1
# properties:
# jdbc.driverClassName: oracle.jdbc.OracleDriver
# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE
# jdbc.username: mytest
# jdbc.password: m121212
# - name: rdb
# key: postgres1
# properties:
# jdbc.driverClassName: org.postgresql.Driver
# jdbc.url: jdbc:postgresql://localhost:5432/postgres
# jdbc.username: postgres
# jdbc.password: 121212
# threads: 1
# commitSize: 3000
# - name: hbase
# properties:
# hbase.zookeeper.quorum: 127.0.0.1
# hbase.zookeeper.property.clientPort: 2181
# zookeeper.znode.parent: /hbase
- name: es7
key: exampleKey
hosts: http://127.0.0.1:9200 # 127.0.0.1:9200 for rest mode
properties:
mode: rest # or rest
# security.auth: test:123456 # only used for rest mode
cluster.name: geektime
# - name: kudu
# key: kudu
# properties:
# kudu.master.address: 127.0.0.1 # ',' split multi address
具體同步表的配置
my.canal.adapter-1.1.5\conf\es7\book.yml
dataSourceKey: defaultDS # 源數(shù)據(jù)源的key, 對(duì)應(yīng)上面配置的srcDataSources中的值
outerAdapterKey: exampleKey # 對(duì)應(yīng)application.yml中es配置的key
destination: example # cannal的instance或者M(jìn)Q的topic
groupId: # 對(duì)應(yīng)MQ模式下的groupId, 只會(huì)同步對(duì)應(yīng)groupId的數(shù)據(jù)
esMapping:
_index: book # es 的索引名稱
_type: _doc # es 的type名稱, es7下無需配置此項(xiàng)
_id: _id # es 的_id, 如果不配置該項(xiàng)必須配置下面的pk項(xiàng)_id則會(huì)由es自動(dòng)分配
# pk: id # 如果不需要_id, 則需要指定一個(gè)屬性為主鍵屬性
# sql映射
sql: "select a.id as _id, a.name, a.year, a.last_updated,a.name2,a.name3,a.name4,a.name5,a.name6,date_format(a.create_time,'%Y-%m-%d %H:%I:%S') as create_time,a.my_json from book a"
# objFields:
# _labels: array:; # 數(shù)組或者對(duì)象屬性, array:; 代表以;字段里面是以;分隔的
# _obj: object # json對(duì)象
etlCondition: "where last_updated>='{0}'" # etl 的條件參數(shù)
commitBatch: 3000 # 提交批大小
問題
1、Illegal character in scheme name at index 0: 127.0.0.1:9200
rest方式加http
2、Caused by: org.springframework.boot.context.properties.bind.BindException: Failed to bind properties under 'es-mapping' to com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig$ESMapping
原因：
但是從github上下載了最新的canal1.1.6，該版本并不能兼容java8，只能用低版本的canal1.1.5
3、ClassCastException: com.alibaba.druid.pool.DruidDataSource cannot be cast to com.alibaba.druid.pool.DruidDataSource** ）
原因：
druid版本沖突，解決思路：下載canal源碼包，更改maven配置，本地打包后替換jar包即可
編輯后替換client-adapter.es7x-1.1.5-jar-with-dependencies.jar
4、又一個(gè)空指針
這個(gè)問題是app.yml的配置錯(cuò)誤
2023-09-29 11:48:02.003 [main] ERROR c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## something goes wrong when starting up the canal client adapters:
java.lang.NullPointerException: null
at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.<init>(AdapterProcessor.java:69) ~[client-adapter.launcher-1.1.5.jar:na]
at com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterLoader.lambda$init$0(CanalAdapterLoader.java:65) ~[client-adapter.launcher-1.1.5.jar:na]
at java.util.HashMap.computeIfAbsent(HashMap.java:1128) ~[na:1.8.0_382]
5、又是一個(gè)錯(cuò)誤。。
先啟動(dòng)canal.deployer 后啟動(dòng)canal.adapter就能解決
... 4 common frames omitted
2023-09-29 12:08:43.235 [Thread-4] INFO c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Start to connect destination: example <=============
2023-09-29 12:08:45.247 [Thread-4] ERROR c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - process error!
com.alibaba.otter.canal.protocol.exception.CanalClientException: java.net.ConnectException: Connection refused: connect
at com.alibaba.otter.canal.client.impl.SimpleCanalConnector.doConnect(SimpleCanalConnector.java:198) ~[na:na]
at com.alibaba.otter.canal.client.impl.SimpleCanalConnector.connect(SimpleCanalConnector.java:115) ~[na:na]
at com.alibaba.otter.canal.connector.tcp.consumer.CanalTCPConsumer.connect(CanalTCPConsumer.java:63) ~[na:na]
at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.process(AdapterProcessor.java:184) ~[client-adapter.launcher-1.1.5.jar:na]
at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_382]
Caused by: java.net.ConnectException: Connection refused: connect
at sun.nio.ch.Net.connect0(Native Method) ~[na:1.8.0_382]
at sun.nio.ch.Net.connect(Net.java:482) ~[na:1.8.0_382]
at sun.nio.ch.Net.connect(Net.java:474) ~[na:1.8.0_382]
6、明顯就是一個(gè)索引不存在就去操作數(shù)據(jù)的問題，是不是一定要手動(dòng)創(chuàng)建es的index、mappings
java.lang.RuntimeException: [book] ElasticsearchStatusException[Elasticsearch exception [type=index_not_found_exception, reason=no such index [book]]]
at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.sync(ESSyncService.java:116) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.sync(ESSyncService.java:64) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es.core.ESAdapter.sync(ESAdapter.java:115) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es.core.ESAdapter.sync(ESAdapter.java:94) ~[na:na]
at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.batchSync(AdapterProcessor.java:139) ~[client-adapter.launcher-1.1.5.jar:na]
at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.lambda$null$1(AdapterProcessor.java:97) ~[client-adapter.launcher-1.1.5.jar:na]
at java.util.concurrent.CopyOnWriteArrayList.forEach(CopyOnWriteArrayList.java:895) ~[na:1.8.0_382]
at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.lambda$null$2(AdapterProcessor.java:94) ~[client-adapter.launcher-1.1.5.jar:na]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_382]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_382]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_382]
at java.lang.Thread.run(Thread.java:750) ~[na:1.8.0_382]
Caused by: org.elasticsearch.ElasticsearchStatusException: Elasticsearch exception [type=index_not_found_exception, reason=no such index [book]]
at org.elasticsearch.rest.BytesRestResponse.errorFromXContent(BytesRestResponse.java:177) ~[na:na]
at org.elasticsearch.client.RestHighLevelClient.parseEntity(RestHighLevelClient.java:1727) ~[na:na]
at org.elasticsearch.client.RestHighLevelClient.parseResponseException(RestHighLevelClient.java:1704) ~[na:na]
at org.elasticsearch.client.RestHighLevelClient.internalPerformRequest(RestHighLevelClient.java:1467) ~[na:na]
at org.elasticsearch.client.RestHighLevelClient.performRequest(RestHighLevelClient.java:1439) ~[na:na]
at org.elasticsearch.client.RestHighLevelClient.performRequestAndParseEntity(RestHighLevelClient.java:1406) ~[na:na]
at org.elasticsearch.client.IndicesClient.getMapping(IndicesClient.java:266) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection.getMapping(ESConnection.java:132) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate.getEsType(ES7xTemplate.java:393) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate.getValFromData(ES7xTemplate.java:274) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate.getESDataFromDmlData(ES7xTemplate.java:298) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.singleTableSimpleFiledInsert(ESSyncService.java:439) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.insert(ESSyncService.java:139) ~[na:na]
at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.sync(ESSyncService.java:99) ~[na:na]
... 11 common frames omitted
Suppressed: org.elasticsearch.client.ResponseException: method [GET], host [http://127.0.0.1:9200], URI [/book/_mapping?master_timeout=30s&ignore_throttled=false&ignore_unavailable=false&expand_wildcards=open&allow_no_indices=true], status line [HTTP/1.1 404 Not Found]
{"error":{"root_cause":[{"type":"index_not_found_exception","reason":"no such index [book]","index_uuid":"_na_","resource.type":"index_or_alias","resource.id":"book","index":"book"}],"type":"index_not_found_exception","reason":"no such index [book]","index_uuid":"_na_","resource.type":"index_or_alias","resource.id":"book","index":"book"},"status":404}
at org.elasticsearch.client.RestClient.convertResponse(RestClient.java:253) ~[na:na]
at org.elasticsearch.client.RestClient.performRequest(RestClient.java:231) ~[na:na]
at org.elasticsearch.client.RestClient.performRequest(RestClient.java:205) ~[na:na]
at org.elasticsearch.client.RestHighLevelClient.internalPerformRequest(RestHighLevelClient.java:1454) ~[na:na]
... 21 common frames omitted
2023-09-29 12:52:32.488 [Thread-4] ERROR c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - Outer adapter sync failed! Error sync but AC
那就手動(dòng)建立索引
PUT book
{
"mappings": {
"properties": {
"create_time": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"id": {
"type": "long"
},
"last_updated": {
"type": "long"
},
"my_json": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"name": {
"type": "text"
},
"name2": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"name3": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"name4": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"name5": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"name6": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
},
"year": {
"type": "long"
}
}
}
}
注意事項(xiàng)
- adapter配置的ES7目錄下yml中SQL語句，每個(gè)表必須要有別名，哪怕單表也是！??！不然修改就報(bào)錯(cuò)
- mysql改字段類型對(duì)于同步es沒啥影響，主要是加字段要特殊處理
我的問題
1、不能同步建表語句到es的mappings。解決：手動(dòng)put
2、添加mysql字段后不能把這個(gè)alter ddl直接同步到es里，還需要手動(dòng)修改table_name.yml的sql語句加字段（雖然說不用重啟canal服務(wù)會(huì)自動(dòng)刷新配置但還是麻煩），尤其是在修改途中如果出現(xiàn)了數(shù)據(jù)表的新字段insert和update。那么這些差異數(shù)據(jù)怎么同步過去？確實(shí)有這種臟數(shù)據(jù)存在??！
解決方案：按一下步驟操作就可以避免這種問題。就是加mysql字段之前先停掉復(fù)制
- 停止canal adapter服務(wù)停止復(fù)制
- mysql加字段
- 改adapter的book.yml加字段
- 啟動(dòng)canal adapter啟動(dòng)復(fù)制
3、字段類型為tinyint
并且當(dāng)我不配制status字段映射時(shí)，可以同步成功，但是status字段為null，是0被轉(zhuǎn)成了false的問題，已修改源碼解決。
4、新增為datetime類型的字段時(shí)adapter也不支持，報(bào)錯(cuò)：
java.lang.IllegalArgumentException: cannot write xcontent for unknown value of type class java.sql.Timestamp
java.lang.RuntimeException: java.lang.IllegalArgumentException: cannot write xcontent for unknown value of type class java.sql.Timestamp
解決：
在select的時(shí)候手動(dòng)format時(shí)間字段，轉(zhuǎn)成es可以接受的時(shí)間格式如date_format(trv.create_time,'%Y-%m-%d %H:%I:%S') ，當(dāng)然這個(gè)格式要和es mapping中的時(shí)間字段的format要對(duì)齊，然后etl就沒有問題了；方法二就是ES重建索引
5、如果期間同步錯(cuò)誤，那么錯(cuò)誤的那條就會(huì)丟失。修復(fù)后也找不到了。是不是引入mq就能解決？有待驗(yàn)證
正常同步insert的日志
properties:name=configurationPropertiesRebinder,context=484094a5,type=ConfigurationPropertiesRebinder]
2023-09-29 12:17:33.739 [main] INFO c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## syncSwitch refreshed.
2023-09-29 12:17:33.740 [main] INFO c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## start the canal client adapters.
2023-09-29 12:17:33.742 [main] INFO c.a.otter.canal.client.adapter.support.ExtensionLoader - extension classpath dir: E:\workspace\canal\canal.adapter-1.1.5\plugin
2023-09-29 12:17:33.792 [main] INFO c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: logger succeed
2023-09-29 12:17:34.013 [main] INFO c.a.o.c.client.adapter.es.core.config.ESSyncConfigLoader - ## Start loading es mapping config ...
2023-09-29 12:17:34.076 [main] INFO c.a.o.c.client.adapter.es.core.config.ESSyncConfigLoader - ## ES mapping config loaded
2023-09-29 12:17:34.370 [main] INFO c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: es7 succeed
2023-09-29 12:17:34.378 [main] INFO c.alibaba.otter.canal.connector.core.spi.ExtensionLoader - extension classpath dir: E:\workspace\canal\canal.adapter-1.1.5\plugin
2023-09-29 12:17:34.395 [main] INFO c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Start adapter for canal-client mq topic: example-g1 succeed
2023-09-29 12:17:34.395 [Thread-4] INFO c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Start to connect destination: example <=============
2023-09-29 12:17:34.395 [main] INFO c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## the canal client adapters are running now ......
2023-09-29 12:17:34.404 [main] INFO org.apache.coyote.http11.Http11NioProtocol - Starting ProtocolHandler ["http-nio-8081"]
2023-09-29 12:17:34.406 [main] INFO org.apache.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
2023-09-29 12:17:34.423 [main] INFO o.s.boot.web.embedded.tomcat.TomcatWebServer - Tomcat started on port(s): 8081 (http) with context path ''
2023-09-29 12:17:34.427 [main] INFO c.a.otter.canal.adapter.launcher.CanalAdapterApplication - Started CanalAdapterApplication in 5.239 seconds (JVM running for 5.788)
2023-09-29 12:17:34.464 [Thread-4] INFO c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Subscribe destination: example succeed <=============
2023-09-29 12:17:35.101 [pool-2-thread-1] INFO c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":4,"name":"涓夊浗婕斾箟4","year":"1987","last_updated":1695651267677}],"database":"db_example","destination":"example","es":1695960855000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"book","ts":1695961054998,"type":"INSERT"}
