## File: README.md [](https://travis-ci.com/alibaba/canal) [](https://codecov.io/gh/alibaba/canal) [](http://isitmaintained.com/project/alibaba/canal "average time to resolve an issue") [](http://isitmaintained.com/project/alibaba/canal "percentage of issues still open") [](https://opensource.alibaba.com/contribution_leaderboard/details?projectValue=canal) ## 简介 **canal [kə'næl]**，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费 早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。 基于日志增量订阅和消费的业务包括 - 数据库镜像 - 数据库实时备份 - 索引构建和实时维护(拆分异构索引、倒排索引等) - 业务 cache 刷新 - 带业务逻辑的增量数据处理 当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x ## 工作原理 #### MySQL主备复制原理 - MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看) - MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log) - MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据 #### canal 工作原理 - canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议 - MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal ) - canal 解析 binary log 对象(原始为 byte 流) ## 重要版本更新说明 1. canal 1.1.x 版本（[release_note](https://github.com/alibaba/canal/releases)）,性能与功能层面有较大的突破,重要提升包括: - 整体性能测试&优化,提升了150%. #726 参考: [Performance](https://github.com/alibaba/canal/wiki/Performance) - 原生支持prometheus监控 #765 [Prometheus QuickStart](https://github.com/alibaba/canal/wiki/Prometheus-QuickStart) - 原生支持kafka消息投递 #695 [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart) - 原生支持aliyun rds的binlog订阅 (解决自动主备切换/oss binlog离线解析) 参考: [Aliyun RDS QuickStart](https://github.com/alibaba/canal/wiki/aliyun-RDS-QuickStart) - 原生支持docker镜像 #801 参考: [Docker QuickStart](https://github.com/alibaba/canal/wiki/Docker-QuickStart) 2. canal 1.1.4版本，迎来最重要的WebUI能力，引入canal-admin工程，支持面向WebUI的canal动态管理能力，支持配置、任务、日志等在线白屏运维能力，具体文档：[Canal Admin Guide](https://github.com/alibaba/canal/wiki/Canal-Admin-Guide) ## 文档 - [Home](https://github.com/alibaba/canal/wiki/Home) - [Introduction](https://github.com/alibaba/canal/wiki/Introduction) - [QuickStart](https://github.com/alibaba/canal/wiki/QuickStart) - [Docker QuickStart](https://github.com/alibaba/canal/wiki/Docker-QuickStart) - [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart) - [Aliyun RDS for MySQL QuickStart](https://github.com/alibaba/canal/wiki/aliyun-RDS-QuickStart) - [Prometheus QuickStart](https://github.com/alibaba/canal/wiki/Prometheus-QuickStart) - Canal Admin - [Canal Admin QuickStart](https://github.com/alibaba/canal/wiki/Canal-Admin-QuickStart) - [Canal Admin Guide](https://github.com/alibaba/canal/wiki/Canal-Admin-Guide) - [Canal Admin ServerGuide](https://github.com/alibaba/canal/wiki/Canal-Admin-ServerGuide) - [Canal Admin Docker](https://github.com/alibaba/canal/wiki/Canal-Admin-Docker) - [AdminGuide](https://github.com/alibaba/canal/wiki/AdminGuide) - [ClientExample](https://github.com/alibaba/canal/wiki/ClientExample) - [ClientAPI](https://github.com/alibaba/canal/wiki/ClientAPI) - [Performance](https://github.com/alibaba/canal/wiki/Performance) - [DevGuide](https://github.com/alibaba/canal/wiki/DevGuide) - [BinlogChange(MySQL 5.6)](https://github.com/alibaba/canal/wiki/BinlogChange%28mysql5.6%29) - [BinlogChange(MariaDB)](https://github.com/alibaba/canal/wiki/BinlogChange%28MariaDB%29) - [TableMetaTSDB](https://github.com/alibaba/canal/wiki/TableMetaTSDB) - [ReleaseNotes](http://alibaba.github.com/canal/release.html) - [Download](https://github.com/alibaba/canal/releases) - [FAQ](https://github.com/alibaba/canal/wiki/FAQ) ## 多语言 canal 特别设计了 client-server 模式，交互协议使用 protobuf 3.0 , client 端可采用不同语言实现不同的消费逻辑，欢迎大家提交 pull request - canal java 客户端: [https://github.com/alibaba/canal/wiki/ClientExample](https://github.com/alibaba/canal/wiki/ClientExample) - canal c# 客户端: [https://github.com/dotnetcore/CanalSharp](https://github.com/dotnetcore/CanalSharp) - canal go客户端: [https://github.com/CanalClient/canal-go](https://github.com/CanalClient/canal-go) - canal php客户端: [https://github.com/xingwenge/canal-php](https://github.com/xingwenge/canal-php) - canal Python客户端：[https://github.com/haozi3156666/canal-python](https://github.com/haozi3156666/canal-python) - canal Rust客户端：[https://github.com/laohanlinux/canal-rs](https://github.com/laohanlinux/canal-rs) - canal Nodejs客户端：[https://github.com/marmot-z/canal-nodejs](https://github.com/marmot-z/canal-nodejs) canal 作为 MySQL binlog 增量获取和解析工具，可将变更记录投递到 MQ 系统中，比如 Kafka/RocketMQ，可以借助于 MQ 的多语言能力 - 参考文档: [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart) ## 基于canal开发的工具 - canal2sql(基于binlog生成SQL) : [https://github.com/zhuchao941/canal2sql] ## 相关开源&产品 - [canal 消费端开源项目: Otter](http://github.com/alibaba/otter) - [阿里巴巴去 Oracle 数据迁移同步工具: yugong](http://github.com/alibaba/yugong) - [阿里巴巴离线同步开源项目 DataX](https://github.com/alibaba/datax) - [阿里巴巴数据库连接池开源项目 Druid](https://github.com/alibaba/druid) - [阿里巴巴实时数据同步工具 DTS](https://www.aliyun.com/product/dts) ## 问题反馈 - 报告 issue: [github issues](https://github.com/alibaba/canal/issues) 本项目的Issues会被同步沉淀至[阿里云开发者社区](https://developer.aliyun.com/ask) --- ## File: charts/README.md # helm charts 使用说明 包含`canal-admin`和`canal-server`两个chart，基本配置保存在k8s的`secret`里，`canal.properties`和`instance.properties`保存在mysql数据库中，使用zookeeper注册中心实现高可用。 - [helm charts 使用说明](#helm-charts-使用说明) - [创建zk集群](#创建zk集群) - [初始化数据库](#初始化数据库) - [部署canal-admin chart](#部署canal-admin-chart) - [配置values.yaml](#配置valuesyaml) - [配置集群信息](#配置集群信息) - [部署canal-server chart](#部署canal-server-chart) - [配置values.yaml](#配置valuesyaml-1) - [instance 实例创建](#instance-实例创建) ## 创建zk集群 ```sh helm install canal-zookeeper oci://registry-1.docker.io/bitnamicharts/zookeeper ``` 如需挂载磁盘参考 [bitnami/zookeeper](`https://github.com/bitnami/charts/tree/main/bitnami/zookeeper`) ## 初始化数据库 [/admin/admin-web/src/main/resources/canal_manager.sql](/admin/admin-web/src/main/resources/canal_manager.sql) [/deployer/src/main/resources/spring/tsdb/sql/create_table.sql](/deployer/src/main/resources/spring/tsdb/sql/create_table.sql) ## 部署canal-admin chart ### 配置values.yaml ```yaml # 主要配置 admin: config: | server: port: 8089 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 spring.datasource: address: your_db_host:3306 database: canal_manager username: **** password: **** driver-class-name: com.mysql.jdbc.Driver url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true hikari: maximum-pool-size: 30 minimum-idle: 1 canal: adminUser: admin adminPasswd: admin # debug: true ``` 全量参数见：[canal-admin/values.yaml](./canal-admin/values.yaml)，测试用例见：[tests/admin-values.yaml](./tests/admin-values.yaml) ```sh helm install canal-admin -f ./admin-values.yaml ./canal-admin ``` ### 配置集群信息 等待pod启动后，使用默认账户`admin`密码`123456`登录 创建集群`default`并配置`zk地址`为`canal-zookeeper.default.svc.cluster.local:2181` 配置完成后，点击`配置`-`主配置`，修改以下配置项，不是覆盖整个文件 ```yaml canal.user = admin canal.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441 canal.zkServers = canal-zookeeper.default.svc.cluster.local:2181 # tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ 看实际需求，此处改为了rabbitMQ canal.serverMode = rabbitMQ #canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:} canal.instance.tsdb.url = jdbc:mysql://you_db_host:3306/canal_tsdb?useUnicode=true&characterEncoding=UTF-8&useSSL=false canal.instance.tsdb.dbUsername = **** canal.instance.tsdb.dbPassword = **** #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.spring.xml = classpath:spring/file-instance.xml canal.instance.global.spring.xml = classpath:spring/default-instance.xml # 如果用的是阿里云polardb需要配置 canal.aliyun.accessKey = **** canal.aliyun.secretKey = **** canal.aliyun.uid = 主账号id # 我们是推送到rabbitmq所以需要配置 rabbitmq.host = ****:5672 rabbitmq.virtual.host = **** rabbitmq.exchange = **** # canal.aliyun.accessKey rabbitmq.username = **** # canal.aliyun.secretKey rabbitmq.password = **** rabbitmq.deliveryMode = direct ``` ## 部署canal-server chart ### 配置values.yaml ```yaml # 主要配置 server: config: | canal.port = 11111 canal.metrics.pull.port = 11112 # register ip canal.register.ip = # canal admin config canal.admin.manager = canal-admin.default: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 = default ``` ```sh helm install canal-server -f ./server-values.yaml ./canal-server ``` 全量参数见：[canal-server/values.yaml](./canal-server/values.yaml)，测试用例见：[tests/server-values.yaml](./tests/server-values.yaml) 至此，canal-server已自动创建，并自动注册。 ## instance 实例创建 主要配置如下 ```yaml # 配置你要监听的数据库 canal.instance.master.address=127.0.0.1:3306 canal.instance.dbUsername=**** canal.instance.dbPassword=**** # 过滤监听哪些表 canal.instance.filter.regex=.*\\..* # 如果是推送到rabbitmq，需要配置 Routing Key canal.mq.topic=你的Routing Key ``` --- ## File: client-adapter/README.md ## 基本说明 canal 1.1.1版本之后, 增加客户端数据落地的适配及启动功能, 目前支持功能: * 客户端启动器 * 同步管理REST接口 * 日志适配器, 作为DEMO * HBase的数据同步(表对表同步), ETL功能 * (后续支持) ElasticSearch多表数据同步,ETL功能 ## 环境版本 * 操作系统：无要求 * java版本: jdk1.8 以上 * canal 版本: 请下载最新的安装包，本文以当前v1.1.1 的canal.deployer-1.1.1.tar.gz为例 * MySQL版本 ：5.7.18 * HBase版本: Apache HBase 1.1.2, 若和服务端版本不一致可自行替换客户端HBase依赖 ## 一、适配器启动器 client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载 启动器为 SpringBoot 项目, 支持canal-client启动的同时提供相关REST管理接口, 运行目录结构为: ``` - bin restart.sh startup.bat startup.sh stop.sh - lib client-adapter.logger-1.1.1-jar-with-dependencies.jar client-adapter.hbase-1.1.1-jar-with-dependencies.jar ... - conf application.yml - hbase mytest_person2.yml - logs ``` 以上目录结构最终会打包成 canal-adapter-launcher.tar.gz 压缩包 ## 二、启动器 ### 2.1 启动器配置 application.yml #### canal相关配置部分说明 ``` canal.conf: canalServerHost: 127.0.0.1:11111 # 对应单机模式下的canal server的ip:port zookeeperHosts: slave1:2181 # 对应集群模式下的zk地址, 如果配置了canalServerHost, 则以canalServerHost为准 bootstrapServers: slave1:6667 # kafka或rocketMQ地址, 与canalServerHost不能并存 flatMessage: true # 扁平message开关, 是否以json字符串形式投递数据, 仅在kafka/rocketMQ模式下有效 canalInstances: # canal实例组, 如果是tcp模式可配置此项 - instance: example # 对应canal destination groups: # 对应适配器分组, 分组间的适配器并行运行 - outAdapters: # 适配器列表, 分组内的适配串行运行 - name: logger # 适配器SPI名 - name: hbase properties: # HBase相关连接参数 hbase.zookeeper.quorum: slave1 hbase.zookeeper.property.clientPort: 2181 zookeeper.znode.parent: /hbase mqTopics: # MQ topic组, 如果是kafka或者rockeMQ模式可配置此项, 与canalInstances不能并存 - mqMode: kafka # MQ的模式: kafak/rocketMQ topic: example # MQ topic groups: # group组 - groupId: g2 # group id outAdapters: # 适配器列表, 以下配置和canalInstances中一样 - name: logger ``` #### 适配器相关配置部分说明 ``` adapter.conf: datasourceConfigs: # 数据源配置列表, 数据源将在适配器中用于ETL、数据同步回查等使用 defaultDS: # 数据源 dataSourceKey, 适配器中通过该值获取指定数据源 url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true username: root password: 121212 adapterConfigs: # 适配器内部配置列表 - hbase/mytest_person2.yml # 类型/配置文件名, 这里示例就是对应HBase适配器hbase目录下的mytest_person2.yml文件 ``` ## 2.2 同步管理REST接口 #### 2.2.1 查询所有订阅同步的canal destination或MQ topic ``` curl http://127.0.0.1:8081/destinations ``` #### 2.2.2 数据同步开关 ``` curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT ``` 针对 example 这个canal destination/MQ topic 进行开关操作. off代表关闭, 该destination/topic下的同步将阻塞或者断开连接不再接收数据, on代表开启 注: 如果在配置文件中配置了 zookeeperHosts 项, 则会使用分布式锁来控制HA中的数据同步开关, 如果是单机模式则使用本地锁来控制开关 #### 2.2.3 数据同步开关状态 ``` curl http://127.0.0.1:8081/syncSwitch/example ``` 查看指定 canal destination/MQ topic 的数据同步开关状态 #### 2.2.4 手动ETL ``` curl http://127.0.0.1:8081/etl/hbase/mytest_person2.yml -X POST -d "params=2018-10-21 00:00:00" ``` 导入数据到指定类型的库 #### 2.2.5 查看相关库总数据 ``` curl http://127.0.0.1:8081/count/hbase/mytest_person2.yml ``` ### 2.3 启动canal-adapter示例 #### 2.3.1 启动canal server (单机模式), 参考: [Canal QuickStart](https://github.com/alibaba/canal/wiki/QuickStart) #### 2.3.2 修改conf/application.yml为: ``` server: port: 8081 logging: level: com.alibaba.otter.canal.client.adapter.hbase: DEBUG spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null canal.conf: canalServerHost: 127.0.0.1:11111 flatMessage: true canalInstances: - instance: example adapterGroups: - outAdapters: - name: logger ``` 启动 ``` bin/startup.sh ``` ## 三、HBase适配器 ### 3.1 修改启动器配置: application.yml ``` server: port: 8081 logging: level: com.alibaba.otter.canal.client.adapter.hbase: DEBUG spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null canal.conf: canalServerHost: 127.0.0.1:11111 flatMessage: true srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true username: root password: 121212 canalInstances: - instance: example adapterGroups: - outAdapters: - name: hbase properties: hbase.zookeeper.quorum: slave1 hbase.zookeeper.property.clientPort: 2181 zookeeper.znode.parent: /hbase ``` adapter将会自动加载 conf/hbase 下的所有.yml结尾的配置文件 ### 3.2 适配器表映射文件 修改 conf/hbase/mytest_person.yml文件: ``` dataSourceKey: defaultDS # 对应application.yml中的datasourceConfigs下的配置 hbaseMapping: # mysql--HBase的单表映射配置 mode: STRING # HBase中的存储类型, 默认统一存为String, 可选: #PHOENIX #NATIVE #STRING # NATIVE: 以java类型为主, PHOENIX: 将类型转换为Phoenix对应的类型 destination: example # 对应 canal destination/MQ topic 名称 database: mytest # 数据库名/schema名 table: person # 表名 hbaseTable: MYTEST.PERSON # HBase表名 family: CF # 默认统一Column Family名称 uppercaseQualifier: true # 字段名转大写, 默认为true commitBatch: 3000 # 批量提交的大小, ETL中用到 #rowKey: id,type # 复合字段rowKey不能和columns中的rowKey并存 # 复合rowKey会以 '|' 分隔 columns: # 字段映射, 如果不配置将自动映射所有字段, # 并取第一个字段为rowKey, HBase字段名以mysql字段名为主 id: ROWKE name: CF:NAME email: EMAIL # 如果column family为默认CF, 则可以省略 type: # 如果HBase字段和mysql字段名一致, 则可以省略 c_time: birthday: ``` 如果涉及到类型转换,可以如下形式: ``` ... columns: id: ROWKE$STRING ... type: TYPE$BYTE ... ``` 类型转换涉及到Java类型和Phoenix类型两种, 分别定义如下: ``` #Java 类型转换, 对应配置 mode: NATIVE $DEFAULT $STRING $INTEGER $LONG $SHORT $BOOLEAN $FLOAT $DOUBLE $BIGDECIMAL $DATE $BYTE $BYTES ``` ``` #Phoenix 类型转换, 对应配置 mode: PHOENIX $DEFAULT 对应PHOENIX里的VARCHAR $UNSIGNED_INT 对应PHOENIX里的UNSIGNED_INT 4字节 $UNSIGNED_LONG 对应PHOENIX里的UNSIGNED_LONG 8字节 $UNSIGNED_TINYINT 对应PHOENIX里的UNSIGNED_TINYINT 1字节 $UNSIGNED_SMALLINT 对应PHOENIX里的UNSIGNED_SMALLINT 2字节 $UNSIGNED_FLOAT 对应PHOENIX里的UNSIGNED_FLOAT 4字节 $UNSIGNED_DOUBLE 对应PHOENIX里的UNSIGNED_DOUBLE 8字节 $INTEGER 对应PHOENIX里的INTEGER 4字节 $BIGINT 对应PHOENIX里的BIGINT 8字节 $TINYINT 对应PHOENIX里的TINYINT 1字节 $SMALLINT 对应PHOENIX里的SMALLINT 2字节 $FLOAT 对应PHOENIX里的FLOAT 4字节 $DOUBLE 对应PHOENIX里的DOUBLE 8字节 $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节 $TIME 对应PHOENIX里的TIME 8字节 $DATE 对应PHOENIX里的DATE 8字节 $TIMESTAMP 对应PHOENIX里的TIMESTAMP 12字节 $UNSIGNED_TIME 对应PHOENIX里的UNSIGNED_TIME 8字节 $UNSIGNED_DATE 对应PHOENIX里的UNSIGNED_DATE 8字节 $UNSIGNED_TIMESTAMP 对应PHOENIX里的UNSIGNED_TIMESTAMP 12字节 $VARCHAR 对应PHOENIX里的VARCHAR 动态长度 $VARBINARY 对应PHOENIX里的VARBINARY 动态长度 $DECIMAL 对应PHOENIX里的DECIMAL 动态长度 ``` 如果不配置将以java对象原生类型默认映射转换 ### 3.3 启动HBase数据同步 #### 创建HBase表 在HBase shell中运行: ``` create 'MYTEST.PERSON', {NAME=>'CF'} ``` #### 启动canal-adapter启动器 ``` bin/startup.sh ``` #### 验证 修改mysql mytest.person表的数据, 将会自动同步到HBase的MYTEST.PERSON表下面, 并会打出DML的log ## 四、关系型数据库适配器 RDB adapter 用于适配mysql到任意关系型数据库(需支持jdbc)的数据同步及导入 ### 4.1 修改启动器配置: application.yml, 这里以oracle目标库为例 ``` server: port: 8081 logging: level: com.alibaba.otter.canal.client.adapter.rdb: DEBUG ...... canal.conf: canalServerHost: 127.0.0.1:11111 srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true username: root password: 121212 canalInstances: - instance: example groups: - outAdapters: - name: rdb # 指定为rdb类型同步 key: oracle1 # 指定adapter的唯一key, 与表映射配置中outerAdapterKey对应 properties: jdbc.driverClassName: oracle.jdbc.OracleDriver # jdbc驱动名, jdbc的jar包需要自行放致lib目录下 jdbc.url: jdbc:oracle:thin:@localhost:49161:XE # jdbc url jdbc.username: mytest # jdbc username jdbc.password: m121212 # jdbc password threads: 5 # 并行执行的线程数, 默认为1 commitSize: 3000 # 批次提交的最大行数 ``` 其中 outAdapter 的配置: name统一为rdb, key为对应的数据源的唯一标识需和下面的表映射文件中的outerAdapterKey对应, properties为目标库jdb的相关参数 adapter将会自动加载 conf/rdb 下的所有.yml结尾的表映射配置文件 ### 4.2 适配器表映射文件 修改 conf/rdb/mytest_user.yml文件: ``` dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值 destination: example # cannal的instance或者MQ的topic outerAdapterKey: oracle1 # adapter key, 对应上面配置outAdapters中的key concurrent: true # 是否按主键hase并行同步, 并行同步的表必须保证主键不会更改及主键不能为其他同步表的外键!! dbMapping: database: mytest # 源数据源的database/shcema table: user # 源数据源表名 targetTable: mytest.tb_user # 目标数据源的库名.表名 targetPk: # 主键映射 id: id # 如果是复合主键可以换行映射多个 # mapAll: true # 是否整表映射, 要求源表和目标表字段名一模一样 (如果targetColumns也配置了映射,则以targetColumns配置为准) targetColumns: # 字段映射, 格式: 目标表字段: 源表字段, 如果字段名一样源表字段名可不填 id: name: role_id: c_time: test1: ``` 导入的类型以目标表的元类型为准, 将自动转换 ### 4.3 启动RDB数据同步 #### 将目标库的jdbc jar包放入lib文件夹, 这里放入ojdbc6.jar #### 启动canal-adapter启动器 ``` bin/startup.sh ``` #### 验证 修改mysql mytest.user表的数据, 将会自动同步到Oracle的MYTEST.TB_USER表下面, 并会打出DML的log ## 五、ElasticSearch适配器 ### 5.1 修改启动器配置: application.yml ``` server: port: 8081 logging: level: com.alibaba.otter.canal.client.adapter.es: DEBUG spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null canal.conf: canalServerHost: 127.0.0.1:11111 flatMessage: true srcDataSources: defaultDS: url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true username: root password: 121212 canalInstances: - instance: example adapterGroups: - outAdapters: - name: es hosts: 127.0.0.1:9300 # es 集群地址, 逗号分隔 properties: cluster.name: elasticsearch # es cluster name ``` adapter将会自动加载 conf/es 下的所有.yml结尾的配置文件 ### 5.2 适配器表映射文件 修改 conf/es/mytest_user.yml文件: ``` dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值 destination: example # cannal的instance或者MQ的topic esMapping: _index: mytest_user # es 的索引名称 _type: _doc # es 的doc名称 _id: _id # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配 # pk: id # 如果不需要_id, 则需要指定一个属性为主键属性 # sql映射 sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name, a.c_time as _c_time, c.labels as _labels from user a left join role b on b.id=a.role_id left join (select user_id, group_concat(label order by id desc separator ';') as labels from label group by user_id) c on c.user_id=a.id" # objFields: # _labels: array:; # 数组或者对象属性, array:; 代表以;字段里面是以;分隔的 # _obj: obj:{"test":"123"} etlCondition: "where a.c_time>='{0}'" # etl 的条件参数 commitBatch: 3000 # 提交批大小 ``` sql映射说明: sql支持多表关联自由组合, 但是有一定的限制: 1. 主表不能为子查询语句 2. 只能使用left outer join即最左表一定要是主表 3. 关联从表如果是子查询不能有多张表 4. 主sql中不能有where查询条件(从表子查询中可以有where条件但是不推荐, 可能会造成数据同步的不一致, 比如修改了where条件中的字段内容) 5. 关联条件只允许主外键的'='操作不能出现其他常量判断比如: on a.role_id=b.id and b.statues=1 6. 关联条件必须要有一个字段出现在主查询语句中比如: on a.role_id=b.id 其中的 a.role_id 或者 b.id 必须出现在主select语句中 Elastic Search的mapping 属性与sql的查询值将一一对应(不支持 select *), 比如: select a.id as _id, a.name, a.email as _email from user, 其中name将映射到es mapping的name field, _email将 映射到mapping的_email field, 这里以别名(如果有别名)作为最终的映射字段. 这里的_id可以填写到配置文件的 _id: _id映射. #### 5.2.1.单表映射索引示例sql: ``` select a.id as _id, a.name, a.role_id, a.c_time from user a ``` 该sql对应的es mapping示例: ``` { "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "c_time": { "type": "date" } } } } } } ``` #### 5.2.2.单表映射索引示例sql带函数或运算操作: ``` select a.id as _id, concat(a.name,'_test') as name, a.role_id+10000 as role_id, a.c_time from user a ``` 函数字段后必须跟上别名, 该sql对应的es mapping示例: ``` { "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "c_time": { "type": "date" } } } } } } ``` #### 5.2.3.多表映射(一对一, 多对一)索引示例sql: ``` select a.id as _id, a.name, a.role_id, b.role_name, a.c_time from user a left join role b on b.id = a.role_id ``` 注:这里join操作只能是left outer join, 第一张表必须为主表!! 该sql对应的es mapping示例: ``` { "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "role_name": { "type": "text" }, "c_time": { "type": "date" } } } } } } ``` #### 5.2.4.多表映射(一对多)索引示例sql: ``` select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a left join (select user_id, group_concat(label order by id desc separator ';') as labels from label group by user_id) c on c.user_id=a.id ``` 注:left join 后的子查询只允许一张表, 即子查询中不能再包含子查询或者关联!! 该sql对应的es mapping示例: ``` { "mytest_user": { "mappings": { "_doc": { "properties": { "name": { "type": "text" }, "role_id": { "type": "long" }, "c_time": { "type": "date" }, "labels": { "type": "text" } } } } } } ``` #### 5.2.5.其它类型的sql示例: - geo type ``` select ... concat(IFNULL(a.latitude, 0), ',', IFNULL(a.longitude, 0)) AS location, ... ``` - 复合主键 ``` select concat(a.id,'_',b.type) as _id, ... from user a left join role b on b.id=a.role_id ``` - 数组字段 ``` select a.id as _id, a.name, a.role_id, c.labels, a.c_time from user a left join (select user_id, group_concat(label order by id desc separator ';') as labels from label group by user_id) c on c.user_id=a.id ``` 配置中使用: ``` objFields: labels: array:; ``` ### 5.3 启动ES数据同步 #### 启动canal-adapter启动器 ``` bin/startup.sh ``` #### 验证 1. 新增mysql mytest.user表的数据, 将会自动同步到es的mytest_user索引下面, 并会打出DML的log 2. 修改mysql mytest.role表的role_name, 将会自动同步es的mytest_user索引中的role_name数据 3. 新增或者修改mysql mytest.label表的label, 将会自动同步es的mytest_user索引中的labels数据 --- ## File: docker/base/README.md # osbase cd amd64 && docker build --no-cache -t canal/osbase ./ -f Dockerfile # osadmin cd amd64 && docker build --no-cache -t canal/osadmin ./ -f ./Dockerfile_admin