canal.deploy 配置
vim canal.properties canal.serverMode = tcp
vim example/instance.properties # 只同步 FlinkEtl 数据库的数据 # table regex canal.instance.filter.regex=FlinkEtl\\.tb\\d_\\d # 配置 tsdb ,支持表结构修改 # tsdb 是为预防表结构发生变化,而在 canal 订阅 binlog 时产生问题 # canal使用数据库存储上一次的表结构信息,然后对比两次的表结构,可解决此错误。 # table meta tsdb info canal.instance.tsdb.enable=true canal.instance.tsdb.url=jdbc:mysql://hadoop106:3306/canal_tsdb?useUnicode=true&characterEncoding=UTF-8&useSSL=false canal.instance.tsdb.dbUsername=root canal.instance.tsdb.dbPassword=root # 新增 配置文件 # 因为tsdb默认使用的是h2数据库(java内置数据库)。不增加此配置默认还是使用内置数据库, canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
vim instance.properties
################################################# ## mysql serverId , v1.0.26+ will autoGen # canal.instance.mysql.slaveId=0 # enable gtid use true/false canal.instance.gtidon=false # position info canal.instance.master.address=hadoop106:3306 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= canal.instance.master.gtid= # rds oss binlog canal.instance.rds.accesskey= canal.instance.rds.secretkey= canal.instance.rds.instanceId= # tsdb 是为预防表结构发生变化,而在 canal 订阅 binlog 时产生问题 # canal使用数据库存储上一次的表结构信息,然后对比两次的表结构,可解决此错误。 # 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=root canal.instance.tsdb.dbPassword=root # 新增 配置文件 # 因为tsdb默认使用的是h2数据库(java内置数据库)。不增加此配置默认还是使用内置数据库, canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml #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=root canal.instance.dbPassword=root canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== # table regex canal.instance.filter.regex=.*\\..* # table black regex canal.instance.filter.black.regex=mysql\\.slave_.* # 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,.*\\..* #################################################
canal 使用 tsdb 保存保存表结构 meta 的信息变化,本地使用默认使用的 h2db,生产建议使用 mysql.
create database canal_tsdb; use canal_tsdb; CREATE TABLE `meta_snapshot` ( `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键', `gmt_create` datetime NOT NULL COMMENT '创建时间', `gmt_modified` datetime NOT NULL COMMENT '修改时间', `destination` varchar(128) DEFAULT NULL COMMENT '通道名称', `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名', `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量', `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', `data` longtext COMMENT '表结构数据', `extra` text COMMENT '额外的扩展信息', PRIMARY KEY (`id`), UNIQUE KEY `binlog_file_offest` (`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), KEY `destination` (`destination`), KEY `destination_timestamp` (`destination`,`binlog_timestamp`), KEY `gmt_modified` (`gmt_modified`) ) COMMENT='表结构记录表快照表'; CREATE TABLE `meta_history` ( `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键', `gmt_create` datetime NOT NULL COMMENT '创建时间', `gmt_modified` datetime NOT NULL COMMENT '修改时间', `destination` varchar(128) DEFAULT NULL COMMENT '通道名称', `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名', `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量', `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema', `sql_schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema', `sql_table` varchar(1024) DEFAULT NULL COMMENT '对应的table', `sql_text` longtext COMMENT '执行的sql', `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql类型', `extra` text COMMENT '额外的扩展信息', PRIMARY KEY (`id`), UNIQUE KEY `binlog_file_offest` (`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), KEY `destination` (`destination`), KEY `destination_timestamp` (`destination`,`binlog_timestamp`), KEY `gmt_modified` (`gmt_modified`) ) COMMENT='表结构变化明细表'; ———————————————— https://blog.csdn.net/wfh45678/article/details/118546668
canal_tsdb 数据库表:
meta_history
meta_snapshot
若 canal 宕机后很久没启动,建议删除 conf/example/meta.dat
然后使用 canal_tsdb 数据库的话,清空这两个表,重启canal即可
vim canal.properties
# 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.url = jdbc:mysql://hadoop106:3306/canal_tsdb?useUnicode=true&characterEncoding=UTF-8&useSSL=false canal.instance.tsdb.dbUsername = root canal.instance.tsdb.dbPassword = root # 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
# 报错 com.alibaba.druid.pool.DruidDataSource # canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
删除 conf/example下面的 meta.dat 和 h2.mv.db
adapter/bin/stop.sh deployer/bin/stop.sh deployer/bin/startup.sh adapter/bin/startup.sh vim deployer/logs/canal/canal.log vim adapter/logs/adapter/adapter.log
Client Adapter配置
conf/application.yml 配置