canal环境部署

docker 部署canal同步数据

1 环境:

2 Mysql 配置

1. 查看 binlog 是否启用 SHOW VARIABLES LIKE ‘%log_bin%’;
2. 
   开启 binlog,
   修改 my.cnf
   docker cp mysql:/etc/my.cnf /data/mysql/conf 拷贝文件到临时目录修改后再拷贝回去

# For advice on how to change settings please see
# http://dev.mysql.com/doc/refman/5.7/en/server-configuration-defaults.html

[mysqld]
sql_mode=STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION
## 开启二进制日志功能
log-bin=mall-mysql-bin
server-id=1
## 设置二进制日志使用内存大小
binlog_cache_size=1M

## 设置使用的二进制日志格式
binlog_format=row

## 二进制日志过期清理时间，默认值未0，表示不自动清理
expire_logs_days=7

## 跳过主从复制中遇到的所有错误或者置顶类型的错误，避免slave端复制中断
## 1062 主键重复、1032主从数据不一致
slave_skip_errors=1062

#
# Remove leading # and set to the amount of RAM for the most important data
# cache in MySQL. Start at 70% of total RAM for dedicated server, else 10%.
# innodb_buffer_pool_size = 128M
#
# Remove leading # to turn on a very important data integrity option: logging
# changes to the binary log between backups.
# log_bin
#
# Remove leading # to set options mainly useful for reporting servers.
# The server defaults are faster for transactions and fast SELECTs.
# Adjust sizes as needed, experiment to find the optimal values.
# join_buffer_size = 128M
# sort_buffer_size = 2M
# read_rnd_buffer_size = 2M
skip-host-cache
skip-name-resolve
datadir=/var/lib/mysql
socket=/var/run/mysqld/mysqld.sock
secure-file-priv=/var/lib/mysql-files
user=mysql

# Disabling symbolic-links is recommended to prevent assorted security risks
symbolic-links=0

#log-error=/var/log/mysqld.log
pid-file=/var/run/mysqld/mysqld.pid
[client]
socket=/var/run/mysqld/mysqld.sock

!includedir /etc/mysql/conf.d/
!includedir /etc/mysql/mysql.conf.d/

问题:

解决:
设置bin log日志的时候，没有设置server_id参数,配置文件增加
server-id=1 ,重启服务.

3 创建订阅 binlog 账号

创建账户：create user canal IDENTIFIED by ‘canal’;
授权账户：grant select,replication slave, replication client on . to ‘canal’@‘%’;
刷新权限：flush PRIVILEGES;
查询账户;select host, user, authentication_string, plugin from user;
更新密码: ALTER USER ‘canal’@‘%’ IDENTIFIED WITH mysql_native_password BY ‘canal’;

4 Canal 配置

4.1 canal.deployer 对 conf/instance.properties 进行修改，配置你的数据库地址、账号信息

4.2 canal.adapter
4.2.1 application.yml 服务信息配置、es7 文件夹下同步表配置

4.2.2 配置数据库信息
dataSourceKey: lottery_01 是对应的读取库信息，_index: lottery.user_take_activity 是索引。

dataSourceKey: lottery_01
destination: example
groupId: g1
esMapping:
  _index: lottery.user_take_activity
  _id: _id
  sql: "select t.uuid as _id,
               t.u_id as _u_id,
               t.take_id as _take_id,
               t.activity_id as _activity_id,
               t.activity_name as _activity_name,
               t.take_date as _take_date,
               t.take_count as _take_count,
               t.strategy_id as _strategy_id,
               t.state as _state,
               t.uuid as _uuid,
               t.create_time as _create_time,
               t.update_time as _update_time
        from user_take_activity t"
  etlCondition: "where t.update_time>={}"
  commitBatch: 3000

#  PUT lottery.user_take_activity
#  {
#    "mappings": {
#      "properties": {
#        "_u_id":{"type": "text"},
#        "_take_id":{"type": "text"},
#        "_activity_id":{"type": "long"},
#        "_activity_name":{"type": "text"},
#        "_take_date":{"type": "date"},
#        "_take_count":{"type": "long"},
#        "_strategy_id":{"type": "long"},
#        "_state":{"type": "long"},
#        "_uuid":{"type": "text"},
#        "_create_time":{"type": "date"},
#        "_update_time":{"type": "date"},
#      }
#    }
#  }

#{
#  "mappings":{
#    "_doc":{
#      "properties":{
#        "id": {
#          "type": "long"
#        },
#        "name": {
#          "type": "text"
#        },
#        "email": {
#          "type": "text"
#        },
#        "order_id": {
#          "type": "long"
#        },
#        "order_serial": {
#          "type": "text"
#        },
#        "order_time": {
#          "type": "date"
#        },
#        "customer_order":{
#          "type":"join",
#          "relations":{
#            "customer":"order"
#          }
#        }
#      }
#    }
#  }
#}

4.3 Kinaba 添加索引
每需要同步一个表，就需要在 Kinaba 创建索引映射

PUT lottery.user_take_activity
{
  "mappings": {
    "properties": {
      "_u_id":{"type": "text"},
      "_take_id":{"type": "text"},
      "_activity_id":{"type": "long"},
      "_activity_name":{"type": "text"},
      "_take_date":{"type": "date"},
      "_take_count":{"type": "long"},
      "_strategy_id":{"type": "long"},
      "_state":{"type": "long"},
      "_uuid":{"type": "text"},
      "_create_time":{"type": "date"},
      "_update_time":{"type": "date"}
    }
  }
}

五 启动服务