canal es适配器

旧城等待， 2022-10-18 11:30 156阅读 0赞

--------------------

## canal es适配器 ##

**官网：[https://github.com/alibaba/canal/wiki/Sync-ES][https_github.com_alibaba_canal_wiki_Sync-ES]**

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**配置文件**

**application.yml：启动器配置文件**

server:
      port: 8081
    spring:
      jackson:
        date-format: yyyy-MM-dd HH:mm:ss
        time-zone: GMT+8
        default-property-inclusion: non_null
    
    canal.conf:
      mode: tcp #tcp kafka rocketMQ rabbitMQ
      flatMessage: true
      zookeeperHosts:
      syncBatchSize: 1000
      retries: 0
      timeout:
      accessKey:
      secretKey:
      consumerProperties:
        # canal tcp consumer
        canal.tcp.server.host: 127.0.0.1:11111
        canal.tcp.zookeeper.hosts:
        canal.tcp.batch.size: 500
        canal.tcp.username:
        canal.tcp.password:
        # kafka consumer
        kafka.bootstrap.servers: 127.0.0.1:9092
        kafka.enable.auto.commit: false
        kafka.auto.commit.interval.ms: 1000
        kafka.auto.offset.reset: latest
        kafka.request.timeout.ms: 40000
        kafka.session.timeout.ms: 30000
        kafka.isolation.level: read_committed
        kafka.max.poll.records: 1000
        # rocketMQ consumer
        rocketmq.namespace:
        rocketmq.namesrv.addr: 127.0.0.1:9876
        rocketmq.batch.size: 1000
        rocketmq.enable.message.trace: false
        rocketmq.customized.trace.topic:
        rocketmq.access.channel:
        rocketmq.subscribe.filter:
        # rabbitMQ consumer
        rabbitmq.host:
        rabbitmq.virtual.host:
        rabbitmq.username:
        rabbitmq.password:
        rabbitmq.resource.ownerId:
    
      srcDataSources:
        defaultDS:
          url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true
          username: root
          password: 121212
      canalAdapters:
      - instance: example # canal instance Name or mq topic name
        groups:
        - groupId: g1
          outerAdapters:
          - name: es7             # es6、es7
            hosts: 127.0.0.1:9300 # 127.0.0.1:9200 for rest mode
            properties:
              mode: transport # or rest
              # security.auth: test:123456 #  only used for rest mode
              cluster.name: elasticsearch

**mytest\_user.yml：canal adapter会自动加载conf/es下的所有以.yml结尾的适配器配置文件**

dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: mytest_user
      _id: _id
    #  upsert: true
    #  pk: id
      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"
    #  objFields:
    #    _labels: array:;
      etlCondition: "where a.c_time>={}"
      commitBatch: 3000

**custom.yml：父子文档父文档配置**

dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: customer
      _id: id
      relations:
        customer_order:
          name: customer
      sql: "select t.id, t.name, t.email from customer t"
      etlCondition: "where t.c_time>={}"
      commitBatch: 3000
    
    
    #{
    #  "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"
    #          }
    #        }
    #      }
    #    }
    #  }
    #}

**biz\_order.yml：父子文档子文档配置**

dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: customer
      _id: _id
      relations:
        customer_order:
          name: order
          parent: customer_id
      sql: "select concat('oid_', t.id) as _id,
            t.customer_id,
            t.id as order_id,
            t.serial_code as order_serial,
            t.c_time as order_time
            from biz_order t"
      skips:
        - customer_id
      etlCondition: "where t.c_time>={}"
      commitBatch: 3000

**sql 映射说明**

左表不能为子查询语句
    左表只能是主表，且只能使用left outer join语句
    关联的从表如果是子查询，则从表只能使用一张表，不能使用多张表
    
    主sql中不能有where
    从表如果是子查询，则从表可以有where，但不推荐，可能会造成数据不一致（如修改where条件中的字段内容）
    
    关联条件只允许主外键=操作，不能出现其他符号、常量
    如 on a.role_id=b.id and b.statues = 1
    
    关联语句中的字段，至少有一个出现在查询结果中
    如：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 use
    id ==> _id
    name ==> name
    email ==> _email

**sql 示例**

单表索引：select a.id as _id, a.name, a.role_id, a.c_time from user a
    
    单表索引（带函数、运算）：
    select a.id as _id, concat(a.name,'_test') as name, 
    a.role_id+10000 as role_id, a.c_time from user a
    
    
    多表映射：一对一、多对一
    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
    
    
    多表映射：多对多
    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
    
    # labels使用数组存储，适配器配置文件中对应字段设置
    objFields:
      labels: array:;       # array:开头，表示字段类型为数组，; 表示数组元素用 ; 间隔
    
    
    geotype
    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, a.description from user a
    
    # description 适配器文件对应字段设置
    objFields:
      description: object
    
    
    ****************
    父子文档索引
    
    父文档适配器配置文件
    esMapping:
      _index: customer
      _type: _doc
      _id: id
      relations:
        customer_order:
          name: customer
      sql: "select t.id, t.name, t.email from customer t"
    
    
    子文档适配器配置文件
    esMapping:
      _index: customer
      _type: _doc
      _id: _id
      relations:
        customer_order:
          name: order
          parent: customer_id
      sql: "select concat('oid_', t.id) as _id,
            t.customer_id,
            t.id as order_id,
            t.serial_code as order_serial,
            t.c_time as order_time
            from biz_order t"
      skips:
        - customer_id
    
    
    mapping 示例（字段映射）
    {
      "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"
              }
            }
          }
        }
      }
    }

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**相关类与接口**

**ESSyncConfig**

public class ESSyncConfig implements AdapterConfig {
    
        private String    dataSourceKey;    // 数据源key
    
        private String    outerAdapterKey;  // adapter key
    
        private String    groupId;          // group id
    
        private String    destination;      // canal destination
    
        private ESMapping esMapping;
    
        private String    esVersion = "es6";
    
        public void validate() {
            if (esMapping._index == null) {
                throw new NullPointerException("esMapping._index");
            }   //_index 不能为null
    
            if ("es6".equals(esVersion) && esMapping._type == null) {
                throw new NullPointerException("esMapping._type");
            }   //如果是es6，则type不能为null
    
            if (esMapping._id == null && esMapping.getPk() == null) {
                throw new NullPointerException("esMapping._id or esMapping.pk");
            }   //_id、pk不能同时为空
    
            if (esMapping.sql == null) {
                throw new NullPointerException("esMapping.sql");
            }   //sql不能为空
        }
    
        。。。
    
    
    ***************
    内部类：ESMapping
    
        public static class ESMapping implements AdapterMapping {
    
            private String                       _index;
            private String                       _type;
            private String                       _id;
            private boolean                      upsert          = false;
            private String                       pk;
            private Map<String, RelationMapping> relations       = new LinkedHashMap<>();
            private String                       sql;
            // 对象字段, 例: objFields:
            // - _labels: array:;
            private Map<String, String>          objFields       = new LinkedHashMap<>();
            private List<String>                 skips           = new ArrayList<>();
            private int                          commitBatch     = 1000;
            private String                       etlCondition;
            private boolean                      syncByTimestamp = false;                // 是否按时间戳定时同步
            private Long                         syncInterval;                           // 同步时间间隔
    
            private SchemaItem                   schemaItem;                             // sql解析结果模型
    
    
    ***************
    内部类：RelationMapping
    
        public static class RelationMapping {
    
            private String name;      //文档是父文档、还是子文档
            private String parent;    //父文档id
    
    
    ***********
    示例
    
            "customer_order":{        #字段映射，custom_order为join类型
              "type":"join",
              "relations":{
                "customer":"order"    #custom_order的值为customer ==> 父文档
                                      #custom_order的值为order    ==> 子文档
              }
            }
    
    
    # 父文档适配器配置文件
    esMapping:
      _index: customer
      _type: _doc
      _id: id
      relations:
        customer_order:
          name: customer
      sql: "select t.id, t.name, t.email from customer t"
    
    
    # 子文档适配器配置文件
    esMapping:
      _index: customer
      _type: _doc
      _id: _id
      relations:
        customer_order:
          name: order             # 子文档
          parent: customer_id     # 子文档的parent id为 custom_id
      sql: "select concat('oid_', t.id) as _id,
            t.customer_id,
            t.id as order_id,
            t.serial_code as order_serial,
            t.c_time as order_time
            from biz_order t"
      skips:
        - customer_id              # 跳过custom_id，该字段不索引

**ESSyncUtil：es同步工具类**

public class ESSyncUtil {
    
        private static Logger logger = LoggerFactory.getLogger(ESSyncUtil.class);
    
        public static Object convertToEsObj(Object val, String fieldInfo) {
                             //将字段转换为对应的对象类型
            if (val == null) {
                return null;
            }
    
            if (fieldInfo.startsWith("array:")) {
                          //如果fieldInfo以array:开头，则对应的字段类型为数组
                String separator = fieldInfo.substring("array:".length()).trim();
                String[] values = val.toString().split(separator);
                return Arrays.asList(values);
            } else if (fieldInfo.startsWith("object")) {
                          //如果fieldInfo以object开头，则对应的字段类型为json对象
                if (val instanceof String){
                    return JSON.parse(val.toString());
                }
    
                return JSON.parse(new String((byte[])val));
            }
            return null;
        }
    
    
        。。。

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**示例**

**创建 mysql实例**

docker run -it -d --net fixed --ip 172.18.0.2 -p 3306:3306 --privileged=true \
    --name mysql -e  MYSQL_ROOT_PASSWORD=123456 mysql
    
    # 新建用户、并授权
    mysql> create user canal identified with mysql_native_password by "123456";
    Query OK, 0 rows affected (0.15 sec)
    
    mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    Query OK, 0 rows affected (0.00 sec)
    
    mysql> flush privileges;
    Query OK, 0 rows affected (0.01 sec)
    
    
    # 创建数据库、表
    mysql> create database lihu;
    Query OK, 1 row affected (0.26 sec)
    
    mysql> use lihu;
    Database changed
    
    mysql> create table person(id int not null auto_increment primary key,name varchar(20) not null,age int not null);
    Query OK, 0 rows affected (0.11 sec)

**创建 canal server实例**

docker run -it -d --net fixed --ip 172.18.0.3 \
    -p 11111:11111 --name canal-server \
    -e canal.instance.master.address=172.18.0.2:3306  \
    -e canal.instance.dbUsername=canal \
    -e canal.instance.dbPassword=123456 canal/canal-server

**创建 canal adapter实例**

docker run -it -d --net fixed --ip 172.18.0.4 -p 8081:8081 \
    -v /usr/canal/canal-adapter/es/application.yml:/opt/canal-adapter/conf/application.yml \
    -v /usr/canal/canal-adapter/es/adapter:/opt/canal-adapter/conf/es7 \
    --name canal-adapter slpcat/canal-adapter:v1.1.5-jdk8
    
    
    *************
    application.yml：启动器配置文件
    
    server:
      port: 8081
    spring:
      jackson:
        date-format: yyyy-MM-dd HH:mm:ss
        time-zone: GMT+8
        default-property-inclusion: non_null
    
    canal.conf:
      mode: tcp #tcp kafka rocketMQ rabbitMQ
      flatMessage: true
      zookeeperHosts:
      syncBatchSize: 1000
      retries: 0
      timeout:
      accessKey:
      secretKey:
      consumerProperties:
        # canal tcp consumer
        canal.tcp.server.host: 172.18.0.3:11111
        canal.tcp.zookeeper.hosts:
        canal.tcp.batch.size: 500
        canal.tcp.username:
        canal.tcp.password:
    
      srcDataSources:
        defaultDS:
          url: jdbc:mysql://172.18.0.2:3306/lihu?useUnicode=true
          username: root
          password: 123456
      canalAdapters:
      - instance: example # canal instance Name or mq topic name
        groups:
        - groupId: g1
          outerAdapters:
          - name: es7
            hosts: 172.18.0.5:9300 # 127.0.0.1:9200 for rest mode
            properties:
              mode: transport # or rest
              # security.auth: test:123456 #  only used for rest mode
              cluster.name: elasticsearch
    
    
    *************
    mytest_user.yml：适配器配置文件
    
    dataSourceKey: defaultDS
    destination: example
    groupId: g1
    esMapping:
      _index: person
      _id: _id
    #  upsert: true
    #  pk: id
      sql: "select id as _id, name,age from person "
    #  objFields:
    #    _labels: array:;
      etlCondition: "where a.c_time>={}"
      commitBatch: 3000

**创建 elasticsearch实例**

docker run -it -d --net fixed --ip 172.18.0.5 -p 9200:9200 -p 9300:9300 \
    -v /usr/canal/canal-adapter/es/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml \
    --name es-single docker.elastic.co/elasticsearch/elasticsearch:7.13.3
    
    
    # 创建 person index
    [root@centos adapter]# curl -X PUT localhost:9200/person?pretty
    {
      "acknowledged" : true,
      "shards_acknowledged" : true,
      "index" : "person"
    }
    
    # 创建 mapping
    [root@centos es]# curl -X PUT -H "Content-Type: application/json" localhost:9200/person/_mapping?pretty \
    > -d '{"properties":{"name":{"type":"text"},"age":{"type":"integer"}}}'
    {
      "acknowledged" : true
    }

**说明：如果不创建mapping，同步数据时则会报错**

2021-07-10 22:31:34.572 [pool-1-thread-1] ERROR c.a.o.canal.client.adapter.es.core.service.ESSyncService - sync error, es index: person, DML : Dml{destination='example', database='lihu', table='person', type='INSERT', es=1625927494000, ts=1625927494517, sql='', data=[{id=2, name=lihu, age=2}], old=null}
    2021-07-10 22:31:34.573 [pool-1-thread-1] ERROR c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - java.lang.IllegalArgumentException: Not found the mapping info of index: person
    java.lang.RuntimeException: java.lang.IllegalArgumentException: Not found the mapping info of index: person
    	at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.sync(ESSyncService.java:116) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.sync(ESSyncService.java:64) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es.core.ESAdapter.sync(ESAdapter.java:115) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es.core.ESAdapter.sync(ESAdapter.java:94) ~[na:na]
    	at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.batchSync(AdapterProcessor.java:139) ~[client-adapter.launcher-1.1.5-SNAPSHOT.jar:na]
    	at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.lambda$null$1(AdapterProcessor.java:97) ~[client-adapter.launcher-1.1.5-SNAPSHOT.jar:na]
    	at java.util.concurrent.CopyOnWriteArrayList.forEach(CopyOnWriteArrayList.java:895) ~[na:1.8.0_282]
    	at com.alibaba.otter.canal.adapter.launcher.loader.AdapterProcessor.lambda$null$2(AdapterProcessor.java:94) ~[client-adapter.launcher-1.1.5-SNAPSHOT.jar:na]
    	at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_282]
    	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_282]
    	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_282]
    	at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_282]
    Caused by: java.lang.IllegalArgumentException: Not found the mapping info of index: person
    	at com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate.getEsType(ES7xTemplate.java:395) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate.getValFromData(ES7xTemplate.java:273) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate.getESDataFromDmlData(ES7xTemplate.java:297) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.singleTableSimpleFiledInsert(ESSyncService.java:439) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.insert(ESSyncService.java:139) ~[na:na]
    	at com.alibaba.otter.canal.client.adapter.es.core.service.ESSyncService.sync(ESSyncService.java:99) ~[na:na]
    	... 11 common frames omitted
    2021-07-10 22:31:34.574 [Thread-4] ERROR c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - Outer adapter sync failed!  Error sync but ACK!

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**使用测试**

**mysql 数据库**

mysql> insert into person(id,name,age) values(3,'lihu',2);
    Query OK, 1 row affected (0.01 sec)

**canal adapter 日志**

2021-07-10 22:27:15.845 [main] INFO  c.a.o.c.client.adapter.es.core.config.ESSyncConfigLoader - ## Start loading es mapping config ... 
    2021-07-10 22:27:16.034 [main] INFO  c.a.o.c.client.adapter.es.core.config.ESSyncConfigLoader - ## ES mapping config loaded
    2021-07-10 22:27:16.764 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: es7 succeed
    2021-07-10 22:27:16.822 [main] INFO  c.alibaba.otter.canal.connector.core.spi.ExtensionLoader - extension classpath dir: /opt/canal-adapter/plugin
    2021-07-10 22:27:17.027 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Start adapter for canal-client mq topic: example-g1 succeed
    2021-07-10 22:27:17.028 [main] INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterService - ## the canal client adapters are running now ......
    2021-07-10 22:27:17.034 [main] INFO  org.apache.coyote.http11.Http11NioProtocol - Starting ProtocolHandler ["http-nio-8081"]
    2021-07-10 22:27:17.072 [Thread-4] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Start to connect destination: example <=============
    2021-07-10 22:27:17.136 [main] INFO  org.apache.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
    2021-07-10 22:27:17.334 [main] INFO  o.s.boot.web.embedded.tomcat.TomcatWebServer - Tomcat started on port(s): 8081 (http) with context path ''
    2021-07-10 22:27:17.373 [main] INFO  c.a.otter.canal.adapter.launcher.CanalAdapterApplication - Started CanalAdapterApplication in 31.579 seconds (JVM running for 34.05)
    2021-07-10 22:27:20.104 [Thread-4] INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =============> Subscribe destination: example succeed <=============
    
    # 插入数据："data":[{"id":3,"name":"lihu","age":2}
    2021-07-10 22:32:33.254 [pool-1-thread-1] DEBUG c.a.o.canal.client.adapter.es.core.service.ESSyncService - DML: {"data":[{"id":3,"name":"lihu","age":2}],"database":"lihu","destination":"example","es":1625927552000,"groupId":"g1","isDdl":false,"old":null,"pkNames":["id"],"sql":"","table":"person","ts":1625927553029,"type":"INSERT"} 
    
    # 数据插入成功
    Affected indexes: person

[https_github.com_alibaba_canal_wiki_Sync-ES]: https://github.com/alibaba/canal/wiki/Sync-ES