备注：该代码基于canal1.1.5
    client-adapter：
    resources下es7配置文件：
    添加自定义配置：
    indexSuffixfield: created_at #db table 创建时间
    indexSuffixPattern: yyyyMM # 时间需要对应ES索引后缀格式
    示例：

    1. dataSourceKey: defaultDS
    2. destination: example
    3. groupId: g1
    4. esMapping:
    5.   _index: payments
    6.   _id: _id
    7.   upsert: true
    8.   sql: "select t.payment_id as _id, t.payment_id, t.created_at,t.deleted_at,t.root_mch_id,t.agent_mch_id,t.bank_id,t.root_bank_id,t.cup_org_id,t.pay_balance_acct_id,t.amount,t.out_order_no,t.remark,t.fee,t.is_refunded,t.refunded_amount,t.extra,t.metadata,t.req_id,t.currency,t.recv_balance_acct_id,t.payment_status,t.finished_at,t.sent_at,t.version,t.reason,t.batch_payment_id from payments t"
    9.   etlCondition: "where t.created_at>={}"
    10.   commitBatch: 3000
    11.   indexSuffixfield: created_at
    12.   indexSuffixPattern: yyyyMM

    创建自定义索引扩展帮助类：

    1. package com.alibaba.otter.canal.client.adapter.es.core.support;
    3. import com.alibaba.fastjson.JSONObject;
    4. import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
    5. import org.slf4j.Logger;
    6. import org.slf4j.LoggerFactory;
    7. import org.springframework.util.CollectionUtils;
    8. import org.springframework.util.StringUtils;
    10. import java.text.SimpleDateFormat;
    11. import java.util.Map;
    13. public class ESMapppingPlus {
    15.     private static Logger logger = LoggerFactory.getLogger(ESMapppingPlus.class);
    17.     /**
    18.      * 重构索引
    19.      * @param esMapping
    20.      * @param esFieldData
    21.      */
    22.     public static void rebuildIndex(ESSyncConfig.ESMapping esMapping, Map<String, Object> esFieldData) {
    23.         if (CollectionUtils.isEmpty(esFieldData)){
    24.             return;
    25.         }
    26.         String indexSuffixfield = esMapping.getIndexSuffixfield();
    27.         if (StringUtils.isEmpty(indexSuffixfield)) {
    28.             //索引后缀字段不存在 不带后缀
    29.             esMapping.set_index(esMapping.getIndexPrefix());
    30.             return;
    31.         }
    32.         String indexSuffix = esFieldData.containsKey(indexSuffixfield) ? esFieldData.get(indexSuffixfield).toString() : null;
    33.         if (StringUtils.isEmpty(indexSuffix)) {
    34.             //未匹配到索引后缀字段
    35.             logger.error("esMapping.indexSuffixfield not matched to dml field, index:{} ,indexSuffixfield:{}, sql:{}, esFieldData:{}",
    36.                     esMapping.getIndexPrefix(), indexSuffixfield, esMapping.getSql(), JSONObject.toJSONString(esFieldData));
    37.             throw new RuntimeException("esMapping.indexSuffixfield not matched to dml field");
    38.         }
    40.         String indexSuffixPattern = esMapping.getIndexSuffixPattern();
    42.         if (indexSuffix.indexOf("-") != -1) {
    43.             //时间字符串格式
    44.             indexSuffix = datePattern(indexSuffix, indexSuffixPattern);
    45.             String index = esMapping.getIndexPrefix()+"-"+ indexSuffix;
    46.             esMapping.set_index(index);
    47.             return;
    48.         } else if(indexSuffix.length()==13){
    49.             //时间戳格式
    50.             SimpleDateFormat formatter = new SimpleDateFormat(indexSuffixPattern);
    51.             indexSuffix = formatter.format(indexSuffix);
    52.             String index = esMapping.getIndexPrefix()+"-"+ indexSuffix;
    53.             esMapping.set_index(index);
    54.             return;
    55.         } else {
    56.             //未匹配到索引后缀字段
    57.             logger.error("esMapping.indexSuffixPattern can't null, index:{}",
    58.                     esMapping.getIndexPrefix());
    59.             throw new RuntimeException("esMapping.indexSuffixPattern can't null");
    60.         }
    61.     }
    64.     /**
    65.      * 重构索引
    66.      * @param config
    67.      * @param esFieldData
    68.      */
    69.     public static void rebuildIndex(ESSyncConfig config, Map<String, Object> esFieldData) {
    70.         rebuildIndex(config.getEsMapping(), esFieldData);
    71.     }
    76.     private static String datePattern(String date, String pattern) {
    77.         switch (pattern) {
    78.             case "yyyy" :
    79.                 return date.substring(0,4);
    80.             case "yyyyMM":
    81.                 return date.substring(0,7).replaceAll("-", "");
    82.             case "yyyyMMdd":
    83.                 return date.substring(0,10).replaceAll("-", "");
    84.             case "yyyy-MM":
    85.                 return date.substring(0,7);
    86.             case "yyyy-MM-dd":
    87.                 return date.substring(0,10);
    88.             default:
    89.                 return date.substring(0,7).replaceAll("-", "");
    90.         }
    91.     }
    92. }

    ES7xTemplate.java添加

    1. ESMapppingPlus.rebuildIndex(mapping, esFieldData);

    �或者添加：

    1. ESMapppingPlus.rebuildIndex(config, esFieldData);

    修改后的完整代码：

    1. package com.alibaba.otter.canal.client.adapter.es7x.support;
    3. import java.sql.ResultSet;
    4. import java.sql.SQLException;
    5. import java.util.ArrayList;
    6. import java.util.HashMap;
    7. import java.util.LinkedHashMap;
    8. import java.util.List;
    9. import java.util.Map;
    10. import java.util.concurrent.ConcurrentHashMap;
    11. import java.util.concurrent.ConcurrentMap;
    13. import javax.sql.DataSource;
    15. import com.alibaba.otter.canal.client.adapter.es.core.support.ESMapppingPlus;
    16. import org.apache.commons.lang.StringUtils;
    17. import org.elasticsearch.action.search.SearchResponse;
    18. import org.elasticsearch.cluster.metadata.MappingMetaData;
    19. import org.elasticsearch.index.query.BoolQueryBuilder;
    20. import org.elasticsearch.index.query.QueryBuilders;
    21. import org.elasticsearch.search.SearchHit;
    22. import org.slf4j.Logger;
    23. import org.slf4j.LoggerFactory;
    25. import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
    26. import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
    27. import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem;
    28. import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem;
    29. import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem;
    30. import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest;
    31. import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESBulkResponse;
    32. import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESDeleteRequest;
    33. import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESIndexRequest;
    34. import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESUpdateRequest;
    35. import com.alibaba.otter.canal.client.adapter.es.core.support.ESSyncUtil;
    36. import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate;
    37. import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection.ESSearchRequest;
    38. import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
    39. import com.alibaba.otter.canal.client.adapter.support.Util;
    41. public class ES7xTemplate implements ESTemplate {
    43.     private static final Logger                               logger         = LoggerFactory
    44.         .getLogger(ESTemplate.class);
    46.     private static final int                                  MAX_BATCH_SIZE = 1000;
    48.     private ESConnection                                      esConnection;
    50.     private ESBulkRequest                                     esBulkRequest;
    52.     // es 字段类型本地缓存
    53.     private static ConcurrentMap<String, Map<String, String>> esFieldTypes   = new ConcurrentHashMap<>();
    55.     public ES7xTemplate(ESConnection esConnection){
    56.         this.esConnection = esConnection;
    57.         this.esBulkRequest = this.esConnection.new ES7xBulkRequest();
    58.     }
    60.     public ESBulkRequest getBulk() {
    61.         return esBulkRequest;
    62.     }
    64.     public void resetBulkRequestBuilder() {
    65.         this.esBulkRequest.resetBulk();
    66.     }
    68.     @Override
    69.     public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
    70.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    71.         if (mapping.get_id() != null) {
    72.             String parentVal = (String) esFieldData.remove("$parent_routing");
    73.             if (mapping.isUpsert()) {
    74.                 ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.get_index(),
    75.                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
    76.                 if (StringUtils.isNotEmpty(parentVal)) {
    77.                     updateRequest.setRouting(parentVal);
    78.                 }
    79.                 getBulk().add(updateRequest);
    80.             } else {
    81.                 ESIndexRequest indexRequest = esConnection.new ES7xIndexRequest(mapping.get_index(), pkVal.toString())
    82.                     .setSource(esFieldData);
    83.                 if (StringUtils.isNotEmpty(parentVal)) {
    84.                     indexRequest.setRouting(parentVal);
    85.                 }
    86.                 getBulk().add(indexRequest);
    87.             }
    88.             commitBulk();
    89.         } else {
    90.             ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
    91.                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
    92.                 .size(10000);
    93.             SearchResponse response = esSearchRequest.getResponse();
    95.             for (SearchHit hit : response.getHits()) {
    96.                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
    97.                     hit.getId()).setDoc(esFieldData);
    98.                 getBulk().add(esUpdateRequest);
    99.                 commitBulk();
    100.             }
    101.         }
    102.     }
    104.     @Override
    105.     public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
    106.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    108.         Map<String, Object> esFieldDataTmp = new LinkedHashMap<>(esFieldData.size());
    109.         esFieldData.forEach((k, v) -> esFieldDataTmp.put(Util.cleanColumn(k), v));
    110.         append4Update(mapping, pkVal, esFieldDataTmp);
    111.         commitBulk();
    112.     }
    114.     @Override
    115.     public void updateByQuery(ESSyncConfig config, Map<String, Object> paramsTmp, Map<String, Object> esFieldData) {
    116.         ESMapppingPlus.rebuildIndex(config, esFieldData);
    118.         if (paramsTmp.isEmpty()) {
    119.             return;
    120.         }
    121.         ESMapping mapping = config.getEsMapping();
    122.         BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
    123.         paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value)));
    125.         // 查询sql批量更新
    126.         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
    127.         StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE ");
    128.         List<Object> values = new ArrayList<>();
    129.         paramsTmp.forEach((fieldName, value) -> {
    130.             sql.append("_v.").append(fieldName).append("=? AND ");
    131.             values.add(value);
    132.         });
    133.         // TODO 直接外部包裹sql会导致全表扫描性能低, 待优化拼接内部where条件
    134.         int len = sql.length();
    135.         sql.delete(len - 4, len);
    136.         Integer syncCount = (Integer) Util.sqlRS(ds, sql.toString(), values, rs -> {
    137.             int count = 0;
    138.             try {
    139.                 while (rs.next()) {
    140.                     Object idVal = getIdValFromRS(mapping, rs);
    141.                     append4Update(mapping, idVal, esFieldData);
    142.                     commitBulk();
    143.                     count++;
    144.                 }
    145.             } catch (Exception e) {
    146.                 throw new RuntimeException(e);
    147.             }
    148.             return count;
    149.         });
    150.         if (logger.isTraceEnabled()) {
    151.             logger.trace("Update ES by query affected {} records", syncCount);
    152.         }
    153.     }
    155.     @Override
    156.     public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
    157.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    159.         if (mapping.get_id() != null) {
    160.             ESDeleteRequest esDeleteRequest = this.esConnection.new ES7xDeleteRequest(mapping.get_index(),
    161.                 pkVal.toString());
    162.             getBulk().add(esDeleteRequest);
    163.             commitBulk();
    164.         } else {
    165.             ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
    166.                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
    167.                 .size(10000);
    168.             SearchResponse response = esSearchRequest.getResponse();
    169.             for (SearchHit hit : response.getHits()) {
    170.                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
    171.                     hit.getId()).setDoc(esFieldData);
    172.                 getBulk().add(esUpdateRequest);
    173.                 commitBulk();
    174.             }
    175.         }
    176.     }
    178.     @Override
    179.     public void commit() {
    180.         if (getBulk().numberOfActions() > 0) {
    181.             ESBulkResponse response = getBulk().bulk();
    182.             if (response.hasFailures()) {
    183.                 response.processFailBulkResponse("ES sync commit error ");
    184.             }
    185.             resetBulkRequestBuilder();
    186.         }
    187.     }
    189.     @Override
    190.     public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
    191.                                String columnName) throws SQLException {
    192.         fieldName = Util.cleanColumn(fieldName);
    193.         columnName = Util.cleanColumn(columnName);
    194.         String esType = getEsType(mapping, fieldName);
    196.         Object value = resultSet.getObject(columnName);
    197.         if (value instanceof Boolean) {
    198.             if (!"boolean".equals(esType)) {
    199.                 value = resultSet.getByte(columnName);
    200.             }
    201.         }
    203.         // 如果是对象类型
    204.         if (mapping.getObjFields().containsKey(fieldName)) {
    205.             return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName));
    206.         } else {
    207.             return ESSyncUtil.typeConvert(value, esType);
    208.         }
    209.     }
    211.     @Override
    212.     public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet,
    213.                                   Map<String, Object> esFieldData) throws SQLException {
    214.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    216.         SchemaItem schemaItem = mapping.getSchemaItem();
    217.         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
    218.         Object resultIdVal = null;
    219.         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
    220.             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
    222.             if (fieldItem.getFieldName().equals(idFieldName)) {
    223.                 resultIdVal = value;
    224.             }
    226.             if (!fieldItem.getFieldName().equals(mapping.get_id())
    227.                 && !mapping.getSkips().contains(fieldItem.getFieldName())) {
    228.                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
    229.             }
    230.         }
    232.         // 添加父子文档关联信息
    233.         putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
    235.         return resultIdVal;
    236.     }
    238.     @Override
    239.     public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
    240.         SchemaItem schemaItem = mapping.getSchemaItem();
    241.         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
    242.         Object resultIdVal = null;
    243.         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
    244.             Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
    246.             if (fieldItem.getFieldName().equals(idFieldName)) {
    247.                 resultIdVal = value;
    248.                 break;
    249.             }
    250.         }
    251.         return resultIdVal;
    252.     }
    254.     @Override
    255.     public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
    256.                                   Map<String, Object> esFieldData) throws SQLException {
    257.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    259.         SchemaItem schemaItem = mapping.getSchemaItem();
    260.         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
    261.         Object resultIdVal = null;
    262.         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
    263.             if (fieldItem.getFieldName().equals(idFieldName)) {
    264.                 resultIdVal = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
    265.             }
    267.             for (ColumnItem columnItem : fieldItem.getColumnItems()) {
    268.                 if (dmlOld.containsKey(columnItem.getColumnName())
    269.                     && !mapping.getSkips().contains(fieldItem.getFieldName())) {
    270.                     esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
    271.                         getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()));
    272.                     break;
    273.                 }
    274.             }
    275.         }
    277.         // 添加父子文档关联信息
    278.         putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
    280.         return resultIdVal;
    281.     }
    283.     @Override
    284.     public Object getValFromData(ESMapping mapping, Map<String, Object> dmlData, String fieldName, String columnName) {
    285.         String esType = getEsType(mapping, fieldName);
    286.         Object value = dmlData.get(columnName);
    287.         if (value instanceof Byte) {
    288.             if ("boolean".equals(esType)) {
    289.                 value = ((Byte) value).intValue() != 0;
    290.             }
    291.         }
    293.         // 如果是对象类型
    294.         if (mapping.getObjFields().containsKey(fieldName)) {
    295.             return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName));
    296.         } else {
    297.             return ESSyncUtil.typeConvert(value, esType);
    298.         }
    299.     }
    301.     @Override
    302.     public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData,
    303.                                        Map<String, Object> esFieldData) {
    304.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    306.         SchemaItem schemaItem = mapping.getSchemaItem();
    307.         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
    308.         Object resultIdVal = null;
    309.         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
    310.             String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
    311.             Object value = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
    313.             if (fieldItem.getFieldName().equals(idFieldName)) {
    314.                 resultIdVal = value;
    315.             }
    317.             if (!fieldItem.getFieldName().equals(mapping.get_id())
    318.                 && !mapping.getSkips().contains(fieldItem.getFieldName())) {
    319.                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
    320.             }
    321.         }
    323.         // 添加父子文档关联信息
    324.         putRelationData(mapping, schemaItem, dmlData, esFieldData);
    325.         return resultIdVal;
    326.     }
    328.     @Override
    329.     public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> dmlOld,
    330.                                        Map<String, Object> esFieldData) {
    331.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    333.         SchemaItem schemaItem = mapping.getSchemaItem();
    334.         String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
    335.         Object resultIdVal = null;
    336.         for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
    337.             String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
    339.             if (fieldItem.getFieldName().equals(idFieldName)) {
    340.                 resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
    341.             }
    343.             if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) {
    344.                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
    345.                         getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
    346.             }
    347.         }
    349.         // 添加父子文档关联信息
    350.         putRelationData(mapping, schemaItem, dmlOld, esFieldData);
    351.         return resultIdVal;
    352.     }
    354.     /**
    355.      * 如果大于批量数则提交批次
    356.      */
    357.     private void commitBulk() {
    358.         if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {
    359.             commit();
    360.         }
    361.     }
    363.     private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
    364.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    366.         if (mapping.get_id() != null) {
    367.             String parentVal = (String) esFieldData.remove("$parent_routing");
    368.             if (mapping.isUpsert()) {
    369.                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
    370.                     pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
    371.                 if (StringUtils.isNotEmpty(parentVal)) {
    372.                     esUpdateRequest.setRouting(parentVal);
    373.                 }
    374.                 getBulk().add(esUpdateRequest);
    375.             } else {
    376.                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
    377.                     pkVal.toString()).setDoc(esFieldData);
    378.                 if (StringUtils.isNotEmpty(parentVal)) {
    379.                     esUpdateRequest.setRouting(parentVal);
    380.                 }
    381.                 getBulk().add(esUpdateRequest);
    382.             }
    383.         } else {
    384.             ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
    385.                 .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
    386.                 .size(10000);
    387.             SearchResponse response = esSearchRequest.getResponse();
    388.             for (SearchHit hit : response.getHits()) {
    389.                 ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
    390.                     hit.getId()).setDoc(esFieldData);
    391.                 getBulk().add(esUpdateRequest);
    392.             }
    393.         }
    394.     }
    396.     /**
    397.      * 获取es mapping中的属性类型
    398.      *
    399.      * @param mapping mapping配置
    400.      * @param fieldName 属性名
    401.      * @return 类型
    402.      */
    403.     @SuppressWarnings("unchecked")
    404.     private String getEsType(ESMapping mapping, String fieldName) {
    405.         String key = mapping.get_index() + "-" + mapping.get_type();
    406.         Map<String, String> fieldType = esFieldTypes.get(key);
    407.         if (fieldType != null) {
    408.             return fieldType.get(fieldName);
    409.         } else {
    410. //            MappingMetaData mappingMetaData = esConnection.getMapping(mapping.get_index());
    411.             //模版获取
    412.             MappingMetaData mappingMetaData = esConnection.getMapping(mapping.getIndexPrefix());
    414.             if (mappingMetaData == null) {
    415.                 throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
    416.             }
    418.             fieldType = new LinkedHashMap<>();
    420.             Map<String, Object> sourceMap = mappingMetaData.getSourceAsMap();
    421.             Map<String, Object> esMapping = (Map<String, Object>) sourceMap.get("properties");
    422.             for (Map.Entry<String, Object> entry : esMapping.entrySet()) {
    423.                 Map<String, Object> value = (Map<String, Object>) entry.getValue();
    424.                 if (value.containsKey("properties")) {
    425.                     fieldType.put(entry.getKey(), "object");
    426.                 } else {
    427.                     fieldType.put(entry.getKey(), (String) value.get("type"));
    428.                 }
    429.             }
    430.             esFieldTypes.put(key, fieldType);
    432.             return fieldType.get(fieldName);
    433.         }
    434.     }
    436.     private void putRelationDataFromRS(ESMapping mapping, SchemaItem schemaItem, ResultSet resultSet,
    437.                                        Map<String, Object> esFieldData) {
    438.         ESMapppingPlus.rebuildIndex(mapping, esFieldData);
    440.         // 添加父子文档关联信息
    441.         if (!mapping.getRelations().isEmpty()) {
    442.             mapping.getRelations().forEach((relationField, relationMapping) -> {
    443.                 Map<String, Object> relations = new HashMap<>();
    444.                 relations.put("name", relationMapping.getName());
    445.                 if (StringUtils.isNotEmpty(relationMapping.getParent())) {
    446.                     FieldItem parentFieldItem = schemaItem.getSelectFields().get(relationMapping.getParent());
    447.                     Object parentVal;
    448.                     try {
    449.                         parentVal = getValFromRS(mapping,
    450.                             resultSet,
    451.                             parentFieldItem.getFieldName(),
    452.                             parentFieldItem.getFieldName());
    453.                     } catch (SQLException e) {
    454.                         throw new RuntimeException(e);
    455.                     }
    456.                     if (parentVal != null) {
    457.                         relations.put("parent", parentVal.toString());
    458.                         esFieldData.put("$parent_routing", parentVal.toString());
    460.                     }
    461.                 }
    462.                 esFieldData.put(relationField, relations);
    463.             });
    464.         }
    465.     }
    467.     private void putRelationData(ESMapping mapping, SchemaItem schemaItem, Map<String, Object> dmlData,
    468.                                  Map<String, Object> esFieldData) {
    469.         // 添加父子文档关联信息
    470.         if (!mapping.getRelations().isEmpty()) {
    471.             mapping.getRelations().forEach((relationField, relationMapping) -> {
    472.                 Map<String, Object> relations = new HashMap<>();
    473.                 relations.put("name", relationMapping.getName());
    474.                 if (StringUtils.isNotEmpty(relationMapping.getParent())) {
    475.                     FieldItem parentFieldItem = schemaItem.getSelectFields().get(relationMapping.getParent());
    476.                     String columnName = parentFieldItem.getColumnItems().iterator().next().getColumnName();
    477.                     Object parentVal = getValFromData(mapping, dmlData, parentFieldItem.getFieldName(), columnName);
    478.                     if (parentVal != null) {
    479.                         relations.put("parent", parentVal.toString());
    480.                         esFieldData.put("$parent_routing", parentVal.toString());
    482.                     }
    483.                 }
    484.                 esFieldData.put(relationField, relations);
    485.             });
    486.         }
    487.     }
    488. }

    修改ESSyncService.java文件：
    同样添加：

    1. ESMapppingPlus.rebuildIndex(config, data);

    或者：

    1. ESMapppingPlus.rebuildIndex(config, esFieldData);

    完整修改如下：

    1. package com.alibaba.otter.canal.client.adapter.es.core.service;
    3. import java.util.ArrayList;
    4. import java.util.Collection;
    5. import java.util.LinkedHashMap;
    6. import java.util.List;
    7. import java.util.Map;
    9. import javax.sql.DataSource;
    11. import com.alibaba.otter.canal.client.adapter.es.core.support.ESMapppingPlus;
    12. import org.slf4j.Logger;
    13. import org.slf4j.LoggerFactory;
    15. import com.alibaba.fastjson.JSON;
    16. import com.alibaba.fastjson.serializer.SerializerFeature;
    17. import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock;
    18. import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
    19. import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
    20. import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem;
    21. import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem;
    22. import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem;
    23. import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.TableItem;
    24. import com.alibaba.otter.canal.client.adapter.es.core.config.SqlParser;
    25. import com.alibaba.otter.canal.client.adapter.es.core.support.ESSyncUtil;
    26. import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate;
    27. import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
    28. import com.alibaba.otter.canal.client.adapter.support.Dml;
    29. import com.alibaba.otter.canal.client.adapter.support.Util;
    31. /**
    32.  * ES 同步 Service
    33.  *
    34.  * @author rewerma 2018-11-01
    35.  * @version 1.0.0
    36.  */
    37. public class ESSyncService {
    39.     private static Logger logger = LoggerFactory.getLogger(ESSyncService.class);
    41.     private ESTemplate    esTemplate;
    43.     public ESSyncService(ESTemplate esTemplate){
    44.         this.esTemplate = esTemplate;
    45.     }
    47.     public void sync(Collection<ESSyncConfig> esSyncConfigs, Dml dml) {
    48.         long begin = System.currentTimeMillis();
    49.         if (esSyncConfigs != null) {
    50.             if (logger.isTraceEnabled()) {
    51.                 logger.trace("Destination: {}, database:{}, table:{}, type:{}, affected index count: {}",
    52.                     dml.getDestination(),
    53.                     dml.getDatabase(),
    54.                     dml.getTable(),
    55.                     dml.getType(),
    56.                     esSyncConfigs.size());
    57.             }
    59.             for (ESSyncConfig config : esSyncConfigs) {
    60.                 if (logger.isTraceEnabled()) {
    61.                     logger.trace("Prepared to sync index: {}, destination: {}",
    62.                         config.getEsMapping().get_index(),
    63.                         dml.getDestination());
    64.                 }
    65.                 this.sync(config, dml);
    66.                 if (logger.isTraceEnabled()) {
    67.                     logger.trace("Sync completed: {}, destination: {}",
    68.                         config.getEsMapping().get_index(),
    69.                         dml.getDestination());
    70.                 }
    71.             }
    72.             if (logger.isTraceEnabled()) {
    73.                 logger.trace("Sync elapsed time: {} ms, affected indexes count：{}, destination: {}",
    74.                     (System.currentTimeMillis() - begin),
    75.                     esSyncConfigs.size(),
    76.                     dml.getDestination());
    77.             }
    78.             if (logger.isDebugEnabled()) {
    79.                 StringBuilder configIndexes = new StringBuilder();
    80.                 esSyncConfigs
    81.                     .forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
    82.                 logger.debug("DML: {} \nAffected indexes: {}",
    83.                     JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue),
    84.                     configIndexes.toString());
    85.             }
    86.         }
    87.     }
    89.     public void sync(ESSyncConfig config, Dml dml) {
    90.         try {
    91.             // 如果是按时间戳定时更新则返回
    92.             if (config.getEsMapping().isSyncByTimestamp()) {
    93.                 return;
    94.             }
    96.             long begin = System.currentTimeMillis();
    98.             String type = dml.getType();
    99.             if (type != null && type.equalsIgnoreCase("INSERT")) {
    100.                 insert(config, dml);
    101.             } else if (type != null && type.equalsIgnoreCase("UPDATE")) {
    102.                 update(config, dml);
    103.             } else if (type != null && type.equalsIgnoreCase("DELETE")) {
    104.                 delete(config, dml);
    105.             } else {
    106.                 return;
    107.             }
    109.             if (logger.isTraceEnabled()) {
    110.                 logger.trace("Sync elapsed time: {} ms,destination: {}, es index: {}",
    111.                     (System.currentTimeMillis() - begin),
    112.                     dml.getDestination(),
    113.                     config.getEsMapping().get_index());
    114.             }
    115.         } catch (Throwable e) {
    116.             logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().get_index(), dml);
    117.             throw new RuntimeException(e);
    118.         }
    119.     }
    121.     /**
    122.      * 插入操作dml
    123.      *
    124.      * @param config es配置
    125.      * @param dml dml数据
    126.      */
    127.     private void insert(ESSyncConfig config, Dml dml) {
    128.         List<Map<String, Object>> dataList = dml.getData();
    129.         if (dataList == null || dataList.isEmpty()) {
    130.             return;
    131.         }
    132.         SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
    133.         for (Map<String, Object> data : dataList) {
    134.             if (data == null || data.isEmpty()) {
    135.                 continue;
    136.             }
    137.             ESMapppingPlus.rebuildIndex(config, data);
    139.             if (schemaItem.getAliasTableItems().size() == 1 && schemaItem.isAllFieldsSimple()) {
    140.                 // ------单表 & 所有字段都为简单字段------
    141.                 singleTableSimpleFiledInsert(config, dml, data);
    142.             } else {
    143.                 // ------是主表 查询sql来插入------
    144.                 if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
    145.                     mainTableInsert(config, dml, data);
    146.                 }
    148.                 // 从表的操作
    149.                 for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
    150.                     if (tableItem.isMain()) {
    151.                         continue;
    152.                     }
    153.                     if (!tableItem.getTableName().equals(dml.getTable())) {
    154.                         continue;
    155.                     }
    156.                     // 关联条件出现在主表查询条件是否为简单字段
    157.                     boolean allFieldsSimple = true;
    158.                     for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    159.                         if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
    160.                             allFieldsSimple = false;
    161.                             break;
    162.                         }
    163.                     }
    164.                     // 所有查询字段均为简单字段
    165.                     if (allFieldsSimple) {
    166.                         // 不是子查询
    167.                         if (!tableItem.isSubQuery()) {
    168.                             // ------关联表简单字段插入------
    169.                             Map<String, Object> esFieldData = new LinkedHashMap<>();
    170.                             for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    171.                                 Object value = esTemplate.getValFromData(config.getEsMapping(),
    172.                                     data,
    173.                                     fieldItem.getFieldName(),
    174.                                     fieldItem.getColumn().getColumnName());
    175.                                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
    176.                             }
    178.                             joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
    179.                         } else {
    180.                             // ------关联子表简单字段插入------
    181.                             subTableSimpleFieldOperation(config, dml, data, null, tableItem);
    182.                         }
    183.                     } else {
    184.                         // ------关联子表复杂字段插入 执行全sql更新es------
    185.                         wholeSqlOperation(config, dml, data, null, tableItem);
    186.                     }
    187.                 }
    188.             }
    189.         }
    190.     }
    192.     /**
    193.      * 更新操作dml
    194.      *
    195.      * @param config es配置
    196.      * @param dml dml数据
    197.      */
    198.     private void update(ESSyncConfig config, Dml dml) {
    199.         List<Map<String, Object>> dataList = dml.getData();
    200.         List<Map<String, Object>> oldList = dml.getOld();
    201.         if (dataList == null || dataList.isEmpty() || oldList == null || oldList.isEmpty()) {
    202.             return;
    203.         }
    204.         SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
    205.         int i = 0;
    206.         for (Map<String, Object> data : dataList) {
    207.             Map<String, Object> old = oldList.get(i);
    208.             if (data == null || data.isEmpty() || old == null || old.isEmpty()) {
    209.                 continue;
    210.             }
    211.             ESMapppingPlus.rebuildIndex(config, data);
    213.             if (schemaItem.getAliasTableItems().size() == 1 && schemaItem.isAllFieldsSimple()) {
    214.                 // ------单表 & 所有字段都为简单字段------
    215.                 singleTableSimpleFiledUpdate(config, dml, data, old);
    216.             } else {
    217.                 // ------主表 查询sql来更新------
    218.                 if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
    219.                     ESMapping mapping = config.getEsMapping();
    220.                     String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
    221.                     FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName);
    223.                     boolean idFieldSimple = true;
    224.                     if (idFieldItem.isMethod() || idFieldItem.isBinaryOp()) {
    225.                         idFieldSimple = false;
    226.                     }
    228.                     boolean allUpdateFieldSimple = true;
    229.                     out: for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
    230.                         for (ColumnItem columnItem : fieldItem.getColumnItems()) {
    231.                             if (old.containsKey(columnItem.getColumnName())) {
    232.                                 if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
    233.                                     allUpdateFieldSimple = false;
    234.                                     break out;
    235.                                 }
    236.                             }
    237.                         }
    238.                     }
    240.                     // 不支持主键更新!!
    242.                     // 判断是否有外键更新
    243.                     boolean fkChanged = false;
    244.                     for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
    245.                         if (tableItem.isMain()) {
    246.                             continue;
    247.                         }
    248.                         boolean changed = false;
    249.                         for (List<FieldItem> fieldItems : tableItem.getRelationTableFields().values()) {
    250.                             for (FieldItem fieldItem : fieldItems) {
    251.                                 if (old.containsKey(fieldItem.getColumn().getColumnName())) {
    252.                                     fkChanged = true;
    253.                                     changed = true;
    254.                                     break;
    255.                                 }
    256.                             }
    257.                         }
    258.                         // 如果外键有修改,则更新所对应该表的所有查询条件数据
    259.                         if (changed) {
    260.                             for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    261.                                 fieldItem.getColumnItems()
    262.                                     .forEach(columnItem -> old.put(columnItem.getColumnName(), null));
    263.                             }
    264.                         }
    265.                     }
    267.                     // 判断主键和所更新的字段是否全为简单字段
    268.                     if (idFieldSimple && allUpdateFieldSimple && !fkChanged) {
    269.                         singleTableSimpleFiledUpdate(config, dml, data, old);
    270.                     } else {
    271.                         mainTableUpdate(config, dml, data, old);
    272.                     }
    273.                 }
    275.                 // 从表的操作
    276.                 for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
    277.                     if (tableItem.isMain()) {
    278.                         continue;
    279.                     }
    280.                     if (!tableItem.getTableName().equals(dml.getTable())) {
    281.                         continue;
    282.                     }
    284.                     // 关联条件出现在主表查询条件是否为简单字段
    285.                     boolean allFieldsSimple = true;
    286.                     for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    287.                         if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
    288.                             allFieldsSimple = false;
    289.                             break;
    290.                         }
    291.                     }
    293.                     // 所有查询字段均为简单字段
    294.                     if (allFieldsSimple) {
    295.                         // 不是子查询
    296.                         if (!tableItem.isSubQuery()) {
    297.                             // ------关联表简单字段更新------
    298.                             Map<String, Object> esFieldData = new LinkedHashMap<>();
    299.                             for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    300.                                 if (old.containsKey(fieldItem.getColumn().getColumnName())) {
    301.                                     Object value = esTemplate.getValFromData(config.getEsMapping(),
    302.                                         data,
    303.                                         fieldItem.getFieldName(),
    304.                                         fieldItem.getColumn().getColumnName());
    305.                                     esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
    306.                                 }
    307.                             }
    308.                             joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
    309.                         } else {
    310.                             // ------关联子表简单字段更新------
    311.                             subTableSimpleFieldOperation(config, dml, data, old, tableItem);
    312.                         }
    313.                     } else {
    314.                         // ------关联子表复杂字段更新 执行全sql更新es------
    315.                         wholeSqlOperation(config, dml, data, old, tableItem);
    316.                     }
    317.                 }
    318.             }
    320.             i++;
    321.         }
    322.     }
    324.     /**
    325.      * 删除操作dml
    326.      *
    327.      * @param config es配置
    328.      * @param dml dml数据
    329.      */
    330.     private void delete(ESSyncConfig config, Dml dml) {
    331.         List<Map<String, Object>> dataList = dml.getData();
    332.         if (dataList == null || dataList.isEmpty()) {
    333.             return;
    334.         }
    335.         SchemaItem schemaItem = config.getEsMapping().getSchemaItem();
    337.         for (Map<String, Object> data : dataList) {
    338.             if (data == null || data.isEmpty()) {
    339.                 continue;
    340.             }
    341.             ESMapppingPlus.rebuildIndex(config, data);
    343.             ESMapping mapping = config.getEsMapping();
    344.             // ------是主表------
    345.             if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
    346.                 if (mapping.get_id() != null) {
    347.                     FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping);
    348.                     // 主键为简单字段
    349.                     if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) {
    350.                         Object idVal = esTemplate.getValFromData(mapping,
    351.                             data,
    352.                             idFieldItem.getFieldName(),
    353.                             idFieldItem.getColumn().getColumnName());
    355.                         if (logger.isTraceEnabled()) {
    356.                             logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, id: {}",
    357.                                 config.getDestination(),
    358.                                 dml.getTable(),
    359.                                 mapping.get_index(),
    360.                                 idVal);
    361.                         }
    362.                         esTemplate.delete(mapping, idVal, null);
    363.                     } else {
    364.                         // ------主键带函数, 查询sql获取主键删除------
    365.                         // FIXME 删除时反查sql为空记录, 无法获获取 id field 值
    366.                         mainTableDelete(config, dml, data);
    367.                     }
    368.                 } else {
    369.                     FieldItem pkFieldItem = schemaItem.getIdFieldItem(mapping);
    370.                     if (!pkFieldItem.isMethod() && !pkFieldItem.isBinaryOp()) {
    371.                         Map<String, Object> esFieldData = new LinkedHashMap<>();
    372.                         Object pkVal = esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
    374.                         if (logger.isTraceEnabled()) {
    375.                             logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, pk: {}",
    376.                                 config.getDestination(),
    377.                                 dml.getTable(),
    378.                                 mapping.get_index(),
    379.                                 pkVal);
    380.                         }
    381.                         esFieldData.remove(pkFieldItem.getFieldName());
    382.                         esFieldData.keySet().forEach(key -> esFieldData.put(key, null));
    383.                         esTemplate.delete(mapping, pkVal, esFieldData);
    384.                     } else {
    385.                         // ------主键带函数, 查询sql获取主键删除------
    386.                         mainTableDelete(config, dml, data);
    387.                     }
    388.                 }
    390.             }
    392.             // 从表的操作
    393.             for (TableItem tableItem : schemaItem.getAliasTableItems().values()) {
    394.                 if (tableItem.isMain()) {
    395.                     continue;
    396.                 }
    397.                 if (!tableItem.getTableName().equals(dml.getTable())) {
    398.                     continue;
    399.                 }
    401.                 // 关联条件出现在主表查询条件是否为简单字段
    402.                 boolean allFieldsSimple = true;
    403.                 for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    404.                     if (fieldItem.isMethod() || fieldItem.isBinaryOp()) {
    405.                         allFieldsSimple = false;
    406.                         break;
    407.                     }
    408.                 }
    410.                 // 所有查询字段均为简单字段
    411.                 if (allFieldsSimple) {
    412.                     // 不是子查询
    413.                     if (!tableItem.isSubQuery()) {
    414.                         // ------关联表简单字段更新为null------
    415.                         Map<String, Object> esFieldData = new LinkedHashMap<>();
    416.                         for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    417.                             esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), null);
    418.                         }
    419.                         joinTableSimpleFieldOperation(config, dml, data, tableItem, esFieldData);
    420.                     } else {
    421.                         // ------关联子表简单字段更新------
    422.                         subTableSimpleFieldOperation(config, dml, data, null, tableItem);
    423.                     }
    424.                 } else {
    425.                     // ------关联子表复杂字段更新 执行全sql更新es------
    426.                     wholeSqlOperation(config, dml, data, null, tableItem);
    427.                 }
    428.             }
    429.         }
    430.     }
    432.     /**
    433.      * 单表简单字段insert
    434.      *
    435.      * @param config es配置
    436.      * @param dml dml信息
    437.      * @param data 单行dml数据
    438.      */
    439.     private void singleTableSimpleFiledInsert(ESSyncConfig config, Dml dml, Map<String, Object> data) {
    440.         ESMapping mapping = config.getEsMapping();
    441.         Map<String, Object> esFieldData = new LinkedHashMap<>();
    442.         Object idVal = esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
    444.         if (logger.isTraceEnabled()) {
    445.             logger.trace("Single table insert to es index, destination:{}, table: {}, index: {}, id: {}",
    446.                 config.getDestination(),
    447.                 dml.getTable(),
    448.                 mapping.get_index(),
    449.                 idVal);
    450.         }
    451.         esTemplate.insert(mapping, idVal, esFieldData);
    452.     }
    454.     /**
    455.      * 主表(单表)复杂字段insert
    456.      *
    457.      * @param config es配置
    458.      * @param dml dml信息
    459.      * @param data 单行dml数据
    460.      */
    461.     private void mainTableInsert(ESSyncConfig config, Dml dml, Map<String, Object> data) {
    462.         ESMapping mapping = config.getEsMapping();
    463.         String sql = mapping.getSql();
    464.         String condition = ESSyncUtil.pkConditionSql(mapping, data);
    465.         sql = ESSyncUtil.appendCondition(sql, condition);
    466.         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
    467.         if (logger.isTraceEnabled()) {
    468.             logger.trace("Main table insert to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
    469.                 config.getDestination(),
    470.                 dml.getTable(),
    471.                 mapping.get_index(),
    472.                 sql.replace("\n", " "));
    473.         }
    474.         Util.sqlRS(ds, sql, rs -> {
    475.             try {
    476.                 while (rs.next()) {
    477.                     Map<String, Object> esFieldData = new LinkedHashMap<>();
    478.                     Object idVal = esTemplate.getESDataFromRS(mapping, rs, esFieldData);
    480.                     if (logger.isTraceEnabled()) {
    481.                         logger.trace(
    482.                             "Main table insert to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
    483.                             config.getDestination(),
    484.                             dml.getTable(),
    485.                             mapping.get_index(),
    486.                             idVal);
    487.                     }
    488.                     esTemplate.insert(mapping, idVal, esFieldData);
    489.                 }
    490.             } catch (Exception e) {
    491.                 throw new RuntimeException(e);
    492.             }
    493.             return 0;
    494.         });
    495.     }
    497.     private void mainTableDelete(ESSyncConfig config, Dml dml, Map<String, Object> data) {
    498.         ESMapping mapping = config.getEsMapping();
    499.         String sql = mapping.getSql();
    500.         String condition = ESSyncUtil.pkConditionSql(mapping, data);
    501.         sql = ESSyncUtil.appendCondition(sql, condition);
    502.         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
    503.         if (logger.isTraceEnabled()) {
    504.             logger.trace("Main table delete es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
    505.                 config.getDestination(),
    506.                 dml.getTable(),
    507.                 mapping.get_index(),
    508.                 sql.replace("\n", " "));
    509.         }
    510.         Util.sqlRS(ds, sql, rs -> {
    511.             try {
    512.                 Map<String, Object> esFieldData = null;
    513.                 if (mapping.getPk() != null) {
    514.                     esFieldData = new LinkedHashMap<>();
    515.                     esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
    516.                     esFieldData.remove(mapping.getPk());
    517.                     for (String key : esFieldData.keySet()) {
    518.                         esFieldData.put(Util.cleanColumn(key), null);
    519.                     }
    520.                 }
    521.                 while (rs.next()) {
    522.                     Object idVal = esTemplate.getIdValFromRS(mapping, rs);
    524.                     if (logger.isTraceEnabled()) {
    525.                         logger.trace(
    526.                             "Main table delete to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
    527.                             config.getDestination(),
    528.                             dml.getTable(),
    529.                             mapping.get_index(),
    530.                             idVal);
    531.                     }
    532.                     esTemplate.delete(mapping, idVal, esFieldData);
    533.                 }
    534.             } catch (Exception e) {
    535.                 throw new RuntimeException(e);
    536.             }
    537.             return 0;
    538.         });
    539.     }
    541.     /**
    542.      * 关联表主表简单字段operation
    543.      *
    544.      * @param config es配置
    545.      * @param dml dml信息
    546.      * @param data 单行dml数据
    547.      * @param tableItem 当前表配置
    548.      */
    549.     private void joinTableSimpleFieldOperation(ESSyncConfig config, Dml dml, Map<String, Object> data,
    550.                                                TableItem tableItem, Map<String, Object> esFieldData) {
    551.         ESMapping mapping = config.getEsMapping();
    553.         Map<String, Object> paramsTmp = new LinkedHashMap<>();
    554.         for (Map.Entry<FieldItem, List<FieldItem>> entry : tableItem.getRelationTableFields().entrySet()) {
    555.             for (FieldItem fieldItem : entry.getValue()) {
    556.                 if (fieldItem.getColumnItems().size() == 1) {
    557.                     Object value = esTemplate.getValFromData(mapping,
    558.                         data,
    559.                         fieldItem.getFieldName(),
    560.                         entry.getKey().getColumn().getColumnName());
    562.                     String fieldName = fieldItem.getFieldName();
    563.                     // 判断是否是主键
    564.                     if (fieldName.equals(mapping.get_id())) {
    565.                         fieldName = "_id";
    566.                     }
    567.                     paramsTmp.put(fieldName, value);
    568.                 }
    569.             }
    570.         }
    572.         if (logger.isDebugEnabled()) {
    573.             logger.trace("Join table update es index by foreign key, destination:{}, table: {}, index: {}",
    574.                 config.getDestination(),
    575.                 dml.getTable(),
    576.                 mapping.get_index());
    577.         }
    578.         esTemplate.updateByQuery(config, paramsTmp, esFieldData);
    579.     }
    581.     /**
    582.      * 关联子查询, 主表简单字段operation
    583.      *
    584.      * @param config es配置
    585.      * @param dml dml信息
    586.      * @param data 单行dml数据
    587.      * @param old 单行old数据
    588.      * @param tableItem 当前表配置
    589.      */
    590.     private void subTableSimpleFieldOperation(ESSyncConfig config, Dml dml, Map<String, Object> data,
    591.                                               Map<String, Object> old, TableItem tableItem) {
    592.         ESMapping mapping = config.getEsMapping();
    594.         MySqlSelectQueryBlock queryBlock = SqlParser.parseSQLSelectQueryBlock(tableItem.getSubQuerySql());
    595.         StringBuilder sql = new StringBuilder();
    596.         sql.append("SELECT ")
    597.             .append(SqlParser.parse4SQLSelectItem(queryBlock))
    598.             .append(" FROM ")
    599.             .append(SqlParser.parse4FromTableSource(queryBlock));
    601.         String whereSql = SqlParser.parse4WhereItem(queryBlock);
    602.         if (whereSql != null) {
    603.             sql.append(" WHERE ").append(whereSql);
    604.         } else {
    605.             sql.append(" WHERE 1=1 ");
    606.         }
    608.         List<Object> values = new ArrayList<>();
    610.         for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) {
    611.             String columnName = fkFieldItem.getColumn().getColumnName();
    612.             Object value = esTemplate.getValFromData(mapping, data, fkFieldItem.getFieldName(), columnName);
    613.             sql.append(" AND ").append(columnName).append("=? ");
    614.             values.add(value);
    615.         }
    617.         String groupSql = SqlParser.parse4GroupBy(queryBlock);
    618.         if (groupSql != null) {
    619.             sql.append(groupSql);
    620.         }
    622.         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
    623.         if (logger.isTraceEnabled()) {
    624.             logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
    625.                 config.getDestination(),
    626.                 dml.getTable(),
    627.                 mapping.get_index(),
    628.                 sql.toString().replace("\n", " "));
    629.         }
    630.         Util.sqlRS(ds, sql.toString(), values, rs -> {
    631.             try {
    632.                 while (rs.next()) {
    633.                     Map<String, Object> esFieldData = new LinkedHashMap<>();
    635.                     for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    636.                         if (old != null) {
    637.                             out: for (FieldItem fieldItem1 : tableItem.getSubQueryFields()) {
    638.                                 for (ColumnItem columnItem0 : fieldItem.getColumnItems()) {
    639.                                     if (fieldItem1.getFieldName().equals(columnItem0.getColumnName()))
    640.                                         for (ColumnItem columnItem : fieldItem1.getColumnItems()) {
    641.                                             if (old.containsKey(columnItem.getColumnName())) {
    642.                                                 Object val = esTemplate.getValFromRS(mapping,
    643.                                                     rs,
    644.                                                     fieldItem.getFieldName(),
    645.                                                     fieldItem.getColumn().getColumnName());
    646.                                                 esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
    647.                                                 break out;
    648.                                             }
    649.                                         }
    650.                                 }
    651.                             }
    652.                         } else {
    653.                             Object val = esTemplate.getValFromRS(mapping,
    654.                                 rs,
    655.                                 fieldItem.getFieldName(),
    656.                                 fieldItem.getColumn().getColumnName());
    657.                             esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
    658.                         }
    659.                     }
    661.                     Map<String, Object> paramsTmp = new LinkedHashMap<>();
    662.                     for (Map.Entry<FieldItem, List<FieldItem>> entry : tableItem.getRelationTableFields().entrySet()) {
    663.                         for (FieldItem fieldItem : entry.getValue()) {
    664.                             if (fieldItem.getColumnItems().size() == 1) {
    665.                                 Object value = esTemplate.getValFromRS(mapping,
    666.                                     rs,
    667.                                     fieldItem.getFieldName(),
    668.                                     entry.getKey().getColumn().getColumnName());
    669.                                 String fieldName = fieldItem.getFieldName();
    670.                                 // 判断是否是主键
    671.                                 if (fieldName.equals(mapping.get_id())) {
    672.                                     fieldName = "_id";
    673.                                 }
    674.                                 paramsTmp.put(fieldName, value);
    675.                             }
    676.                         }
    677.                     }
    679.                     if (logger.isDebugEnabled()) {
    680.                         logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}",
    681.                             config.getDestination(),
    682.                             dml.getTable(),
    683.                             mapping.get_index());
    684.                     }
    685.                     esTemplate.updateByQuery(config, paramsTmp, esFieldData);
    686.                 }
    687.             } catch (Exception e) {
    688.                 throw new RuntimeException(e);
    689.             }
    690.             return 0;
    691.         });
    692.     }
    694.     /**
    695.      * 关联(子查询), 主表复杂字段operation, 全sql执行
    696.      *
    697.      * @param config es配置
    698.      * @param dml dml信息
    699.      * @param data 单行dml数据
    700.      * @param tableItem 当前表配置
    701.      */
    702.     private void wholeSqlOperation(ESSyncConfig config, Dml dml, Map<String, Object> data, Map<String, Object> old,
    703.                                    TableItem tableItem) {
    704.         ESMapping mapping = config.getEsMapping();
    705.         // 防止最后出现groupby 导致sql解析异常
    706.         String[] sqlSplit = mapping.getSql().split("GROUP\\ BY(?!(.*)ON)");
    707.         String sqlNoWhere = sqlSplit[0];
    709.         String sqlGroupBy = "";
    711.         if (sqlSplit.length > 1) {
    712.             sqlGroupBy = "GROUP BY " + sqlSplit[1];
    713.         }
    715.         StringBuilder sql = new StringBuilder(sqlNoWhere + " WHERE ");
    717.         for (FieldItem fkFieldItem : tableItem.getRelationTableFields().keySet()) {
    718.             String columnName = fkFieldItem.getColumn().getColumnName();
    719.             Object value = esTemplate.getValFromData(mapping, data, fkFieldItem.getFieldName(), columnName);
    720.             ESSyncUtil.appendCondition(sql, value, tableItem.getAlias(), columnName);
    721.         }
    722.         int len = sql.length();
    723.         sql.delete(len - 5, len);
    724.         sql.append(sqlGroupBy);
    726.         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
    727.         if (logger.isTraceEnabled()) {
    728.             logger.trace("Join table update es index by query whole sql, destination:{}, table: {}, index: {}, sql: {}",
    729.                 config.getDestination(),
    730.                 dml.getTable(),
    731.                 mapping.get_index(),
    732.                 sql.toString().replace("\n", " "));
    733.         }
    734.         Util.sqlRS(ds, sql.toString(), rs -> {
    735.             try {
    736.                 while (rs.next()) {
    737.                     Map<String, Object> esFieldData = new LinkedHashMap<>();
    738.                     for (FieldItem fieldItem : tableItem.getRelationSelectFieldItems()) {
    739.                         if (old != null) {
    740.                             // 从表子查询
    741.                             out: for (FieldItem fieldItem1 : tableItem.getSubQueryFields()) {
    742.                                 for (ColumnItem columnItem0 : fieldItem.getColumnItems()) {
    743.                                     if (fieldItem1.getFieldName().equals(columnItem0.getColumnName()))
    744.                                         for (ColumnItem columnItem : fieldItem1.getColumnItems()) {
    745.                                             if (old.containsKey(columnItem.getColumnName())) {
    746.                                                 Object val = esTemplate.getValFromRS(mapping,
    747.                                                     rs,
    748.                                                     fieldItem.getFieldName(),
    749.                                                     fieldItem.getFieldName());
    750.                                                 esFieldData.put(fieldItem.getFieldName(), val);
    751.                                                 break out;
    752.                                             }
    753.                                         }
    754.                                 }
    755.                             }
    756.                             // 从表非子查询
    757.                             for (FieldItem fieldItem1 : tableItem.getRelationSelectFieldItems()) {
    758.                                 if (fieldItem1.equals(fieldItem)) {
    759.                                     for (ColumnItem columnItem : fieldItem1.getColumnItems()) {
    760.                                         if (old.containsKey(columnItem.getColumnName())) {
    761.                                             Object val = esTemplate.getValFromRS(mapping,
    762.                                                 rs,
    763.                                                 fieldItem.getFieldName(),
    764.                                                 fieldItem.getFieldName());
    765.                                             esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
    766.                                             break;
    767.                                         }
    768.                                     }
    769.                                 }
    770.                             }
    771.                         } else {
    772.                             Object val = esTemplate
    773.                                 .getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
    774.                             esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), val);
    775.                         }
    776.                     }
    778.                     Map<String, Object> paramsTmp = new LinkedHashMap<>();
    779.                     for (Map.Entry<FieldItem, List<FieldItem>> entry : tableItem.getRelationTableFields().entrySet()) {
    780.                         for (FieldItem fieldItem : entry.getValue()) {
    781.                             Object value = esTemplate
    782.                                 .getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
    783.                             String fieldName = fieldItem.getFieldName();
    784.                             // 判断是否是主键
    785.                             if (fieldName.equals(mapping.get_id())) {
    786.                                 fieldName = "_id";
    787.                             }
    788.                             paramsTmp.put(fieldName, value);
    789.                         }
    790.                     }
    792.                     if (logger.isDebugEnabled()) {
    793.                         logger.trace(
    794.                             "Join table update es index by query whole sql, destination:{}, table: {}, index: {}",
    795.                             config.getDestination(),
    796.                             dml.getTable(),
    797.                             mapping.get_index());
    798.                     }
    799.                     esTemplate.updateByQuery(config, paramsTmp, esFieldData);
    800.                 }
    801.             } catch (Exception e) {
    802.                 throw new RuntimeException(e);
    803.             }
    804.             return 0;
    805.         });
    806.     }
    808.     /**
    809.      * 单表简单字段update
    810.      *
    811.      * @param config es配置
    812.      * @param dml dml信息
    813.      * @param data 单行data数据
    814.      * @param old 单行old数据
    815.      */
    816.     private void singleTableSimpleFiledUpdate(ESSyncConfig config, Dml dml, Map<String, Object> data,
    817.                                               Map<String, Object> old) {
    818.         ESMapping mapping = config.getEsMapping();
    819.         Map<String, Object> esFieldData = new LinkedHashMap<>();
    821.         Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData);
    823.         if (logger.isTraceEnabled()) {
    824.             logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}",
    825.                 config.getDestination(),
    826.                 dml.getTable(),
    827.                 mapping.get_index(),
    828.                 idVal);
    829.         }
    830.         esTemplate.update(mapping, idVal, esFieldData);
    831.     }
    833.     /**
    834.      * 主表(单表)复杂字段update
    835.      *
    836.      * @param config es配置
    837.      * @param dml dml信息
    838.      * @param data 单行dml数据
    839.      */
    840.     private void mainTableUpdate(ESSyncConfig config, Dml dml, Map<String, Object> data, Map<String, Object> old) {
    841.         ESMapping mapping = config.getEsMapping();
    842.         String sql = mapping.getSql();
    843.         String condition = ESSyncUtil.pkConditionSql(mapping, data);
    844.         sql = ESSyncUtil.appendCondition(sql, condition);
    845.         DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
    846.         if (logger.isTraceEnabled()) {
    847.             logger.trace("Main table update to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
    848.                 config.getDestination(),
    849.                 dml.getTable(),
    850.                 mapping.get_index(),
    851.                 sql.replace("\n", " "));
    852.         }
    853.         Util.sqlRS(ds, sql, rs -> {
    854.             try {
    855.                 while (rs.next()) {
    856.                     Map<String, Object> esFieldData = new LinkedHashMap<>();
    857.                     Object idVal = esTemplate.getESDataFromRS(mapping, rs, old, esFieldData);
    859.                     if (logger.isTraceEnabled()) {
    860.                         logger.trace(
    861.                             "Main table update to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
    862.                             config.getDestination(),
    863.                             dml.getTable(),
    864.                             mapping.get_index(),
    865.                             idVal);
    866.                     }
    867.                     esTemplate.update(mapping, idVal, esFieldData);
    868.                 }
    869.             } catch (Exception e) {
    870.                 throw new RuntimeException(e);
    871.             }
    872.             return 0;
    873.         });
    874.     }
    876.     /**
    877.      * 提交批次
    878.      */
    879.     public void commit() {
    880.         esTemplate.commit();
    881.     }
    882. }