From 96b9484f701f7a54534b6b10abb9136e3c61c5ab Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 1 Nov 2018 16:37:47 +0800 Subject: [PATCH] modify --- .../adapter/es/service/ESSyncService.java | 143 +++++++++++++++++- .../client/adapter/es/support/ESTemplate.java | 53 ++++++- .../es/test/sync/UserSyncJoinOneTest.java | 26 ++++ .../es/test/sync/UserSyncSingleTest.java | 29 ++++ .../loader/CanalAdapterKafkaWorker.java | 6 +- .../src/main/resources/application.yml | 10 +- .../adapter/logger/LoggerAdapterExample.java | 2 +- 7 files changed, 252 insertions(+), 17 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java index f34e38e1..608fde0e 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java @@ -11,13 +11,13 @@ import org.elasticsearch.index.query.QueryBuilders; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.fastjson.JSON; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.TableItem; +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem; import com.alibaba.otter.canal.client.adapter.es.support.ESSyncUtil; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; @@ -35,7 +35,7 @@ public class ESSyncService { public void sync(Dml dml) { if (logger.isDebugEnabled()) { - logger.debug("DML: {}", JSON.toJSONString(dml)); + logger.debug("DML: {}", dml.toString()); } long begin = System.currentTimeMillis(); String database = dml.getDatabase(); @@ -103,7 +103,55 @@ public class ESSyncService { } private void update(ESSyncConfig config, Dml dml) { + List> dataList = dml.getData(); + List> oldList = dml.getOld(); + if (dataList == null || dataList.isEmpty() || oldList == null || oldList.isEmpty()) { + return; + } + SchemaItem schemaItem = config.getEsMapping().getSchemaItem(); + int i = 0; + for (Map data : dataList) { + Map old = oldList.get(i); + if (data == null || data.isEmpty() || old == null || old.isEmpty()) { + continue; + } + if (schemaItem.getAliasTableItems().size() == 1 && schemaItem.isAllFieldsSimple()) { + // ------单表 & 所有字段都为简单字段------ + singleTableSimpleFiledUpdate(config, dml, data, old); + } else { + // ------是主表 查询sql来更新------ + if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { + ESMapping mapping = config.getEsMapping(); + String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); + FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName); + boolean idFieldSimple = true; + if (idFieldItem.isMethod() || idFieldItem.isBinaryOp()) { + idFieldSimple = false; + } + + boolean allUpdateFieldSimple = true; + out: for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { + for (ColumnItem columnItem : fieldItem.getColumnItems()) { + if (old.containsKey(columnItem.getColumnName())) { + if (fieldItem.isMethod() || fieldItem.isBinaryOp()) { + allUpdateFieldSimple = false; + break out; + } + } + } + } + // 判断主键和所更新的字段是否全为简单字段 + if (idFieldSimple && allUpdateFieldSimple) { + singleTableSimpleFiledUpdate(config, dml, data, old); + } else { + mainTableUpdate(config, dml, data, old); + } + } + } + + i++; + } } /** @@ -123,8 +171,8 @@ public class ESSyncService { continue; } - // ------是否单表 & 所有字段都为简单字段------ if (schemaItem.getAliasTableItems().size() == 1 && schemaItem.isAllFieldsSimple()) { + // ------单表 & 所有字段都为简单字段------ singleTableSimpleFiledInsert(config, dml, data); } else { // ------是主表 查询sql来插入------ @@ -210,7 +258,7 @@ public class ESSyncService { sql = ESSyncUtil.appendCondition(sql, condition); DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); if (logger.isTraceEnabled()) { - logger.trace("Single table insert ot es index by query sql, destination:{}, table: {}, index: {}, sql: {}", + logger.trace("Main table insert ot es index by query sql, destination:{}, table: {}, index: {}, sql: {}", config.getDestination(), dml.getTable(), mapping.get_index(), @@ -224,7 +272,7 @@ public class ESSyncService { if (logger.isTraceEnabled()) { logger.trace( - "Single table insert ot es index by query sql, destination:{}, table: {}, index: {}, id: {}", + "Main table insert ot es index by query sql, destination:{}, table: {}, index: {}, id: {}", config.getDestination(), dml.getTable(), mapping.get_index(), @@ -233,7 +281,7 @@ public class ESSyncService { boolean result = esTemplate.insert(config, idVal, esFieldData); if (!result) { logger.error( - "Single table insert to es index by query sql error, destination:{}, table: {}, index: {}, id: {}", + "Main table insert to es index by query sql error, destination:{}, table: {}, index: {}, id: {}", config.getDestination(), dml.getTable(), mapping.get_index(), @@ -486,4 +534,87 @@ public class ESSyncService { return 0; }); } + + /** + * 单表简单字段update + * + * @param config es配置 + * @param dml dml信息 + * @param data 单行data数据 + * @param old 单行old数据 + */ + private void singleTableSimpleFiledUpdate(ESSyncConfig config, Dml dml, Map data, + Map old) { + ESMapping mapping = config.getEsMapping(); + Map esFieldData = new LinkedHashMap<>(); + + Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData); + + if (logger.isTraceEnabled()) { + logger.trace("Single table update ot es index, destination:{}, table: {}, index: {}, id: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index(), + idVal); + } + boolean result = esTemplate.update(config, idVal, esFieldData); + if (!result) { + logger.error("Single table update to es index error, destination:{}, table: {}, index: {}, id: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index(), + idVal); + } + } + + /** + * 主表(单表)复杂字段update + * + * @param config es配置 + * @param dml dml信息 + * @param data 单行dml数据 + */ + private void mainTableUpdate(ESSyncConfig config, Dml dml, Map data, Map old) { + ESMapping mapping = config.getEsMapping(); + String sql = mapping.getSql(); + String condition = ESSyncUtil.pkConditionSql(mapping, data); + sql = ESSyncUtil.appendCondition(sql, condition); + DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); + if (logger.isTraceEnabled()) { + logger.trace("Main table update ot es index by query sql, destination:{}, table: {}, index: {}, sql: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index(), + sql.replace("\n", " ")); + } + ESSyncUtil.sqlRS(ds, sql, rs -> { + try { + while (rs.next()) { + Map esFieldData = new LinkedHashMap<>(); + Object idVal = esTemplate.getESDataFromRS(mapping, rs, old, esFieldData); + + if (logger.isTraceEnabled()) { + logger.trace( + "Main table update ot es index by query sql, destination:{}, table: {}, index: {}, id: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index(), + idVal); + } + boolean result = esTemplate.update(config, idVal, esFieldData); + if (!result) { + logger.error( + "Main table update to es index by query sql error, destination:{}, table: {}, index: {}, id: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index(), + idVal); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } + return 0; + }); + } } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java index e81935ab..c075eb02 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java @@ -34,6 +34,7 @@ import org.springframework.util.CollectionUtils; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; +import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; public class ESTemplate { @@ -143,7 +144,7 @@ public class ESTemplate { } private boolean updateByQuery(ESMapping mapping, QueryBuilder queryBuilder, Map esFieldData, - int counter) { + int counter) { if (CollectionUtils.isEmpty(esFieldData)) { return true; } @@ -307,6 +308,28 @@ public class ESTemplate { return resultIdVal; } + public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map dmlOld, + Map esFieldData) throws SQLException { + SchemaItem schemaItem = mapping.getSchemaItem(); + String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); + Object resultIdVal = null; + for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { + if (fieldItem.getFieldName().equals(idFieldName)) { + resultIdVal = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); + } + + for (ColumnItem columnItem : fieldItem.getColumnItems()) { + if (dmlOld.get(columnItem.getColumnName()) != null + && !mapping.getSkips().contains(fieldItem.getFieldName())) { + esFieldData.put(fieldItem.getFieldName(), + getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); + break; + } + } + } + return resultIdVal; + } + public Object getValFromData(ESMapping mapping, Map dmlData, String fieldName, String columnName) { String esType = getEsType(mapping, fieldName); Object value = dmlData.get(columnName); @@ -347,6 +370,34 @@ public class ESTemplate { return resultIdVal; } + /** + * 将dml的data, old转换为es的data + * + * @param mapping 配置mapping + * @param dmlData dml data + * @param esFieldData es data + * @return 返回 id 值 + */ + public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, Map dmlOld, + Map esFieldData) { + SchemaItem schemaItem = mapping.getSchemaItem(); + String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); + Object resultIdVal = null; + for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { + String columnName = fieldItem.getColumnItems().iterator().next().getColumnName(); + + if (fieldItem.getFieldName().equals(idFieldName)) { + resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName); + } + + if (dmlOld.get(columnName) != null && !mapping.getSkips().contains(fieldItem.getFieldName())) { + esFieldData.put(fieldItem.getFieldName(), + getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); + } + } + return resultIdVal; + } + /** * es 字段类型本地缓存 */ diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java index f2f9f4e3..2406f301 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java @@ -51,6 +51,32 @@ public class UserSyncJoinOneTest { Assert.assertEquals("Eric_", response.getSource().get("_name")); } + @Test + public void updateTest02() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("UPDATE"); + dml.setDatabase("mytest"); + dml.setTable("user"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("name", "Eric"); + dml.setData(dataList); + List> oldList = new ArrayList<>(); + Map old = new LinkedHashMap<>(); + oldList.add(old); + old.put("name", "Eric2"); + dml.setOld(oldList); + + esAdapter.getEsSyncService().sync(dml); + + GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + Assert.assertEquals("Eric_", response.getSource().get("_name")); + } + @After public void after() { esAdapter.destroy(); diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java index 16a25bea..a772d754 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java @@ -54,6 +54,35 @@ public class UserSyncSingleTest { Assert.assertEquals("Eric", response.getSource().get("_name")); } + /** + * 单表更新 + */ + @Test + public void updateTest02() { + Dml dml = new Dml(); + dml.setDestination("example"); + dml.setTs(new Date().getTime()); + dml.setType("UPDATE"); + dml.setDatabase("mytest"); + dml.setTable("user"); + List> dataList = new ArrayList<>(); + Map data = new LinkedHashMap<>(); + dataList.add(data); + data.put("id", 1L); + data.put("name", "Eric2"); + dml.setData(dataList); + List> oldList = new ArrayList<>(); + Map old = new LinkedHashMap<>(); + oldList.add(old); + old.put("name", "Eric"); + dml.setOld(oldList); + + esAdapter.getEsSyncService().sync(dml); + + GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + Assert.assertEquals("Eric2", response.getSource().get("_name")); + } + @Test public void ttt() { SearchResponse response = esAdapter.getTransportClient() diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index b8de5269..2e861642 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -1,16 +1,14 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.List; -import java.util.concurrent.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; -import org.apache.kafka.clients.consumer.CommitFailedException; import org.apache.kafka.common.errors.WakeupException; import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.protocol.Message; /** * kafka对应的client适配器工作线程 diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 3e0e0dc4..9fe6aa2d 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -27,11 +27,11 @@ canal.conf: - instance: example groups: - outAdapters: - - name: es - hosts: 127.0.0.1:9300 - properties: - cluster.name: elasticsearch -# - name: logger +# - name: es +# hosts: 127.0.0.1:9300 +# properties: +# cluster.name: elasticsearch + - name: logger # - name: hbase # properties: # hbase.zookeeper.quorum: ${hbase.zookeeper.quorum} diff --git a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java index c7b7978d..8f95b212 100644 --- a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java +++ b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java @@ -27,7 +27,7 @@ public class LoggerAdapterExample implements OuterAdapter { @Override public void sync(Dml dml) { - logger.info("DML: ", dml.toString()); + logger.info("DML: {}", dml.toString()); } @Override