{
    "job": {
        "setting": {
            "speed": {
                 "channel":1
            }
        },
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "xxx",
                        "password": "xxx",
                        "connection": [
                            {
                                "querySql": [
                                    "select u.user_id as _id,ur.role_id as role_id,r.id as r_id,u.username as username,u.real_name as real_name,r.name as role_name from sys_user u left join sys_user_role ur on u.user_id = ur.user_id left join sys_role r on ur.role_id = r.id"
                                ],
                                "jdbcUrl": [
                                    "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8"
                                ]
                            }
                        ]
                    }
                },
                "writer": {
                     "name": "elasticsearchwriter",
          "parameter": {
            "endpoint": "http://es-v.elasticsearch.aliyuncs.com:9200",
            "accessId": "elastic",
            "accessKey": "xxxx",
            "index": "kefu_user",
            "type": "user_role",
            "cleanup": true,
            "settings": {"index" :{"number_of_shards": 3, "number_of_replicas": 1}},
            "discovery": false,
            "batchSize": 1000,
            "splitter": ",",
            "column": [
              {"name": "_id", "type": "id"},
              {"name": "r_id", "type": "long"},
              {"name": "role_id", "type": "long"},
              { "name": "username","type": "keyword" },
              { "name": "real_name","type": "keyword" },
              { "name": "role_name","type": "keyword" }
            ]
                    }
                }
            }
        ]
    }
}

# position info
canal.instance.master.address=localhost:3306
canal.instance.dbUsername=xxx
canal.instance.dbPassword=xxx
canal.instance.connectionCharset = UTF-8

server:
  port: 8081
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  # 对应单机模式下的canal server的ip:port
  batchSize: 500 										# 每次获取数据的批大小, 单位为K
  syncBatchSize: 1000								# 每次同步的批数量
  retries: 0
  timeout:
  mode: tcp # kafka rocketMQ        # canal client的模式: tcp kafka rocketMQ
  srcDataSources:                   # 源数据库
    defaultDS:                      # 自定义名称
      url: jdbc:mysql://localhost:3306/test?useUnicode=true  # jdbc url 
      username: xxx
      password: xxx
  canalAdapters:
  - instance: example 
    groups:
    - groupId: g1
      outerAdapters:
      - 
        key: exampleKey                    # canal 实例名或者 MQ topic 名
        name: es                           # or es7
        hosts: es-cn-v.elasticsearch.aliyuncs.com:9200      # es 集群地址, 逗号分隔
        properties:
          mode: rest  # or rest         # 可指定transport模式或者rest模式
          security.auth: xxx:aaaaaa      # only used for rest mode
          cluster.name: elasticsearch       # es cluster name

dataSourceKey: defaultDS        # 源数据源的key, 对应上面配置的srcDataSources中的值
outerAdapterKey: exampleKey     # 对应application.yml中es配置的key 
destination: example            # cannal的instance或者MQ的topic
groupId:                        # 对应MQ模式下的groupId, 只会同步对应groupId的数据
esMapping:
  _index: xx_user           # es 的索引名称
  _type: user_role                   # es 的type名称, es7下无需配置此项
  _id: _id
  upsert: true                      # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
  # pk: _id                       # 如果不需要_id, 则需要指定一个属性为主键属性
  # sql映射
  sql: "select u.user_id as _id,ur.role_id as role_id,r.id as r_id,u.username as username,u.real_name as real_name,r.name as role_name from sys_user u left join sys_user_role ur on u.user_id = ur.user_id left join sys_role r on ur.role_id = r.id"
#  objFields:
#    _labels: array:;           # 数组或者对象属性, array:; 代表以;字段里面是以;分隔的
#    _obj: object               # json对象
  # etlCondition: "where a.c_time>='{0}'"     # etl 的条件参数
  commitBatch: 3000

package com.alibaba.otter.canal.instance.spring.support;

import java.beans.PropertyEditorSupport;
import java.net.InetSocketAddress;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.PropertyEditorRegistrar;
import org.springframework.beans.PropertyEditorRegistry;

public class SocketAddressEditor extends PropertyEditorSupport implements PropertyEditorRegistrar {

    public void registerCustomEditors(PropertyEditorRegistry registry) {
        registry.registerCustomEditor(InetSocketAddress.class, this);
    }

    public void setAsText(String text) throws IllegalArgumentException {
        String[] addresses = StringUtils.split(text, ":");
        if (addresses.length > 0) {
            if (addresses.length != 2) {
                throw new RuntimeException("address[" + text + "] is illegal, eg.127.0.0.1:3306");
            } else {
                setValue(new InetSocketAddress(addresses[0], Integer.valueOf(addresses[1])));
            }
        } else {
            setValue(null);
        }
    }
}

{"succeeded":false,"errorMessage":"ES 数据导入异常 =>java.sql.SQLException: Parameter index out of range (1 > number of parameters, which is 0)."}