免备案虚拟主机1元优化设计四年级上册语文答案
Cannl 数据同步
目录
- Cannl 数据同步
- 一、概述
- 1、简介
- 2、原理
- 3、模块
- 二、配置MySQL
- 1、使用版本
- 使用版本
- 2、环境要求
- 1）操作系统
- 2）MySQL要求
- 三、配置Canal-server
- 1、下载安装
- 2、**修改配置**
- **单机配置**
- **集群配置**
- **分库分表配置**
- 四、配置canal-adapter
- 1.同步ES配置
- 1 修改启动配置
- 2 配置映射文件
- 3 启动ES数据同步
- 查询所有订阅同步
- 启动
- 4.验证
- 2.注意事项：
- 五、配置canal-admin
一、概述
1、简介
canal是Alibaba旗下的一款开源项目，Java开发。基于数据库增量日志解析，提供增量数据订阅&消费。
Git地址：https://github.com/alibaba/canal
docker : https://github.com/alibaba/canal/wiki/Docker-QuickStart
docker-hub: https://hub.docker.com/r/canal/canal-server/tags/
2、原理
canal模拟MySQL主库和从库的交互协议，向MySQL主库发送dump协议，MySQL主库收到dump请求会向canal推送binlog，canal通过解析binlog将数据同步到其他存储（MySQL、Elasticsearch、HBase）中去。
3、模块
canal-server：监听MySQL的binlog。
canal-adapter：相当于canal的客户端，会从canal-server中获取数据，然后对数据进行同步，可以同步到MySQL、Elasticsearch和HBase等存储中去。
canal-admin：支持面向WebUI的canal动态管理能力，支持配置、任务、日志等在线白屏运维能力，具体文档：https://github.com/alibaba/canal/wiki/Canal-Admin-Guide
二、配置MySQL
1、使用版本
使用版本
| 软件 | 版本 |
|---|---|
| canal | 1.1.5 |
| es | 6.8.0 |
2、环境要求
https://github.com/alibaba/canal/wiki/AdminGuide
1）操作系统
a. 纯java开发，windows/linux均可支持
b. jdk建议使用1.6.25以上的版本，稳定可靠，目前阿里巴巴使用基本为此版本。
2）MySQL要求
a. 当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)，ps. mysql4.x版本没有经过严格测试，理论上是可以兼容
b. canal的原理是基于mysql binlog技术，所以这里一定需要开启mysql的binlog写入功能，并且配置binlog模式为row.
[mysqld]
log-bin=mysql-bin #添加这一行就ok
binlog-format=ROW #选择row模式
server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复
# 指定同步数据库据库重启后, 简单测试 my.cnf 配置是否生效:
mysql> show variables like 'binlog_format';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW |
+---------------+-------+mysql> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin | ON |
+---------------+-------+c. canal的原理是模拟自己为mysql slave，所以这里一定需要做为mysql slave的相关权限
mysql -uroot -pmysql> CREATE USER canal IDENTIFIED BY 'canal';
mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
#--GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
mysql> FLUSH PRIVILEGES;# 针对已有的账户可通过grants查询权限：
mysql> show grants for 'canal';
三、配置Canal-server
server代表一个canal运行实例，instance对应于一个数据队列（1个server对应n个instance）。
1、下载安装
离线下载
访问：https://github.com/alibaba/canal/tree/gh-pages/download ，会列出所有历史的发布版本包
在线下载
wget https://raw.github.com/alibaba/canal/gh-pages/download/canal.deployer-1.1.5.tar.gz
上传解压
# ---------------------------- canal deployer ----------------------------
# 上传
scp /Users/wangfugui/Downloads/dev/middleware/elasticsearch/tools/canal/canal.deployer-1.1.5.tar.gz root@172.16.227.129:/root/software
# canal.deployer-1.1.5.tar.gz 没有父文件夹
mkdir -p /root/software/canal/canal.deployer-1.1.5
# 解压
tar zxvf canal.deployer-1.1.5.tar.gz -C /root/software/canal/canal.deployer-1.1.5
# ---------------------------- canal adapter ----------------------------
# 上传
scp /Users/wangfugui/Downloads/dev/middleware/elasticsearch/tools/canal/canal.adapter-1.1.5.tar.gz root@172.16.227.129:/root/software
# canal.adapter-1.1.5.tar.gz 没有父文件夹
mkdir -p /root/software/canal/canal.adapter-1.1.5
# 解压
tar zxvf canal.adapter-1.1.5.tar.gz -C /root/software/canal/canal.adapter-1.1.5
# ---------------------------- canal admin ----------------------------
# 上传
scp /Users/wangfugui/Downloads/dev/middleware/elasticsearch/tools/canal/canal.admin-1.1.5.tar.gz root@172.16.227.129:/root/software
# canal.admin-1.1.5.tar.gz 没有父文件夹
mkdir -p /root/software/canal/canal.admin-1.1.5
# 解压
tar zxvf canal.admin-1.1.5.tar.gz -C /root/software/canal/canal.admin-1.1.52、修改配置
[root@localhost conf]# tree
.
├── canal_local.properties
├── canal.properties # 全局配置，很多配置都可以被 instance.propeties 覆盖，配置通用信息
├── example
│ └── instance.properties # 实例配置，存放实例的非共享配置，比如数据库 IP，账号，密码等
├── logback.xml
├── metrics
│ └── Canal_instances_tmpl.json
└── spring├── base-instance.xml├── default-instance.xml├── file-instance.xml├── group-instance.xml # 多库合并时，可以将多个物理instance合并为一个逻辑instance。├── memory-instance.xml└── tsdb├── h2-tsdb.xml├── mysql-tsdb.xml├── sql│ └── create_table.sql└── sql-map├── sqlmap-config.xml├── sqlmap_history.xml└── sqlmap_snapshot.xml单机配置
https://github.com/alibaba/canal/wiki/QuickStart
集群配置
https://github.com/alibaba/canal/wiki/AdminGuide#ha%E6%A8%A1%E5%BC%8F%E9%85%8D%E7%BD%AE
分库分表配置
处理分表分库的场景，主要是要使用配置group-instance.xml。group-instance主要针对需要进行多库合并时，可以将多个物理instance合并为一个逻辑instance，提供客户端访问。
1.修改全局配置
https://github.com/alibaba/canal/wiki/AdminGuide
vim /root/software/canal/canal.deployer-1.1.5/conf/canal.properties
#############################################################
######### common argument #############
#############################################################
canal.ip = 172.16.227.129
canal.port = 11111
# canal.instance.parser.parallel = false # 如果系统是1个 cpu，设置为 false
# canal.instance.parser.parallelThreadSize = 2 # cpu个数
#############################################################
######### destinations #############
#############################################################
# 当前server上部署的instance列表：可设置多个，需要在canal.conf.dir对应的目录下建立同名的文件
canal.destinations = mysql_server_0,mysql_server_1,mysql_server_2,mysql_server_3# 启动 group-instance.xml
# 解析数据库信息 : <property name="address" value="${canal.instance.master1.address}" />
canal.instance.global.spring.xml = classpath:spring/group-instance.xml2.实例配置
垂直分库
有几个数据库，复制几份canal instance的文件夹：
cd /root/software/canal/canal.deployer-1.1.5/conf
cp -R example mysql_server_0;
修改每个实例配置
vim /root/software/canal/canal.deployer-1.1.5/conf/mysql_server_0/instance.properties# position info (group-instance.xml中解析默认为canal.instance.master1.address)
canal.instance.master1.address=172.25.1.246:3307
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
# mq config # topic用同一个，客户端只监听这一个，也可不同示例配成不同，客户端分别监听
canal.mq.topic=enterprise_topic
# 从库配置
#canal.instance.standby.address =# -------------------------- 监听表 --------------------------
# table regex : 监听表规则（Perl正则表达式，多规则用逗号分隔）
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=db_toker_*.tbl_enterprise_mtype_*:id/eid/credit_no/full_name,db_toker_*.tbl_enterprise_stime_*:id/eid/credit_no/full_name
# 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配置好第一个，其他实例只改数据库连接。
cp -R mysql_server_0 mysql_server_1;
cp -R mysql_server_0 mysql_server_2;
cp -R mysql_server_0 mysql_server_3;
rm -rf example;水平分库
修改全局配置
vim /root/software/canal/canal.deployer-1.1.5/conf/canal.properties
canal.destinations = mysql_server修改实例配置
cd /root/software/canal/canal.deployer-1.1.5/conf
mv example mysql_server;vim /root/software/canal/canal.deployer-1.1.5/conf/mysql_server/instance.properties# position info
canal.instance.master1.address=172.25.1.246:3307
canal.instance.master1.journal.name=
canal.instance.master1.position=
canal.instance.master1.timestamp=
canal.instance.master1.gtid=# position info
canal.instance.master2.address=172.25.1.246:3308
canal.instance.master2.journal.name=
canal.instance.master2.position=
canal.instance.master2.timestamp=
canal.instance.master2.gtid=# position info
canal.instance.master3.address=172.25.1.246:3309
canal.instance.master3.journal.name=
canal.instance.master3.position=
canal.instance.master3.timestamp=
canal.instance.master3.gtid=# position info
canal.instance.master4.address=172.25.1.246:3310
canal.instance.master4.journal.name=
canal.instance.master4.position=
canal.instance.master4.timestamp=
canal.instance.master4.gtid=# mq config
canal.mq.topic=enterprise_topic3.修改group-instance.xml
主要针对需要进行多库合并时，可以将多个物理instance合并为一个逻辑instance，提供客户端访问。
场景：分库业务。 比如产品数据拆分了4个库，每个库会有一个instance，如果不用group，业务上要消费数据时，需要启动4个客户端，分别链接4个instance实例。使用group后，可以在canal server上合并为一个逻辑instance，只需要启动1个客户端，链接这个逻辑instance即可。
vim /root/software/canal/canal.deployer-1.1.5/conf/spring/
# -------------------- 修改 eventParser --------------------<bean id="eventParser" class="com.alibaba.otter.canal.parse.inbound.group.GroupEventParser"><property name="eventParsers"><list><ref bean="eventParser1" /><ref bean="eventParser2" /><ref bean="eventParser3" /><ref bean="eventParser4" /></list></property></bean>
# -------------------- 增加 eventParser3,4 --------------------
<bean id="eventParser3" parent="baseEventParser"><property name="destination" value="${canal.instance.destination}" /><property name="slaveId" value="${canal.instance.mysql.slaveId:0}" /><!-- 心跳配置 --><property name="detectingEnable" value="${canal.instance.detecting.enable:false}" /><property name="detectingSQL" value="${canal.instance.detecting.sql}" /><property name="detectingIntervalInSeconds" value="${canal.instance.detecting.interval.time:5}" /><property name="haController"><bean class="com.alibaba.otter.canal.parse.ha.HeartBeatHAController"><property name="detectingRetryTimes" value="${canal.instance.detecting.retry.threshold:3}" /><property name="switchEnable" value="${canal.instance.detecting.heartbeatHaEnable:false}" /></bean></property><property name="alarmHandler" ref="alarmHandler" /><!-- 解析过滤处理 --><property name="eventFilter"><bean class="com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter" ><constructor-arg index="0" value="${canal.instance.filter.regex:.*\..*}" /></bean></property><property name="eventBlackFilter"><bean class="com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter" ><constructor-arg index="0" value="${canal.instance.filter.black.regex:}" /><constructor-arg index="1" value="false" /></bean></property><!-- 最大事务解析大小，超过该大小后事务将被切分为多个事务投递 --><property name="transactionSize" value="${canal.instance.transaction.size:1024}" /><!-- 网络链接参数 --><property name="receiveBufferSize" value="${canal.instance.network.receiveBufferSize:16384}" /><property name="sendBufferSize" value="${canal.instance.network.sendBufferSize:16384}" /><property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" /><!-- 解析编码 --><!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /--><property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" /><!-- 解析位点记录 --><property name="logPositionManager"><bean class="com.alibaba.otter.canal.parse.index.MemoryLogPositionManager" /></property><!-- failover切换时回退的时间 --><property name="fallbackIntervalInSeconds" value="${canal.instance.fallbackIntervalInSeconds:60}" /><!-- 解析数据库信息 --><property name="masterInfo"><bean class="com.alibaba.otter.canal.parse.support.AuthenticationInfo" init-method="initPwd"><property name="address" value="${canal.instance.master3.address}" /><property name="username" value="${canal.instance.dbUsername:retl}" /><property name="password" value="${canal.instance.dbPassword:retl}" /><property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /><property name="enableDruid" value="${canal.instance.enableDruid:false}" /><property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /></bean></property><property name="standbyInfo"><bean class="com.alibaba.otter.canal.parse.support.AuthenticationInfo" init-method="initPwd"><property name="address" value="${canal.instance.standby3.address}" /><property name="username" value="${canal.instance.dbUsername:retl}" /><property name="password" value="${canal.instance.dbPassword:retl}" /><property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /><property name="enableDruid" value="${canal.instance.enableDruid:false}" /><property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /></bean></property><!-- 解析起始位点 --><property name="masterPosition"><bean class="com.alibaba.otter.canal.protocol.position.EntryPosition"><property name="journalName" value="${canal.instance.master3.journal.name}" /><property name="position" value="${canal.instance.master3.position}" /><property name="timestamp" value="${canal.instance.master3.timestamp}" /><property name="gtid" value="${canal.instance.master3.gtid}" /></bean></property><property name="standbyPosition"><bean class="com.alibaba.otter.canal.protocol.position.EntryPosition"><property name="journalName" value="${canal.instance.standby3.journal.name}" /><property name="position" value="${canal.instance.standby3.position}" /><property name="timestamp" value="${canal.instance.standby3.timestamp}" /><property name="gtid" value="${canal.instance.standby3.gtid}" /></bean></property><property name="filterQueryDml" value="${canal.instance.filter.query.dml:false}" /><property name="filterQueryDcl" value="${canal.instance.filter.query.dcl:false}" /><property name="filterQueryDdl" value="${canal.instance.filter.query.ddl:false}" /><property name="useDruidDdlFilter" value="${canal.instance.filter.druid.ddl:true}" /><property name="filterDmlInsert" value="${canal.instance.filter.dml.insert:false}" /><property name="filterDmlUpdate" value="${canal.instance.filter.dml.update:false}" /><property name="filterDmlDelete" value="${canal.instance.filter.dml.delete:false}" /><property name="filterRows" value="${canal.instance.filter.rows:false}" /><property name="filterTableError" value="${canal.instance.filter.table.error:false}" /><property name="supportBinlogFormats" value="${canal.instance.binlog.format}" /><property name="supportBinlogImages" value="${canal.instance.binlog.image}" /><!-- parallel parser --><property name="parallel" value="${canal.instance.parser.parallel:true}" /><property name="parallelThreadSize" value="${canal.instance.parser.parallelThreadSize}" /><property name="parallelBufferSize" value="${canal.instance.parser.parallelBufferSize:256}" /><property name="autoResetLatestPosMode" value="${canal.auto.reset.latest.pos.mode:false}" /></bean>4.启动监听
cd /root/software/canal/canal.deployer-1.1.5/bin
# 虚机1核1G启动失败，改成1核2G启动成功
./startup.sh
# 查看canal进程
ps aux | grep canal
# 日志路径
cd /root/software/canal/canal.deployer-1.1.5/logs
#./stop.sh'
https://github.com/alibaba/canal/blob/master/example/src/main/java/com/alibaba/otter/canal/example
客户端过滤数据会覆盖服务端过滤规则，按需配置。
四、配置canal-adapter
1.同步ES配置
-
Adapter本质上是为了将canal-server订阅到的实时增量数据进行消费，相当于client。
-
client-adapter分为适配器和启动器两部分, 适配器为多个fat jar, 每个适配器会将自己所需的依赖打成一个包, 以SPI的方式让启动器动态加载, 目前所有支持的适配器都放置在plugin目录下。
-
logback.xml中默认日志等级为debug，线上使用时，记得改info。
1 修改启动配置
vim /root/software/canal/canal.adapter-1.1.5/conf/application.ymlcanal.conf:consumerProperties:# canal tcp consumercanal.tcp.server.host: 127.0.0.1:11111canal.tcp.zookeeper.hosts:canal.tcp.batch.size: 500canal.tcp.username:canal.tcp.password:canalAdapters:- instance: mysql_server # canal instance Name or mq topic namegroups:- groupId: g1outerAdapters:- name: logger- name: eshosts: 172.16.227.129:9300,172.16.227.130:9300,172.16.227.131:9300 # 127.0.0.1:9200 for rest mod(es 集群地址, 逗号分隔)properties:mode: rest # or rest# security.auth: test:123456 # only used for rest modecluster.name: my-es-clusteradapter将会自动加载 conf/es 下的所有.yml结尾的配置文件
说明:
一份数据可以被多个group同时消费, 多个group之间会是一个并行执行, 一个group内部是一个串行执行多个outerAdapters
目前client adapter数据订阅的方式支持两种，直连canal server 或者 订阅kafka/RocketMQ的消息
2 配置映射文件
使用adapter同步mysql数据至es，需要预先在es中建立mapping关系，每个索引都需要定义一个配置。
修改 conf/es/mytest_user.yml文件:
dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值
outerAdapterKey: exampleKey # 对应application.yml中es配置的key
destination: mysql_server_0 # cannal的instance或者MQ的topic
groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据
esMapping:_index: enterprise # es 的索引名称_type: _doc # es 的type名称, es7下无需配置此项_id: _id # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
# pk: id # 如果不需要_id, 则需要指定一个属性为主键属性# sql映射sql: "select a.id as _id, a.name, a.role_id, a.c_time from user a"
# objFields:
# _labels: array:; # 数组或者对象属性, array:; 代表以;字段里面是以;分隔的
# _obj: object # json对象etlCondition: "" # etl 的条件参数commitBatch: 3000 # 提交批大小该sql对应的es mapping示例：
{"mytest_user": {"mappings": {"_doc": {"properties": {"name": {"type": "text"},"role_id": {"type": "long"},"c_time": {"type": "date"}}}}}
}ClientAdapter目前只支持垂直分库，不支持水平分库，待完善。
- (后续支持) ElasticSearch多表数据同步,ETL功能
3 启动ES数据同步
查询所有订阅同步
curl http://127.0.0.1:8081/destinations
数据同步开关
增量同步
默认值开启DML 同步，如果需要使用DDL同步能力，需要在rdb中将配置mirrorDb为true注释打开。
全量同步
参考官网 https://github.com/alibaba/canal/wiki/ClientAdapter#322-%E6%95%B0%E6%8D%AE%E5%90%8C%E6%AD%A5%E5%BC%80%E5%85%B3
开关
curl http://127.0.0.1:8081/syncSwitch/example/off -X PUT
开关状态
curl http://127.0.0.1:8081/syncSwitch/example
启动
# 启动canal-adapter启动器
./startup.sh
4.验证
新增mysql mytest.user表的数据, 将会自动同步到es的mytest_user索引下面, 并会打出DML的log
修改mysql mytest.role表的role_name, 将会自动同步es的mytest_user索引中的role_name数据
新增或者修改mysql mytest.label表的label, 将会自动同步es的mytest_user索引中的labels数据
2.注意事项：
Step1: 先把 adapter 下面的 bootstrap.yml，全部注释掉，否则会提示你 XX 表不存在。
Step2: 再修改 adapter 的 application.yml 配置文件。
Step3: 修改我们在 application.yml 中配置的目标数据源 es7 文件夹内容。
五、配置canal-admin
在线下载
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.admin-1.1.5.tar.gz
解压: tar zxvf canal.admin-1.1.5.tar.gz
创建canal-admin对应数据库