1. 使用场景
- canal deployer 为 canal 标准包，可将其看做 canal server。它负责伪装成 mysql 从库，接收、解析 binlog 并投递到指定的目标端（RDS、MQ 或 canal adapter）
- canal adapter 为 canal 的客户端，可将其看作 canal client。其中 RDB 包括 Oracle、MySQL、PostgreSQL、SQLServer 等数据库
2. canal deployer 包参数详解
2.1. canal deployer 包目录结构
tree canal 展开源码
- bin —— 存放启动、重启、停止脚本
- conf —— 配置文件目录
- lib —— 存放库文件，不用操作该目录
- logs —— 存放日志目录
2.2. conf 目录
tree canal conf 展开源码
- canal.properties —— canal 程序主配置文件
| argument | 说明 | 默认值 | 示例（--为保持默认） |
| canal.ip | canal server绑定的本地IP信息，如果不配置，默认选择一个本机IP进行启动服务 | 无 | |
| canal.register.ip | canal server 向 zk 集群中注册的 ip | 无 | |
| canal.port | canal server提供socket服务的端口 | 11111 | -- |
| canal.metrics.pull.port | 监控接口端口(配合Prometheus使用) | 11112 | -- |
| canal.admin.port | canal admin 的访问端口 | 11110 | -- |
| canal.admin.user | canal admin 的登陆用户 | admin | -- |
| canal.admin.passwd | canal admin 的登陆用户密码 | -- | |
| canal.zkServers | zookeeper 服务器集群地址 | -- | |
| canal.zookeeper.flush.period | canal持久化数据到zookeeper上的更新频率，单位毫秒 | 1000 | -- |
| canal.withoutNetty | 是否关闭server端的netty服务 | false | -- |
| canal.serverMode | tcp：客户端通过tcp方式从Canal服务端拉取增量数据 kafka：Canal服务端将增量数据同步到kafka中，客户端从kafka消费数据，此时客户端感知不到Canal的存在，只需要跟kafka交互 RocketMQ：同kafka，增量数据同步到RocketMQ中 | tcp | -- |
| canal.file.data.dir | flush meta cursor/parse position to file。记录消费位点到文件，当使用spring/file-instance.xml配置文件时 | ${canal.conf.dir} | -- |
| canal.file.flush.period | canal持久化数据到文件上的更新频率，单位毫秒 | 1000 | -- |
| canal.instance.memory.buffer.size | canal内存store中可缓存buffer记录数，需要为2的指数 | 16384 | -- |
| canal.instance.memory.buffer.memunit | 内存记录的单位大小，默认1KB，和buffer.size组合决定最终的内存使用大小 | 1024 | -- |
| canal.instance.memory.batch.mode | canal内存store中数据缓存模式 | MEMSIZE | -- |
| canal.instance.memory.rawEntry | batch.mode 相关参数，默认即可 | true | -- |
| canal.instance.detecting.enable | 是否开启心跳检查 | false | -- |
| canal.instance.detecting.sql | 心跳检查sql | select 1 | -- |
| canal.instance.detecting.interval.time | 心跳检查频率，单位秒 | 3 | -- |
| canal.instance.detecting.retry.threshold | 心跳检查失败重试次数 | 3 | -- |
| canal.instance.detecting.heartbeatHaEnable | 心跳检查失败后，是否开启自动mysql自动切换 | false | -- |
| canal.instance.transaction.size | 最大事务完整解析的长度支持。超过该长度后，一个事务可能会被拆分成多次提交到canal store中，无法保证事务的完整可见性 | 1024 | -- |
| canal.instance.fallbackIntervalInSeconds | canal发生mysql切换时，在新的mysql库上查找binlog时需要往前查找的时间，单位秒 | 60 | -- |
| canal.instance.network.receiveBufferSize | 网络链接参数，SocketOptions.SO_RCVBUF | 16384 | -- |
| canal.instance.network.sendBufferSize | 网络链接参数，SocketOptions.SO_SNDBUF | 16384 | -- |
| canal.instance.network.soTimeout | 网络链接参数，SocketOptions.SO_TIMEOUT | 30 | -- |
| canal.instance.filter.druid.ddl | 是否使用druid处理所有的ddl解析来获取库和表名 | true | -- |
| canal.instance.filter.query.dcl | 是否忽略dcl语句 | false | -- |
| canal.instance.filter.query.dml | 是否忽略dml语句 | false | -- |
| canal.instance.filter.query.ddl | 是否忽略ddl语句 | false | -- |
| canal.instance.filter.table.error | 是否忽略binlog表结构获取失败的异常 | false | -- |
| canal.instance.filter.rows | 是否忽略dml的数据变更事件 | false | -- |
| canal.instance.filter.transaction.entry | 是否忽略事务头和尾,比如针对写入kakfa的消息时，不需要写入TransactionBegin/Transactionend事件 | false | -- |
| canal.instance.binlog.format | 支持的binlog/format格式列表 | ROW,STATEMENT,MIXED | -- |
| canal.instance.binlog.image | 支持的binlog/image格式列表 | FULL,MINIMAL,NOBLOB | -- |
| canal.instance.get.ddl.isolation | ddl语句是否单独一个batch返回 | false | -- |
| canal.instance.parser.parallel | 是否开启binlog并行解析模式 | true | -- |
| canal.instance.parser.parallelBufferSize | binlog并行解析的异步ringbuffer队列 | 256 | -- |
| canal.instance.tsdb.enable | 是否开启tablemeta的 tsdb 功能 | true | false |
| canal.instance.tsdb.dir | 主要针对h2-tsdb.xml时对应h2文件的存放目录,默认为conf/xx/h2.mv.db | [Math Processing Error]�����.����.����.���:../����/{canal.instance.destination:} | -- |
| canal.instance.tsdb.url | jdbc url的配置 | jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; | -- |
| canal.instance.tsdb.dbUsername | tsdb用户名 | canal | -- |
| canal.instance.tsdb.dbPassword | tsdb密码 | canal | -- |
| canal.instance.tsdb.snapshot.interval | 导出快照间隔，默认24小时 | 24 | -- |
| canal.instance.tsdb.snapshot.expire | 清理快照时间间隔，默认360小时 | 360 | -- |
| canal.aliyun.accessKey | 阿里云 access key(如果不需要在本地binlog超过18小时被清理后自动下载oss上的binlog，可以忽略该值) | 无 | -- |
| canal.aliyun.secretKey | 阿里云 secret key(如果不需要在本地binlog超过18小时被清理后自动下载oss上的binlog，可以忽略该值) | 无 | -- |
- logback.xml —— 日志功能配置文件，一般不用修改
- metrics —— 监控项配置文件目录
- spring —— spring 框架配置文件目录，其中 *-instance.xml 文件用于指定 binlog 的位点信息存放于文件、内存、zk 等位置
- default-instance.xml： store选择了内存模式，其余的parser/sink依赖的位点管理选择了持久化模式，目前持久化的方式主要是写入zookeeper，保证数据集群共享。
- 特点：支持HA
- 场景：生产环境，集群化部署.
- memory-instance.xml：所有的组件(parser , sink , store)都选择了内存版模式，记录位点的都选择了memory模式，重启后又会回到初始位点进行解析。
- 特点：速度最快，依赖最少(不需要zookeeper)
- 场景：一般应用在quickstart，或者是出现问题后，进行数据分析的场景，不应该将其应用于生产环境
- group-instance.xml：主要针对需要进行多库合并时，可以将多个物理instance合并为一个逻辑instance，提供客户端访问。
- file-instance.xml：store选择了内存模式，其余的parser/sink依赖的位点管理选择了持久化模式，目前持久化的方式主要是写入文件，文件位置在instance目录下meta.dat，当前Canal节点重启会从该文件读取解析位置，无法在数据集群共享。
- default-instance.xml： store选择了内存模式，其余的parser/sink依赖的位点管理选择了持久化模式，目前持久化的方式主要是写入zookeeper，保证数据集群共享。
- mid_db —— 为 canal 实例配置文件，主要配置 mysql binlog 的起始位置、表过滤、MQ 目标 topic 等信息
2.3. canal.properties 文件详解
- canal.properties 文件分为了4大部分
- common argument
- destinations
- MQ
- Kafka Kerberos Info
2.3.1. common argument 部分
2.3.2. destinations 部分
| argument | 说明 | 默认值 | 示例 |
| canal.destinations | 当前server上部署的instance列表，多个实例用逗号分隔 | 无 | mysql_mysql |
| canal.conf.dir | conf 目录所在的路径 | ../conf | -- |
| canal.auto.scan | 开启instance自动扫描 | true | -- |
| canal.auto.scan.interval | instance自动扫描的间隔时间，单位秒 | 5 | -- |
| canal.instance.tsdb.spring.xml | tsdb 的 xml 配置文件 | classpath:spring/tsdb/h2-tsdb.xml | -- |
| canal.instance.global.mode | 全局配置加载方式 | spring | -- |
| canal.instance.global.lazy | 全局lazy模式 | false | -- |
| canal.instance.global.manager.address | 全局的manager配置方式的连接信息 | ${canal.admin.manager} | -- |
| canal.instance.global.spring.xml | 全局的spring配置方式的组件文件，连接 zk 使用 default 文件 | classpath:spring/file-instance.xml | classpath:spring/default-instance.xml |
2.3.3. MQ 部分
| argument | 说明 | 默认值 | 示例 |
| canal.mq.servers | kafka为bootstrap.servers | 127.0.0.1:6667 | -- |
| canal.mq.retries | 发送失败重试次数 | 0 | -- |
| canal.mq.batchSize | kafka为ProducerConfig.BATCH_SIZE_CONFIG | 16384 | -- |
| canal.mq.maxRequestSize | kafka为ProducerConfig.MAX_REQUEST_SIZE_CONFIG | 1048576 | -- |
| canal.mq.lingerMs | kafka为ProducerConfig.LINGER_MS_CONFIG , 如果是flatMessage格式建议将该值调大, 如: 200 | 100 | -- |
| canal.mq.bufferMemory | kafka为ProducerConfig.BUFFER_MEMORY_CONFIG | 33554432 | -- |
| canal.mq.canalBatchSize | Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) | 50 | -- |
| canal.mq.canalGetTimeout | Canal get数据的超时时间, 单位: 毫秒, 空为不限超时 | 100 | -- |
| canal.mq.flatMessage | 是否为flat json格式对象 | true | -- |
| canal.mq.compressionType | 压缩类型 | none | -- |
| canal.mq.acks | kafka为ProducerConfig.ACKS_CONFIG | all | -- |
| canal.mq.producerGroup | kafka无意义 | Canal-Producer | -- |
| canal.mq.accessChannel | kafka无意义 | local | -- |
2.3.4. Kafka Kerberos Info 部分
| argument | 说明 | 默认值 | 示例（--为保持默认） |
| canal.mq.kafka.kerberos.enable | kafka为ProducerConfig.ACKS_CONFIG | false | -- |
| canal.mq.kafka.kerberos.krb5FilePath | kafka kerberos认证 | "../conf/kerberos/krb5.conf" | -- |
| canal.mq.kafka.kerberos.jaasFilePath | kafka kerberos认证 | "../conf/kerberos/jaas.conf" | -- |
2.4. instance.properties 文件详解
| argument | 说明 | 默认值 | 示例（--为保持默认） |
| canal.instance.mysql.slaveId | mysql集群配置中的serverId概念，需要保证和当前mysql集群中id唯一(v1.0.26+版本之后canal会自动生成，不需要手工指定) | 无 | 482801 |
| canal.instance.gtidon | 是否启用mysql gtid的订阅模式 | false | -- |
| canal.instance.master.address | mysql主库地址 | 127.0.0.1:3306 | |
| canal.instance.master.journal.name | mysql主库链接时起始的binlog文件 | 无 | mysql-bin.000421 |
| canal.instance.master.position | mysql主库链接时起始的binlog偏移量 | 无 | 1048741422 |
| canal.instance.master.timestamp | mysql主库链接时起始的binlog的时间戳 | 无 | -- |
| canal.instance.master.gtid | mysql主库链接时对应的gtid位点 | 无 | -- |
| canal.instance.rds.accesskey | aliyun账号的ak信息 | 无 | -- |
| canal.instance.rds.secretkey | aliyun账号的sk信息 | 无 | -- |
| canal.instance.rds.instanceId | aliyun rds对应的实例id信息 | 无 | -- |
| canal.instance.tsdb.enable | 是否开启 tablemeta 的 tsdb 功能 | true | false |
| canal.instance.tsdb.url | tsdb 连接 url | jdbc:mysql://127.0.0.1:3306/canal_tsdb | -- |
| canal.instance.tsdb.dbUsername | tsdb 连接用户名 | -- | -- |
| canal.instance.tsdb.dbPassword | tsdb 用户密码 | -- | -- |
| canal.instance.standby.address | mysql备库连接地址 | -- | -- |
| canal.instance.standby.journal.name | mysql备库连接时起始的binlog文件 | -- | -- |
| canal.instance.standby.position | mysql备库连接时起始的binlog偏移量 | -- | -- |
| canal.instance.standby.timestamp | mysql备库连接时起始的binlog的时间戳 | -- | -- |
| canal.instance.standby.gtid | mysql备库连接时起始的binlog的时间戳 | -- | -- |
| canal.instance.dbUsername | mysql数据库帐号 | canal | u_canal_src_db |
| canal.instance.dbPassword | mysql数据库密码 | canal | xxxxxx |
| canal.instance.connectionCharset | mysql 字符集 | UTF-8 | -- |
| canal.instance.enableDruid | 是否使用 druid 数据库密码加密 | false | -- |
| canal.instance.pwdPublicKey | 密码公钥，需要开启 canal.instance.enableDruid | xxxxx | -- |
| canal.instance.filter.regex | mysql 过滤表白名单，Perl正则表达式 | .*\\..* | ddo_main.objective_release,ddo_main.ddo_post_info, dhr_organization.department,dhr_organization.position_info |
| canal.instance.filter.black.regex | mysql 过滤表黑名单，Perl正则表达式 | -- | -- |
| canal.mq.topic | mq 中的 topic 名 | -- | mysql_mysql |
| canal.mq.dynamicTopic | mq 中的动态 topic 规则，1.1.3 开始支持 | mytest1.user,mytest2\\..*,.*\\..* | -- |
| canal.mq.partition | 单队列模式下的分区下标 | 0 | -- |
| canal.mq.partitionsNum | hash 模式下的分区数 | -- | -- |
| canal.mq.partitionHash | hash 规则定义，库名.表名:唯一主键，1.1.3 支持新语法 | -- | -- |
3. canal adapter 包参数详解
3.1. canal adapter 目录结构
tree canal 折叠源码
| 1 2 3 4 5 6 7 | [root]# ll 总用量 8 drwxr-xr-x 2 root root 90 7月 27 19:52 bin drwxrwxrwx 6 root root 134 7月 28 16:52 conf drwxr-xr-x 2 root root 4096 6月 29 09:39 lib drwxrwxrwx 3 root root 20 6月 29 10:58 logs drwxrwxrwx 2 root root 4096 9月 2 2019 plugin |
- bin —— 可执行文件目录
- conf —— 配置文件目录
- lib —— 库文件目录
- logs —— 日志文件目录
- plugin —— 插件目录
3.2. conf 目录
tree canal conf 折叠源码
| 1 2 3 4 5 6 7 8 9 | [root]# ll 总用量 20 -rwxrwxrwx 1 root root 2099 7月 28 11:30 application.yml -rwxrwxrwx 1 root root 170 9月 2 2019 bootstrap.yml drwxr-xr-x 2 root root 67 6月 29 18:50 es drwxr-xr-x 2 root root 31 6月 29 09:39 hbase -rwxrwxrwx 1 root root 2172 9月 2 2019 logback.xml drwxrwxrwx 2 root root 29 6月 29 09:39 META-INF drwxrwxrwx 2 root root 4096 7月 28 15:19 rdb |
- application.yml —— canal client 主配置文件
- logback.xml —— 日志功能配置文件
- es、hbase、rdb —— 分别对应三类数据库的配置目录
- bootstrap.yml —— canal manager 连接信息配置文件
3.3. application.yml 文件配置详解
application.yml 折叠源码
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | server: port: 8081 # canal metrics 收集端口 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null canal.conf: mode: tcp # kafka rocketMQ # canalServerHost: # canal server 连接地址，单机模式下使用 zookeeperHosts: # zk 集群连接地址，HA 模式时配置 # mqServers: 127.0.0.1:6667 #or rocketmq # flatMessage: true batchSize: 500 syncBatchSize: 1000 retries: 0 timeout: accessKey: secretKey: srcDataSources: defaultDS: url: jdbc:mysql://:3306/ddo_main?useUnicode=true # 源库的连接信息 username: u_canal_src_db # 源库连接用户名 password: xxxxxx # 源库用户名密码 canalAdapters: - instance: mid_db # canal instance Name or mq topic name # canal 实例名 groups: # canal 分组列表 - groupId: g1 # canal 分组id, 如果是MQ模式将用到该值 outerAdapters: # 分组内适配器列表 - name: logger # 日志打印适配器 - name: rdb # 指定为rdb类型同步 key: mysql1 # 指定adapter的唯一key, 与表映射配置中outerAdapterKey对应 properties: jdbc.driverClassName: com.mysql.jdbc.Driver # jdbc驱动名, 部分jdbc的jar包需要自行放致lib目录下 jdbc.url: jdbc:mysql://:3306/canal_mid_db?useUnicode=true # 源数据库 jdbc url 地址 jdbc.username: u_canal_src_mid_db # 源数据库 jdbc 连接用户名 jdbc.password: xxxxxx # 源数据库 jdbc 用户密码 |
3.4. rdb 配置文件详解
rdb yml 展开源码
ataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值
destination: mysql_mysql # cannal的instance或者MQ的topic
groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据
uterAdapterKey: mysql1 # adapter key, 对应上面配置outAdapters中的key
concurrent: true # 是否按主键hash并行同步, 并行同步的表必须保证主键不会更改及主键不能为其他同步表的外键!!
dbMapping:
database: ddo_main # 源数据源的database/shcema
table: ddo_post_info # 源数据源表名
targetTable: canal_mid_db.ddo_post_info # 目标数据源的库名.表名
targetPk: # 主键映射
id: id # 如果是复合主键可以换行映射多个
mapAll: true # 是否整表映射, 要求源表和目标表字段名一模一样 (如果targetColumns也配置了映射,则以targetColumns配置为准)
# targetColumns: # 字段映射, 格式: 目标表字段: 源表字段, 如果字段名一样源表字段名可不填
# id:
# name:
# role_id:
# c_time:
# test1:
# etlCondition: "where c_time>={}"
commitBatch: 3000 # 批量提交的大小
3.5. es 配置文件详解
es yml 折叠源码
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 | dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值 destination: mysql_es # cannal的instance或者MQ的topic groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据 esMapping: _index: objective_index # es 索引名称 _type: _doc # es 的type名称, es7下无需配置此项 _id: oId # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配 upsert: true # 是否开启更新插入 # pk: id sql: "SELECT o.id AS oId, o.type AS oType, o.kind AS oKind, o.strategy AS oStrategy, o.version AS oVersion, o.interval_id AS oIntervalId, o.ldap AS oLdap, o.dept_id AS oDeptId, o.update_time AS oUpdateTime, o.create_time AS oCreateTime, pi.id AS piDdoPostId, pi.state AS piState, p.emp_dept_role_id AS pEmpDeptRoleId, p.emp_id AS pEmpId, p.full_ancestor_emp_dept_role_ids AS pFullAncestorEmpDeptRoleIds, p.offspring_detection AS pOffSpringDetection, d.tenement_id AS dTenementId, d.dept_id AS dDeptId, d.dept_name AS dDeptName, d.dept_level AS dDeptLevel, d.dept_path AS dDeptPath, d.manager_emp_id AS dManagerEmpId, d.manager_ldap AS dManagerLdap, d.hrbp_emp_id AS dHrbpEmpId, d.hrbp_ldap AS dHrbpLdap, d.hrbp_manager_emp_id AS dHrbpManagerEmpId, d.hrbp_manager_ldap AS dHrbpManagerLdap FROM objective_release o LEFT JOIN ddo_post_info pi ON o.ldap = pi.ldap AND o.dept_id = pi.dept_id AND o.interval_id = pi.interval_id LEFT JOIN position_info p ON o.ldap = p.ldap AND o.dept_id = p.dept_id LEFT JOIN department d ON o.dept_id = d.dept_id" # objFields: # _labels: array:; etlCondition: "WHERE p.tenement_id = 1 AND p.deleted = 0 AND d.tenement_id = 1" # etl 的条件参数 commitBatch: 3000 # 提交批大小 |
Back to top
