Canal是阿里巴巴集团提供的一个开源产品,能够通过解析数据库的增量日志,提供增量数据的订阅和消费功能。当您需要将MySQL中的增量数据同步至阿里云ES时,可通过Canal来实现。本文以阿里云RDS MySQL为例,介绍具体的实现方法。

准备工作

在通过Canal同步数据前,请先准备以下依赖环境:
注意 本文使用的RDS MySQL、阿里云ES和ECS在相同的区域可用区VPC子网安全组下。
  • 阿里云RDS MySQL。
    用来存放源数据和增量数据,开通方式请参见创建RDS MySQL数据库。本文使用的配置如下。RDS for MySQL配置
  • canal.adapter-1.1.4.tar.gz和canal.deployer-1.1.4.tar.gz。

    进行数据库日志解析,获取增量变更进行同步,是Github中开源的ETL(Extract Transform Load)软件,详情请参见Canal

  • 阿里云ES。
    用来接收增量数据,开通方式请参见创建阿里云Elasticsearch实例。本文创建的实例的版本通用商业版6.7
    注意 Canal不支持ES 7.0及以上版本。
  • 阿里云ECS。

    用来连接RDS MySQL,以及部署canalDeployer和canalAdapter,开通方式请参见步骤一:创建ECS实例。本文创建的实例的镜像CentOS 7.6 64位

操作步骤

  1. 准备工作
  2. 步骤一:准备MySQL数据源
  3. 步骤二:创建索引和mapping
  4. 步骤三:安装MySQL
  5. 步骤四:安装JDK
  6. 步骤五:安装并启动Canal-server
  7. 步骤六:安装并启动Canal-adapter
  8. 步骤七:查看Canal同步的增量数据

步骤一:准备MySQL数据源

进入RDS控制台,创建RDS MySQL数据库和数据表。详情请参见RDS MySQL快速入门

本文创建的表名称为es_test,包含的字段如下图所示。es_test字段及索引

步骤二:创建索引和mapping

  1. 登录目标阿里云ES实例的Kibana控制台。
    登录控制台的具体步骤请参见登录Kibana控制台
  2. 在左侧导航栏,单击Dev Tools(开发工具)。
  3. Console中执行以下命令创建索引和mapping。
    注意 mapping中的字段需要与步骤一:准备MySQL数据源中创建的字段(名称和类型)保持一致。
    PUT es_test?include_type_name=true
    {
    
        "settings" : {
          "index" : {
            "number_of_shards" : "5",
            "number_of_replicas" : "1"
          }
        },
        "mappings" : {
            "_doc" : {
                "properties" : {
                  "count": {          
                       "type": "text"       
                   },
                  "id": {
                       "type": "integer"
                   },
                    "name" : {
                        "type" : "text",
                       "analyzer": "ik_smart"                   
                    },
                    "color" : {
                        "type" : "text"                    
                    }
                }
            }
        }
    }
    创建成功后,返回如下结果。
    {
      "acknowledged" : true,
      "shards_acknowledged" : true,
      "index" : "es_test"
    }

步骤三:安装MySQL

  1. 连接阿里云ECS实例。
    具体步骤请参见连接ECS实例
  2. 下载MySQL源安装包。
    wget http://dev.mysql.com/get/mysql57-community-release-el7-11.noarch.rpm
  3. 安装MySQL源。
    yum -y install mysql57-community-release-el7-11.noarch.rpm
  4. 检查MySQL源是否安装成功。
    yum repolist enabled | grep mysql.*
    安装成功会返回如下结果。MySQL源安装成功
  5. 安装MySQL。
    yum install mysql-community-server
  6. 启动MySQL服务并查看MySQL服务的状态。
    systemctl start mysqld.service
    systemctl status mysqld.service
    启动成功后会返回如下结果。启动MySQL并查看状态
  7. 配置RDS MySQL白名单。
    将ECS的私有IP添加到RDS MySQL的白名单中,详情请参见设置白名单
    注意 如果RDS MySQL与ECS不在同一VPC下,需要将ECS的公网IP添加到RDS MySQL的白名单中。
  8. 使用如下命令连接RDS MySQL数据库。
    mysql -h<主机名> -P<端口> -u<用户名> -p<密码> -D<数据库>
    变量 说明
    <主机名> RDS MySQL实例的内网地址。可在实例的基本信息页面获取。
    注意 如果RDS MySQL与ECS不在同一VPC下,需要使用RDS MySQL的公网地址。
    <端口> RDS MySQL实例的内网端口,一般为3306。可在实例的基本信息页面获取。
    <用户名> RDS MySQL数据库的账号,可在实例的账号管理页面查看。如果还没有账号,请先创建一个数据库账号,详情请参见创建数据库和账号
    <数据库> RDS MySQL数据库的名称,可在实例的数据库管理页面查看。如果还没有数据库,请首先创建一个数据库,详情请参见创建数据库和账号
    <密码> 登录数据库的密码。
    命令示例如下。
    mysql -hrm-bp1u1xxxxxxxxx6ph.mysql.rds.aliyuncs.com -P3306 -ues -pmima -Delasticsearch
    连接成功后,结果如下所示。连接RDS MySQL
    注意 使用Canal需要开启MySQL binlog模式,阿里云RDS MySQL默认开启,也可以执行以下命令检查binlog是否已经启动。
    show variables like '%log_bin%';

步骤四:安装JDK

  1. 在阿里云ECS实例中,查看可用的JDK软件包列表。
    yum search java | grep -i --color JDK
  2. 选择合适的版本,安装JDK。
    本文选择java-1.8.0-openjdk-devel.x86_64
    yum install java-1.8.0-openjdk-devel.x86_64
  3. 配置环境变量。
    1. 打开etc文件夹下的profile文件。
      vi /etc/profile
    2. 在文件内添加如下的环境变量。
      export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.71-2.b15.el7_2.x86_64
      export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
      export PATH=$PATH:$JAVA_HOME/bin
    3. 使用:wq保存文件并退出vi模式,执行以下命令使配置生效。
      source /etc/profile
  4. 分别执行以下命令,验证JDK是否安装成功。
    • java
    • javac
    • java -version
    显示如下结果说明JDK安装成功。jdk安装成功

步骤五:安装并启动Canal-server

  1. 下载并解压Canal-deployer。
    本文使用Canal-deployer 1.1.4版本。
    wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz
  2. 解压canal.deployer-1.1.4.tar.gz
    tar -zxvf canal.deployer-1.1.4.tar.gz
  3. 修改conf/example/instance.properties文件。
    vi conf/example/instance.properties
    修改conf/example/instance.properties文件
    配置项 说明
    canal.instance.master.address <RDS MySQL数据库的内网地址>:<内网端口>,相关信息可在RDS MySQL实例的基本信息页面获取。例如rm-bp1u1xxxxxxxxx6ph.mysql.rds.aliyuncs.com:3306
    canal.instance.dbUsername RDS MySQL数据库的账号名称,可在实例的账号管理页面获取。
    canal.instance.dbPassword RDS MySQL数据库的密码。
  4. 使用:wq命令保存文件并退出vi模式。
  5. 启动Canal-server,并查看日志。
    ./bin/startup.sh
    cat logs/canal/canal.log
    启动canal-server

步骤六:安装并启动Canal-adapter

  1. 下载并解压Canal-adapter。
    本文使用Canal-adapter1.1.4版本。
    wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.adapter-1.1.4.tar.gz
  2. 解压canal.adapter-1.1.4.tar.gz
    tar -zxvf canal.adapter-1.1.4.tar.gz
  3. 修改conf/application.yml文件。
    vi conf/application.yml
    修改conf/application.yml文件
    配置项 说明
    canal.conf.canalServerHost canalDeployer访问地址。保持默认(127.0.0.1:11111)即可。
    canal.conf.srcDataSources.defaultDS.url jdbc:mysql://<RDS MySQL内网地址>:<内网端口>/<数据库名称>?useUnicode=true,相关信息可在RDS MySQL实例的基本信息页面获取。例如jdbc:mysql://rm-bp1xxxxxxxxxnd6ph.mysql.rds.aliyuncs.com:3306/elasticsearch?useUnicode=true
    canal.conf.srcDataSources.defaultDS.username RDS MySQL数据库的账号名称,可在RDS MySQL实例的账号管理页面获取。
    canal.conf.srcDataSources.defaultDS.password RDS MySQL数据库的密码。
    canal.conf.canalAdapters.groups.outerAdapters.hosts 定位到name:es的位置,将hosts替换为<阿里云ES实例的内网地址>:<内网端口>,相关信息可在ES实例的基本信息页面获取。例如es-cn-v64xxxxxxxxx3medp.elasticsearch.aliyuncs.com:9200
    canal.conf.canalAdapters.groups.outerAdapters.mode 必须设置为rest
    canal.conf.canalAdapters.groups.outerAdapters.properties.security.auth <阿里云ES实例的账号>:<密码>。例如elastic:es_password
    canal.conf.canalAdapters.groups.outerAdapters.properties.cluster.name 阿里云ES实例的ID,可在实例的基本信息页面获取。例如es-cn-v64xxxxxxxxx3medp
  4. 使用:wq命令保存文件并退出vi模式。
  5. 同样的方式,修改conf/es/*.yml文件,定义MySQL数据到ES数据的映射字段。
    修改conf/es/*.yml文件
    配置项 说明
    esMapping._index 步骤二:创建索引和mapping章节中,在ES实例中所创建的索引的名称。本文使用es_test
    esMapping._type 步骤二:创建索引和mapping章节中,在ES实例中所创建的索引的类型。本文使用_doc
    esMapping._id 需要同步到ES实例的文档的id,可自定义。本文使用_id
    esMapping.sql SQL语句,用来查询需要同步到ES中的字段。本文使用select t.id as _id,t.id,t.count,t.name,t.color from es_test t
  6. 启动Canal-adapter服务,并查看日志。
    ./bin/startup.sh
    cat logs/adapter/adapter.log
    服务启动正常时,结果如下所示。Canal-adapter服务日志

步骤七:查看Canal同步的增量数据

  1. 在RDS MySQL数据库中,新增、修改或删除数据库中es_test表的数据。
    insert `elasticsearch`.`es_test`(`count`,`id`,`name`,`color`) values('11',2,'canal_test2','red');
  2. 登录目标阿里云ES实例的Kibana控制台。
    登录控制台的具体步骤请参见登录Kibana控制台
  3. Dev Tools(开发工具)页面的Console中,执行以下命令查询同步成功的数据。
    GET /es_test/_search
    数据同步成功后,返回如下结果。数据同步成功结果