From 8f76b33214a39dd8a3afe80ad166bb1f33e1ff51 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 25 Jan 2019 17:51:38 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E8=A7=A3=E6=9E=90=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/adapter/es/config/SchemaItem.java | 14 ++++++++++++++ .../canal/client/adapter/es/config/SqlParser.java | 15 +++++++++------ .../client/adapter/es/test/SqlParseTest.java | 2 +- 3 files changed, 24 insertions(+), 7 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java index fc6e076c..5a3ab5b1 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java @@ -57,6 +57,11 @@ public class SchemaItem { this.selectFields = selectFields; } + public String toSql() { + // todo + return null; + } + public Map> getTableItemAliases() { if (tableItemAliases == null) { synchronized (SchemaItem.class) { @@ -316,6 +321,7 @@ public class SchemaItem { public static class FieldItem { private String fieldName; + private String expr; private List columnItems = new ArrayList<>(); private List owners = new ArrayList<>(); @@ -330,6 +336,14 @@ public class SchemaItem { this.fieldName = fieldName; } + public String getExpr() { + return expr; + } + + public void setExpr(String expr) { + this.expr = expr; + } + public List getColumnItems() { return columnItems; } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java index 0c2d7b2c..a0e4d83d 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java @@ -25,7 +25,7 @@ import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.TableItem; /** * ES同步指定sql格式解析 - * + * * @author rewerma 2018-10-26 下午03:45:49 * @version 1.0.0 */ @@ -33,7 +33,7 @@ public class SqlParser { /** * 解析sql - * + * * @param sql sql * @return 视图对象 */ @@ -66,7 +66,7 @@ public class SqlParser { /** * 归集字段 - * + * * @param sqlSelectQueryBlock sqlSelectQueryBlock * @return 字段属性列表 */ @@ -74,6 +74,7 @@ public class SqlParser { return sqlSelectQueryBlock.getSelectList().stream().map(selectItem -> { FieldItem fieldItem = new FieldItem(); fieldItem.setFieldName(selectItem.getAlias()); + fieldItem.setExpr(selectItem.toString()); visitColumn(selectItem.getExpr(), fieldItem); return fieldItem; }).collect(Collectors.toList()); @@ -81,7 +82,7 @@ public class SqlParser { /** * 解析字段 - * + * * @param expr sql expr * @param fieldItem 字段属性 */ @@ -91,6 +92,7 @@ public class SqlParser { SQLIdentifierExpr identifierExpr = (SQLIdentifierExpr) expr; if (fieldItem.getFieldName() == null) { fieldItem.setFieldName(identifierExpr.getName()); + fieldItem.setExpr(identifierExpr.toString()); } ColumnItem columnItem = new ColumnItem(); columnItem.setColumnName(identifierExpr.getName()); @@ -101,6 +103,7 @@ public class SqlParser { SQLPropertyExpr sqlPropertyExpr = (SQLPropertyExpr) expr; if (fieldItem.getFieldName() == null) { fieldItem.setFieldName(sqlPropertyExpr.getName()); + fieldItem.setExpr(sqlPropertyExpr.toString()); } fieldItem.getOwners().add(sqlPropertyExpr.getOwnernName()); ColumnItem columnItem = new ColumnItem(); @@ -123,7 +126,7 @@ public class SqlParser { /** * 解析表 - * + * * @param schemaItem 视图对象 * @param sqlTableSource sqlTableSource * @param tableItems 表对象列表 @@ -178,7 +181,7 @@ public class SqlParser { /** * 解析on条件 - * + * * @param expr sql expr * @param tableItem 表对象 */ diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java index d07defe9..6c27e38b 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java @@ -29,7 +29,7 @@ public class SqlParseTest { Assert.assertFalse(tableItem.isMain()); Assert.assertTrue(tableItem.isSubQuery()); // 通过字段名找 FieldItem - List fieldItems = schemaItem.getColumnFields().get(tableItem.getAlias() + ".label".toLowerCase()); + List fieldItems = schemaItem.getColumnFields().get(tableItem.getAlias() + ".labels".toLowerCase()); fieldItems.forEach( fieldItem -> Assert.assertEquals("c.labels", fieldItem.getOwner() + "." + fieldItem.getFieldName())); From 83b760494d1de49e531e0f4b7763204877d286c5 Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 13 Feb 2019 15:19:09 +0800 Subject: [PATCH 2/6] =?UTF-8?q?ES=20adapter=E4=BB=A3=E7=A0=81=E9=87=8D?= =?UTF-8?q?=E6=96=B0=E6=95=B4=E7=90=86=20=E6=96=B0=E5=A2=9Eupsert=E5=B1=9E?= =?UTF-8?q?=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/adapter/es/ESAdapter.java | 7 +- .../adapter/es/config/ESSyncConfig.java | 23 +- .../client/adapter/es/config/SchemaItem.java | 10 +- .../adapter/es/service/ESEtlService.java | 152 ++- .../adapter/es/service/ESSyncService.java | 66 +- .../client/adapter/es/support/ESTemplate.java | 960 ++++++++---------- .../src/main/resources/es/mytest_user.yml | 1 + 7 files changed, 545 insertions(+), 674 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index edd651f9..13e21531 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -146,9 +146,10 @@ public class ESAdapter implements OuterAdapter { for (Dml dml : dmls) { sync(dml); } + esSyncService.commit(); // 批次统一提交 } - public void sync(Dml dml) { + private void sync(Dml dml) { String database = dml.getDatabase(); String table = dml.getTable(); Map configMap; @@ -174,7 +175,7 @@ public class ESAdapter implements OuterAdapter { DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); ESEtlService esEtlService = new ESEtlService(transportClient, config); if (dataSource != null) { - return esEtlService.importData(params, false); + return esEtlService.importData(params); } else { etlResult.setSucceeded(false); etlResult.setErrorMessage("DataSource not found"); @@ -188,7 +189,7 @@ public class ESAdapter implements OuterAdapter { // 取所有的destination为task的配置 if (configTmp.getDestination().equals(task)) { ESEtlService esEtlService = new ESEtlService(transportClient, configTmp); - EtlResult etlRes = esEtlService.importData(params, false); + EtlResult etlRes = esEtlService.importData(params); if (!etlRes.getSucceeded()) { resSuccess = false; resultMsg.append(etlRes.getErrorMessage()).append("\n"); diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java index 09c2f4e0..5f9e7493 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java @@ -30,8 +30,8 @@ public class ESSyncConfig { if (esMapping._type == null) { throw new NullPointerException("esMapping._type"); } - if (esMapping._id == null && esMapping.pk == null) { - throw new NullPointerException("esMapping._id and esMapping.pk"); + if (esMapping._id == null) { + throw new NullPointerException("esMapping._id"); } if (esMapping.sql == null) { throw new NullPointerException("esMapping.sql"); @@ -83,8 +83,9 @@ public class ESSyncConfig { private String _index; private String _type; private String _id; + private boolean upsert = false; private String pk; - private String parent; + // private String parent; private String sql; // 对象字段, 例: objFields: // - _labels: array:; @@ -121,6 +122,14 @@ public class ESSyncConfig { this._id = _id; } + public boolean isUpsert() { + return upsert; + } + + public void setUpsert(boolean upsert) { + this.upsert = upsert; + } + public String getPk() { return pk; } @@ -129,14 +138,6 @@ public class ESSyncConfig { this.pk = pk; } - public String getParent() { - return parent; - } - - public void setParent(String parent) { - this.parent = parent; - } - public Map getObjFields() { return objFields; } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java index 4d45c525..a35dd0f0 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java @@ -134,11 +134,11 @@ public class SchemaItem { } public FieldItem getIdFieldItem(ESMapping mapping) { - if (mapping.get_id() != null) { - return getSelectFields().get(mapping.get_id()); - } else { - return getSelectFields().get(mapping.getPk()); - } + //TODO if (mapping.get_id() != null) { + return getSelectFields().get(mapping.get_id()); + // } else { + // return getSelectFields().get(mapping.getPk()); + // } } public static class TableItem { diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java index 51f0a924..29eac386 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java @@ -1,6 +1,9 @@ package com.alibaba.otter.canal.client.adapter.es.service; -import java.util.*; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -13,11 +16,8 @@ import javax.sql.DataSource; import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; -import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.transport.TransportClient; -import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; -import org.elasticsearch.search.SearchHit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,7 +51,7 @@ public class ESEtlService { this.config = config; } - public EtlResult importData(List params, boolean bulk) { + public EtlResult importData(List params) { EtlResult etlResult = new EtlResult(); AtomicLong impCount = new AtomicLong(); List errMsg = new ArrayList<>(); @@ -93,54 +93,49 @@ public class ESEtlService { logger.debug("etl sql : {}", mapping.getSql()); } - if (bulk) { - // 获取总数 - String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; - long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> { - Long count = null; - try { - if (rs.next()) { - count = ((Number) rs.getObject(1)).longValue(); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); + // 获取总数 + String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; + long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> { + Long count = null; + try { + if (rs.next()) { + count = ((Number) rs.getObject(1)).longValue(); } - return count == null ? 0L : count; - }); - - // 当大于1万条记录时开启多线程 - if (cnt >= 10000) { - int threadCount = 3; // TODO 从配置读取默认为3 - long perThreadCnt = cnt / threadCount; - ExecutorService executor = Executors.newFixedThreadPool(threadCount); - List> futures = new ArrayList<>(threadCount); - for (int i = 0; i < threadCount; i++) { - long offset = i * perThreadCnt; - Long size = null; - if (i != threadCount - 1) { - size = perThreadCnt; - } - String sqlFinal; - if (size != null) { - sqlFinal = sql + " LIMIT " + offset + "," + size; - } else { - sqlFinal = sql + " LIMIT " + offset + "," + cnt; - } - Future future = executor - .submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg)); - futures.add(future); - } - - for (Future future : futures) { - future.get(); - } - - executor.shutdown(); - } else { - executeSqlImport(dataSource, sql, mapping, impCount, errMsg); + } catch (Exception e) { + logger.error(e.getMessage(), e); } + return count == null ? 0L : count; + }); + + // 当大于1万条记录时开启多线程 + if (cnt >= 10000) { + int threadCount = 3; // 从配置读取默认为3 + long perThreadCnt = cnt / threadCount; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + List> futures = new ArrayList<>(threadCount); + for (int i = 0; i < threadCount; i++) { + long offset = i * perThreadCnt; + Long size = null; + if (i != threadCount - 1) { + size = perThreadCnt; + } + String sqlFinal; + if (size != null) { + sqlFinal = sql + " LIMIT " + offset + "," + size; + } else { + sqlFinal = sql + " LIMIT " + offset + "," + cnt; + } + Future future = executor + .submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg)); + futures.add(future); + } + + for (Future future : futures) { + future.get(); + } + + executor.shutdown(); } else { - logger.info("自动ETL，无需统计记录总条数，直接进行ETL, index: {}", esIndex); executeSqlImport(dataSource, sql, mapping, impCount, errMsg); } @@ -205,32 +200,35 @@ public class ESEtlService { } if (idVal != null) { - if (mapping.getParent() == null) { + if (mapping.isUpsert()) { + bulkRequestBuilder.add(transportClient + .prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString()) + .setDoc(esFieldData) + .setDocAsUpsert(true)); + } else { bulkRequestBuilder.add(transportClient .prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString()) .setSource(esFieldData)); - } else { - // ignore } } else { - idVal = rs.getObject(mapping.getPk()); - if (mapping.getParent() == null) { - // 删除pk对应的数据 - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) - .get(); - for (SearchHit hit : response.getHits()) { - bulkRequestBuilder.add(transportClient - .prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); - } - - bulkRequestBuilder - .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()) - .setSource(esFieldData)); - } else { - // ignore - } + // TODO idVal = rs.getObject(mapping.getPk()); + // if (mapping.getParent() == null) { + // // 删除pk对应的数据 + // SearchResponse response = transportClient.prepareSearch(mapping.get_index()) + // .setTypes(mapping.get_type()) + // .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) + // .get(); + // for (SearchHit hit : response.getHits()) { + // bulkRequestBuilder.add(transportClient + // .prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); + // } + // + // bulkRequestBuilder + // .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()) + // .setSource(esFieldData)); + // } else { + // // ignore + // } } if (bulkRequestBuilder.numberOfActions() % mapping.getCommitBatch() == 0 @@ -238,11 +236,11 @@ public class ESEtlService { long esBatchBegin = System.currentTimeMillis(); BulkResponse rp = bulkRequestBuilder.execute().actionGet(); if (rp.hasFailures()) { - this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent())); + this.processFailBulkResponse(rp, /* TODO Objects.nonNull(mapping.getParent()) */ false); } - if (logger.isDebugEnabled()) { - logger.debug("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", + if (logger.isTraceEnabled()) { + logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", (System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - esBatchBegin), bulkRequestBuilder.numberOfActions(), @@ -259,10 +257,10 @@ public class ESEtlService { long esBatchBegin = System.currentTimeMillis(); BulkResponse rp = bulkRequestBuilder.execute().actionGet(); if (rp.hasFailures()) { - this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent())); + this.processFailBulkResponse(rp, /* TODO Objects.nonNull(mapping.getParent()) */false); } - if (logger.isDebugEnabled()) { - logger.debug("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", + if (logger.isTraceEnabled()) { + logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", (System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - esBatchBegin), bulkRequestBuilder.numberOfActions(), 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 dba99028..9faab23e 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 @@ -203,7 +203,7 @@ public class ESSyncService { // ------主表 查询sql来更新------ if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { ESMapping mapping = config.getEsMapping(); - String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); + String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName); boolean idFieldSimple = true; @@ -344,14 +344,7 @@ public class ESSyncService { mapping.get_index(), idVal); } - boolean result = esTemplate.delete(mapping, idVal); - if (!result) { - logger.error("Main table delete es index error, destination:{}, table: {}, index: {}, id: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index(), - idVal); - } + esTemplate.delete(mapping, idVal); } else { // ------主键带函数, 查询sql获取主键删除------ mainTableDelete(config, dml, data); @@ -417,14 +410,7 @@ public class ESSyncService { mapping.get_index(), idVal); } - boolean result = esTemplate.insert(mapping, idVal, esFieldData); - if (!result) { - logger.error("Single table insert to es index error, destination:{}, table: {}, index: {}, id: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index(), - idVal); - } + esTemplate.insert(mapping, idVal, esFieldData); } /** @@ -461,15 +447,7 @@ public class ESSyncService { mapping.get_index(), idVal); } - boolean result = esTemplate.insert(mapping, idVal, esFieldData); - if (!result) { - logger.error( - "Main table insert to es index by query sql error, destination:{}, table: {}, index: {}, id: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index(), - idVal); - } + esTemplate.insert(mapping, idVal, esFieldData); } } catch (Exception e) { throw new RuntimeException(e); @@ -504,15 +482,7 @@ public class ESSyncService { mapping.get_index(), idVal); } - boolean result = esTemplate.delete(mapping, idVal); - if (!result) { - logger.error( - "Main table delete to es index by query sql error, destination:{}, table: {}, index: {}, id: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index(), - idVal); - } + esTemplate.delete(mapping, idVal); } } catch (Exception e) { throw new RuntimeException(e); @@ -799,14 +769,7 @@ public class ESSyncService { mapping.get_index(), idVal); } - boolean result = esTemplate.update(mapping, idVal, esFieldData); - if (!result) { - logger.error("Main table update to es index error, destination:{}, table: {}, index: {}, id: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index(), - idVal); - } + esTemplate.update(mapping, idVal, esFieldData); } /** @@ -843,15 +806,7 @@ public class ESSyncService { mapping.get_index(), idVal); } - boolean result = esTemplate.update(mapping, 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); - } + esTemplate.update(mapping, idVal, esFieldData); } } catch (Exception e) { throw new RuntimeException(e); @@ -859,4 +814,11 @@ public class ESSyncService { return 0; }); } + + /** + * 提交批次 + */ + public void commit() { + esTemplate.commit(); + } } 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 9be2d801..a9b6599e 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 @@ -1,526 +1,434 @@ -package com.alibaba.otter.canal.client.adapter.es.support; - -import java.sql.ResultSet; -import java.sql.SQLException; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.TimeUnit; - -import javax.sql.DataSource; - -import com.alibaba.fastjson.JSON; -import org.elasticsearch.action.bulk.BulkItemResponse; -import org.elasticsearch.action.bulk.BulkRequestBuilder; -import org.elasticsearch.action.bulk.BulkResponse; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.client.transport.TransportClient; -import org.elasticsearch.cluster.metadata.MappingMetaData; -import org.elasticsearch.common.collect.ImmutableOpenMap; -import org.elasticsearch.index.query.BoolQueryBuilder; -import org.elasticsearch.index.query.QueryBuilder; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.index.reindex.BulkByScrollResponse; -import org.elasticsearch.index.reindex.UpdateByQueryAction; -import org.elasticsearch.index.reindex.UpdateByQueryRequestBuilder; -import org.elasticsearch.rest.RestStatus; -import org.elasticsearch.script.Script; -import org.elasticsearch.script.ScriptType; -import org.elasticsearch.search.SearchHit; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -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; -import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; - -/** - * ES 操作模板 - * - * @author rewerma 2018-11-01 - * @version 1.0.0 - */ -public class ESTemplate { - - private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class); - - private static final int MAX_BATCH_SIZE = 1000; - - private TransportClient transportClient; - - public ESTemplate(TransportClient transportClient){ - this.transportClient = transportClient; - } - - /** - * 插入数据 - * - * @param mapping - * @param pkVal - * @param esFieldData - * @return - */ - public boolean insert(ESMapping mapping, Object pkVal, Map esFieldData) { - BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); - if (mapping.get_id() != null) { - bulkRequestBuilder - .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setSource(esFieldData)); - } else { - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .setSize(MAX_BATCH_SIZE) - .get(); - for (SearchHit hit : response.getHits()) { - bulkRequestBuilder - .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); - } - bulkRequestBuilder - .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()).setSource(esFieldData)); - } - return commitBulkRequest(bulkRequestBuilder); - } - - /** - * 根据主键更新数据 - * - * @param mapping - * @param pkVal - * @param esFieldData - * @return - */ - public boolean update(ESMapping mapping, Object pkVal, Map esFieldData) { - BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); - append4Update(bulkRequestBuilder, mapping, pkVal, esFieldData); - return commitBulkRequest(bulkRequestBuilder); - } - - public void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal, - Map esFieldData) { - if (mapping.get_id() != null) { - bulkRequestBuilder - .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData)); - } else { - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .setSize(MAX_BATCH_SIZE) - .get(); - for (SearchHit hit : response.getHits()) { - bulkRequestBuilder - .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) - .setDoc(esFieldData)); - } - } - } - - /** - * update by query - * - * @param config - * @param paramsTmp - * @param esFieldData - * @return - */ - public boolean updateByQuery(ESSyncConfig config, Map paramsTmp, Map esFieldData) { - if (paramsTmp.isEmpty()) { - return false; - } - ESMapping mapping = config.getEsMapping(); - BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); - paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value))); - - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setSize(0) - .setQuery(queryBuilder) - .get(); - long count = response.getHits().getTotalHits(); - // 如果更新量大于Max, 查询sql批量更新 - if (count > MAX_BATCH_SIZE) { - BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); - - DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); - // 查询sql更新 - StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE "); - paramsTmp.forEach( - (fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND ")); - int len = sql.length(); - sql.delete(len - 4, len); - ESSyncUtil.sqlRS(ds, sql.toString(), rs -> { - int exeCount = 1; - try { - BulkRequestBuilder bulkRequestBuilderTmp = bulkRequestBuilder; - while (rs.next()) { - Object idVal = getIdValFromRS(mapping, rs); - append4Update(bulkRequestBuilderTmp, mapping, idVal, esFieldData); - - if (exeCount % mapping.getCommitBatch() == 0 && bulkRequestBuilderTmp.numberOfActions() > 0) { - commitBulkRequest(bulkRequestBuilderTmp); - bulkRequestBuilderTmp = transportClient.prepareBulk(); - } - exeCount++; - } - - if (bulkRequestBuilder.numberOfActions() > 0) { - commitBulkRequest(bulkRequestBuilderTmp); - } - } catch (Exception e) { - throw new RuntimeException(e); - } - return 0; - }); - return true; - } else { - return updateByQuery(mapping, queryBuilder, esFieldData, 1); - } - } - - private boolean updateByQuery(ESMapping mapping, QueryBuilder queryBuilder, Map esFieldData, - int counter) { - if (CollectionUtils.isEmpty(esFieldData)) { - return true; - } - - StringBuilder sb = new StringBuilder(); - esFieldData.forEach((key, value) -> { - if (value instanceof Map) { - Map mapValue = (Map) value; - if (mapValue.containsKey("lon") && mapValue.containsKey("lat") && mapValue.size() == 2) { - sb.append("ctx._source") - .append("['") - .append(key) - .append("']") - .append(" = [") - .append(mapValue.get("lon")) - .append(", ") - .append(mapValue.get("lat")) - .append("];"); - } else { - sb.append("ctx._source").append("[\"").append(key).append("\"]").append(" = "); - sb.append(JSON.toJSONString(value)); - sb.append(";"); - } - } else if (value instanceof List) { - sb.append("ctx._source").append("[\"").append(key).append("\"]").append(" = "); - sb.append(JSON.toJSONString(value)); - sb.append(";"); - } else if (value instanceof String) { - sb.append("ctx._source") - .append("['") - .append(key) - .append("']") - .append(" = '") - .append(value) - .append("';"); - } else { - sb.append("ctx._source").append("['").append(key).append("']").append(" = ").append(value).append(";"); - } - }); - String scriptLine = sb.toString(); - if (logger.isTraceEnabled()) { - logger.trace(scriptLine); - } - - UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(transportClient); - updateByQuery.source(mapping.get_index()) - .abortOnVersionConflict(false) - .filter(queryBuilder) - .script(new Script(ScriptType.INLINE, "painless", scriptLine, Collections.emptyMap())); - - BulkByScrollResponse response = updateByQuery.get(); - if (logger.isTraceEnabled()) { - logger.trace("updateByQuery response: {}", response.getStatus()); - } - if (!CollectionUtils.isEmpty(response.getSearchFailures())) { - logger.error("script update_for_search has search error: " + response.getBulkFailures()); - return false; - } - - if (!CollectionUtils.isEmpty(response.getBulkFailures())) { - logger.error("script update_for_search has update error: " + response.getBulkFailures()); - return false; - } - - if (response.getStatus().getVersionConflicts() > 0) { - if (counter >= 3) { - logger.error("第 {} 次执行updateByQuery, 依旧存在分片版本冲突，不再继续重试。", counter); - return false; - } - logger.warn("本次updateByQuery存在分片版本冲突，准备重新执行..."); - try { - TimeUnit.SECONDS.sleep(1); - } catch (InterruptedException e) { - // ignore - } - return updateByQuery(mapping, queryBuilder, esFieldData, ++counter); - } - - return true; - } - - /** - * 通过主键删除数据 - * - * @param mapping - * @param pkVal - * @return - */ - public boolean delete(ESMapping mapping, Object pkVal) { - BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); - if (mapping.get_id() != null) { - bulkRequestBuilder - .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); - } else { - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .setSize(MAX_BATCH_SIZE) - .get(); - for (SearchHit hit : response.getHits()) { - bulkRequestBuilder - .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); - } - } - return commitBulkRequest(bulkRequestBuilder); - } - - /** - * 批量提交 - * - * @param bulkRequestBuilder - * @return - */ - private static boolean commitBulkRequest(BulkRequestBuilder bulkRequestBuilder) { - if (bulkRequestBuilder.numberOfActions() > 0) { - BulkResponse response = bulkRequestBuilder.execute().actionGet(); - if (response.hasFailures()) { - for (BulkItemResponse itemResponse : response.getItems()) { - if (!itemResponse.isFailed()) { - continue; - } - - if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) { - logger.warn(itemResponse.getFailureMessage()); - } else { - logger.error("ES sync commit error: {}", itemResponse.getFailureMessage()); - } - } - } - - return !response.hasFailures(); - } - return true; - } - - public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, - String columnName) throws SQLException { - String esType = getEsType(mapping, fieldName); - - Object value = resultSet.getObject(columnName); - if (value instanceof Boolean) { - if (!"boolean".equals(esType)) { - value = resultSet.getByte(columnName); - } - } - - // 如果是对象类型 - if (mapping.getObjFields().containsKey(fieldName)) { - return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName)); - } else { - return ESSyncUtil.typeConvert(value, esType); - } - } - - public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, - 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()) { - Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); - - if (fieldItem.getFieldName().equals(idFieldName)) { - resultIdVal = value; - } - - if (!fieldItem.getFieldName().equals(mapping.get_id()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { - esFieldData.put(fieldItem.getFieldName(), value); - } - } - return resultIdVal; - } - - public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) 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()) { - Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); - - if (fieldItem.getFieldName().equals(idFieldName)) { - resultIdVal = value; - break; - } - } - 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.containsKey(columnItem.getColumnName()) - && !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); - if (value instanceof Byte) { - if ("boolean".equals(esType)) { - value = ((Byte) value).intValue() != 0; - } - } - - // 如果是对象类型 - if (mapping.getObjFields().containsKey(fieldName)) { - return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName)); - } else { - return ESSyncUtil.typeConvert(value, esType); - } - } - - /** - * 将dml的data转换为es的data - * - * @param mapping 配置mapping - * @param dmlData dml data - * @param esFieldData es data - * @return 返回 id 值 - */ - public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, - 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(); - Object value = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName); - - if (fieldItem.getFieldName().equals(idFieldName)) { - resultIdVal = value; - } - - if (!fieldItem.getFieldName().equals(mapping.get_id()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { - esFieldData.put(fieldItem.getFieldName(), value); - } - } - 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 字段类型本地缓存 - */ - private static ConcurrentMap> esFieldTypes = new ConcurrentHashMap<>(); - - /** - * 获取es mapping中的属性类型 - * - * @param mapping mapping配置 - * @param fieldName 属性名 - * @return 类型 - */ - @SuppressWarnings("unchecked") - private String getEsType(ESMapping mapping, String fieldName) { - String key = mapping.get_index() + "-" + mapping.get_type(); - Map fieldType = esFieldTypes.get(key); - if (fieldType == null) { - ImmutableOpenMap mappings; - try { - mappings = transportClient.admin() - .cluster() - .prepareState() - .execute() - .actionGet() - .getState() - .getMetaData() - .getIndices() - .get(mapping.get_index()) - .getMappings(); - } catch (NullPointerException e) { - throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); - } - MappingMetaData mappingMetaData = mappings.get(mapping.get_type()); - if (mappingMetaData == null) { - throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); - } - - fieldType = new LinkedHashMap<>(); - - Map sourceMap = mappingMetaData.getSourceAsMap(); - Map esMapping = (Map) sourceMap.get("properties"); - for (Map.Entry entry : esMapping.entrySet()) { - Map value = (Map) entry.getValue(); - if (value.containsKey("properties")) { - fieldType.put(entry.getKey(), "object"); - } else { - fieldType.put(entry.getKey(), (String) value.get("type")); - } - } - esFieldTypes.put(key, fieldType); - } - - return fieldType.get(fieldName); - } -} +package com.alibaba.otter.canal.client.adapter.es.support; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import javax.sql.DataSource; + +import org.elasticsearch.action.bulk.BulkItemResponse; +import org.elasticsearch.action.bulk.BulkRequestBuilder; +import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.client.transport.TransportClient; +import org.elasticsearch.cluster.metadata.MappingMetaData; +import org.elasticsearch.common.collect.ImmutableOpenMap; +import org.elasticsearch.index.query.BoolQueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.rest.RestStatus; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +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; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; + +/** + * ES 操作模板 + * + * @author rewerma 2018-11-01 + * @version 1.0.0 + */ +public class ESTemplate { + + private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class); + + private static final int MAX_BATCH_SIZE = 1000; + + private TransportClient transportClient; + + private volatile BulkRequestBuilder bulkRequestBuilder; + + public ESTemplate(TransportClient transportClient){ + this.transportClient = transportClient; + this.bulkRequestBuilder = transportClient.prepareBulk(); + } + + public BulkRequestBuilder getBulk() { + return bulkRequestBuilder; + } + + /** + * 插入数据 + * + * @param mapping + * @param pkVal + * @param esFieldData + * @return + */ + public void insert(ESMapping mapping, Object pkVal, Map esFieldData) { + if (mapping.get_id() != null) { + if (mapping.isUpsert()) { + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) + .setDoc(esFieldData) + .setDocAsUpsert(true)); + } else { + getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) + .setSource(esFieldData)); + } + commitBatch(); + } else { + // TODO SearchResponse response = + // transportClient.prepareSearch(mapping.get_index()) + // .setTypes(mapping.get_type()) + // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + // .setSize(MAX_BATCH_SIZE) + // .get(); + // for (SearchHit hit : response.getHits()) { + // bulkRequestBuilder + // .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), + // hit.getId())); + // } + // bulkRequestBuilder + // .add(transportClient.prepareIndex(mapping.get_index(), + // mapping.get_type()).setSource(esFieldData)); + } + // return commitBulkRequest(bulkRequestBuilder); + } + + /** + * 根据主键更新数据 + * + * @param mapping + * @param pkVal + * @param esFieldData + * @return + */ + public void update(ESMapping mapping, Object pkVal, Map esFieldData) { + append4Update(getBulk(), mapping, pkVal, esFieldData); + commitBatch(); + } + + private void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal, + Map esFieldData) { + if (mapping.get_id() != null) { + bulkRequestBuilder + .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) + .setDoc(esFieldData)); + } else { + // TODO SearchResponse response = + // transportClient.prepareSearch(mapping.get_index()) + // .setTypes(mapping.get_type()) + // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + // .setSize(MAX_BATCH_SIZE) + // .get(); + // for (SearchHit hit : response.getHits()) { + // bulkRequestBuilder + // .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), + // hit.getId()) + // .setDoc(esFieldData)); + // } + } + } + + /** + * TODO XXX update by query + * + * @param config + * @param paramsTmp + * @param esFieldData + * @return + */ + public boolean updateByQuery(ESSyncConfig config, Map paramsTmp, Map esFieldData) { + if (paramsTmp.isEmpty()) { + return false; + } + ESMapping mapping = config.getEsMapping(); + BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); + paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value))); + + // 查询sql批量更新 + DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); + StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE "); + paramsTmp.forEach( + (fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND ")); + int len = sql.length(); + sql.delete(len - 4, len); + Integer syncCount = (Integer) ESSyncUtil.sqlRS(ds, sql.toString(), rs -> { + int count = 0; + try { + while (rs.next()) { + Object idVal = getIdValFromRS(mapping, rs); + append4Update(getBulk(), mapping, idVal, esFieldData); + commitBatch(); + count++; + } + } catch (Exception e) { + throw new RuntimeException(e); + } + return count; + }); + if (logger.isTraceEnabled()) { + logger.trace("Update ES by query effect {} records", syncCount); + } + return true; + } + + /** + * 通过主键删除数据 + * + * @param mapping + * @param pkVal + * @return + */ + public void delete(ESMapping mapping, Object pkVal) { + if (mapping.get_id() != null) { + getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); + commitBatch(); + } else { + // TODO SearchResponse response = + // transportClient.prepareSearch(mapping.get_index()) + // .setTypes(mapping.get_type()) + // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + // .setSize(MAX_BATCH_SIZE) + // .get(); + // for (SearchHit hit : response.getHits()) { + // bulkRequestBuilder + // .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), + // hit.getId())); + // } + } + // return commitBulkRequest(bulkRequestBuilder); + } + + /** + * 提交批次 + */ + public void commit() { + if (getBulk().numberOfActions() >= 0) { + BulkResponse response = getBulk().execute().actionGet(); + if (response.hasFailures()) { + for (BulkItemResponse itemResponse : response.getItems()) { + if (!itemResponse.isFailed()) { + continue; + } + + if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) { + logger.error(itemResponse.getFailureMessage()); + } else { + throw new RuntimeException("ES sync commit error" + itemResponse.getFailureMessage()); + } + } + } + } + } + + /** + * 如果大于批量则提交批次 + */ + private void commitBatch() { + if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) { + commit(); + } + } + + public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, + String columnName) throws SQLException { + String esType = getEsType(mapping, fieldName); + + Object value = resultSet.getObject(columnName); + if (value instanceof Boolean) { + if (!"boolean".equals(esType)) { + value = resultSet.getByte(columnName); + } + } + + // 如果是对象类型 + if (mapping.getObjFields().containsKey(fieldName)) { + return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName)); + } else { + return ESSyncUtil.typeConvert(value, esType); + } + } + + public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, + Map esFieldData) throws SQLException { + SchemaItem schemaItem = mapping.getSchemaItem(); + String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + Object resultIdVal = null; + for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { + Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); + + if (fieldItem.getFieldName().equals(idFieldName)) { + resultIdVal = value; + } + + if (!fieldItem.getFieldName().equals(mapping.get_id()) + && !mapping.getSkips().contains(fieldItem.getFieldName())) { + esFieldData.put(fieldItem.getFieldName(), value); + } + } + return resultIdVal; + } + + public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException { + SchemaItem schemaItem = mapping.getSchemaItem(); + String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + Object resultIdVal = null; + for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { + Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); + + if (fieldItem.getFieldName().equals(idFieldName)) { + resultIdVal = value; + break; + } + } + return resultIdVal; + } + + public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map dmlOld, + Map esFieldData) throws SQLException { + SchemaItem schemaItem = mapping.getSchemaItem(); + String idFieldName = mapping.get_id(); // TODO == 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.containsKey(columnItem.getColumnName()) + && !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); + if (value instanceof Byte) { + if ("boolean".equals(esType)) { + value = ((Byte) value).intValue() != 0; + } + } + + // 如果是对象类型 + if (mapping.getObjFields().containsKey(fieldName)) { + return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName)); + } else { + return ESSyncUtil.typeConvert(value, esType); + } + } + + /** + * 将dml的data转换为es的data + * + * @param mapping 配置mapping + * @param dmlData dml data + * @param esFieldData es data + * @return 返回 id 值 + */ + public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, + Map esFieldData) { + SchemaItem schemaItem = mapping.getSchemaItem(); + String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + Object resultIdVal = null; + for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { + String columnName = fieldItem.getColumnItems().iterator().next().getColumnName(); + Object value = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName); + + if (fieldItem.getFieldName().equals(idFieldName)) { + resultIdVal = value; + } + + if (!fieldItem.getFieldName().equals(mapping.get_id()) + && !mapping.getSkips().contains(fieldItem.getFieldName())) { + esFieldData.put(fieldItem.getFieldName(), value); + } + } + 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(); // TODO == 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 字段类型本地缓存 + */ + private static ConcurrentMap> esFieldTypes = new ConcurrentHashMap<>(); + + /** + * 获取es mapping中的属性类型 + * + * @param mapping mapping配置 + * @param fieldName 属性名 + * @return 类型 + */ + @SuppressWarnings("unchecked") + private String getEsType(ESMapping mapping, String fieldName) { + String key = mapping.get_index() + "-" + mapping.get_type(); + Map fieldType = esFieldTypes.get(key); + if (fieldType == null) { + ImmutableOpenMap mappings; + try { + mappings = transportClient.admin() + .cluster() + .prepareState() + .execute() + .actionGet() + .getState() + .getMetaData() + .getIndices() + .get(mapping.get_index()) + .getMappings(); + } catch (NullPointerException e) { + throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); + } + MappingMetaData mappingMetaData = mappings.get(mapping.get_type()); + if (mappingMetaData == null) { + throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); + } + + fieldType = new LinkedHashMap<>(); + + Map sourceMap = mappingMetaData.getSourceAsMap(); + Map esMapping = (Map) sourceMap.get("properties"); + for (Map.Entry entry : esMapping.entrySet()) { + Map value = (Map) entry.getValue(); + if (value.containsKey("properties")) { + fieldType.put(entry.getKey(), "object"); + } else { + fieldType.put(entry.getKey(), (String) value.get("type")); + } + } + esFieldTypes.put(key, fieldType); + } + + return fieldType.get(fieldName); + } +} diff --git a/client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml b/client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml index 9585227e..42fbf029 100644 --- a/client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml +++ b/client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml @@ -5,6 +5,7 @@ esMapping: _index: mytest_user _type: _doc _id: _id + upsert: true # pk: id sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name, a.c_time as _c_time from user a From 0ef21e41552298ecf309a15cef00679081e48242 Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 13 Feb 2019 15:47:48 +0800 Subject: [PATCH 3/6] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E6=95=B4=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../alibaba/otter/canal/client/adapter/es/ESAdapter.java | 4 +++- .../canal/client/adapter/es/service/ESSyncService.java | 9 ++++++++- .../canal/client/adapter/es/support/ESTemplate.java | 8 ++++---- 3 files changed, 15 insertions(+), 6 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 13e21531..c2e899e7 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -144,7 +144,9 @@ public class ESAdapter implements OuterAdapter { return; } for (Dml dml : dmls) { - sync(dml); + if (!dml.getIsDdl()) { + sync(dml); + } } esSyncService.commit(); // 批次统一提交 } 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 9faab23e..4cbea77a 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 @@ -71,7 +71,12 @@ public class ESSyncService { dml.getDestination()); } if (logger.isDebugEnabled()) { - logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); + StringBuilder configIndexes = new StringBuilder(); + esSyncConfigs + .forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" ")); + logger.debug("DML: {} \nEffect indexes: {}", + JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue), + configIndexes.toString()); } } } @@ -92,6 +97,8 @@ public class ESSyncService { update(config, dml); } else if (type != null && type.equalsIgnoreCase("DELETE")) { delete(config, dml); + } else { + return; } if (logger.isTraceEnabled()) { 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 a9b6599e..55ba4237 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 @@ -65,8 +65,8 @@ public class ESTemplate { if (mapping.get_id() != null) { if (mapping.isUpsert()) { getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData) - .setDocAsUpsert(true)); + .setDoc(esFieldData) + .setDocAsUpsert(true)); } else { getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) .setSource(esFieldData)); @@ -200,7 +200,7 @@ public class ESTemplate { * 提交批次 */ public void commit() { - if (getBulk().numberOfActions() >= 0) { + if (getBulk().numberOfActions() > 0) { BulkResponse response = getBulk().execute().actionGet(); if (response.hasFailures()) { for (BulkItemResponse itemResponse : response.getItems()) { @@ -369,7 +369,7 @@ public class ESTemplate { resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName); } - if (dmlOld.get(columnName) != null && !mapping.getSkips().contains(fieldItem.getFieldName())) { + if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(fieldItem.getFieldName(), getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); } From c14ed724ff9f3b4f097fe006bf3da9b025ae1837 Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 13 Feb 2019 15:59:46 +0800 Subject: [PATCH 4/6] =?UTF-8?q?update=E4=B8=AD=E5=A2=9E=E5=8A=A0upsert?= =?UTF-8?q?=E5=88=A4=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/adapter/es/support/ESTemplate.java | 54 ++++++++++--------- 1 file changed, 29 insertions(+), 25 deletions(-) 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 55ba4237..5c389fae 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 @@ -100,34 +100,12 @@ public class ESTemplate { * @return */ public void update(ESMapping mapping, Object pkVal, Map esFieldData) { - append4Update(getBulk(), mapping, pkVal, esFieldData); + append4Update(mapping, pkVal, esFieldData); commitBatch(); } - private void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal, - Map esFieldData) { - if (mapping.get_id() != null) { - bulkRequestBuilder - .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData)); - } else { - // TODO SearchResponse response = - // transportClient.prepareSearch(mapping.get_index()) - // .setTypes(mapping.get_type()) - // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - // .setSize(MAX_BATCH_SIZE) - // .get(); - // for (SearchHit hit : response.getHits()) { - // bulkRequestBuilder - // .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), - // hit.getId()) - // .setDoc(esFieldData)); - // } - } - } - /** - * TODO XXX update by query + * update by query * * @param config * @param paramsTmp @@ -154,7 +132,7 @@ public class ESTemplate { try { while (rs.next()) { Object idVal = getIdValFromRS(mapping, rs); - append4Update(getBulk(), mapping, idVal, esFieldData); + append4Update(mapping, idVal, esFieldData); commitBatch(); count++; } @@ -227,6 +205,32 @@ public class ESTemplate { } } + private void append4Update(ESMapping mapping, Object pkVal, Map esFieldData) { + if (mapping.get_id() != null) { + if (mapping.isUpsert()) { + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) + .setDoc(esFieldData) + .setDocAsUpsert(true)); + } else { + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) + .setDoc(esFieldData)); + } + } else { + // TODO SearchResponse response = + // transportClient.prepareSearch(mapping.get_index()) + // .setTypes(mapping.get_type()) + // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + // .setSize(MAX_BATCH_SIZE) + // .get(); + // for (SearchHit hit : response.getHits()) { + // bulkRequestBuilder + // .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), + // hit.getId()) + // .setDoc(esFieldData)); + // } + } + } + public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, String columnName) throws SQLException { String esType = getEsType(mapping, fieldName); From cfdf509f9a93a012fe50e82beb8a7ffcb3a8cc23 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 14 Feb 2019 15:29:27 +0800 Subject: [PATCH 5/6] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=AF=B9pk=E7=9A=84?= =?UTF-8?q?=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/es/config/ESSyncConfig.java | 4 +- .../client/adapter/es/config/SchemaItem.java | 10 +- .../adapter/es/service/ESEtlService.java | 40 +++-- .../adapter/es/service/ESSyncService.java | 95 +++++++----- .../client/adapter/es/support/ESTemplate.java | 140 ++++++++---------- 5 files changed, 145 insertions(+), 144 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java index 5f9e7493..94341fd0 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java @@ -30,8 +30,8 @@ public class ESSyncConfig { if (esMapping._type == null) { throw new NullPointerException("esMapping._type"); } - if (esMapping._id == null) { - throw new NullPointerException("esMapping._id"); + if (esMapping._id == null && esMapping.getPk() == null) { + throw new NullPointerException("esMapping._id and esMapping.pk"); } if (esMapping.sql == null) { throw new NullPointerException("esMapping.sql"); diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java index a35dd0f0..4d45c525 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java @@ -134,11 +134,11 @@ public class SchemaItem { } public FieldItem getIdFieldItem(ESMapping mapping) { - //TODO if (mapping.get_id() != null) { - return getSelectFields().get(mapping.get_id()); - // } else { - // return getSelectFields().get(mapping.getPk()); - // } + if (mapping.get_id() != null) { + return getSelectFields().get(mapping.get_id()); + } else { + return getSelectFields().get(mapping.getPk()); + } } public static class TableItem { diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java index 29eac386..827d9460 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java @@ -16,8 +16,13 @@ import javax.sql.DataSource; import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.search.SearchType; import org.elasticsearch.client.transport.TransportClient; +import org.elasticsearch.common.unit.TimeValue; +import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; +import org.elasticsearch.search.SearchHit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -153,7 +158,7 @@ public class ESEtlService { return etlResult; } - private void processFailBulkResponse(BulkResponse bulkResponse, boolean hasParent) { + private void processFailBulkResponse(BulkResponse bulkResponse) { for (BulkItemResponse response : bulkResponse.getItems()) { if (!response.isFailed()) { continue; @@ -211,24 +216,17 @@ public class ESEtlService { .setSource(esFieldData)); } } else { - // TODO idVal = rs.getObject(mapping.getPk()); - // if (mapping.getParent() == null) { - // // 删除pk对应的数据 - // SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - // .setTypes(mapping.get_type()) - // .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) - // .get(); - // for (SearchHit hit : response.getHits()) { - // bulkRequestBuilder.add(transportClient - // .prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); - // } - // - // bulkRequestBuilder - // .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()) - // .setSource(esFieldData)); - // } else { - // // ignore - // } + idVal = rs.getObject(mapping.getPk()); + SearchResponse response = transportClient.prepareSearch(mapping.get_index()) + .setTypes(mapping.get_type()) + .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) + .setSize(10000) + .get(); + for (SearchHit hit : response.getHits()) { + bulkRequestBuilder.add( + transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) + .setDoc(esFieldData)); + } } if (bulkRequestBuilder.numberOfActions() % mapping.getCommitBatch() == 0 @@ -236,7 +234,7 @@ public class ESEtlService { long esBatchBegin = System.currentTimeMillis(); BulkResponse rp = bulkRequestBuilder.execute().actionGet(); if (rp.hasFailures()) { - this.processFailBulkResponse(rp, /* TODO Objects.nonNull(mapping.getParent()) */ false); + this.processFailBulkResponse(rp); } if (logger.isTraceEnabled()) { @@ -257,7 +255,7 @@ public class ESEtlService { long esBatchBegin = System.currentTimeMillis(); BulkResponse rp = bulkRequestBuilder.execute().actionGet(); if (rp.hasFailures()) { - this.processFailBulkResponse(rp, /* TODO Objects.nonNull(mapping.getParent()) */false); + this.processFailBulkResponse(rp); } if (logger.isTraceEnabled()) { logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", 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 4cbea77a..0ba4d442 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 @@ -210,7 +210,7 @@ public class ESSyncService { // ------主表 查询sql来更新------ if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { ESMapping mapping = config.getEsMapping(); - String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName); boolean idFieldSimple = true; @@ -336,26 +336,50 @@ public class ESSyncService { // ------是主表------ if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { - FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping); - // 主键为简单字段 - if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) { - Object idVal = esTemplate.getValFromData(mapping, - data, - idFieldItem.getFieldName(), - idFieldItem.getColumn().getColumnName()); + if (mapping.get_id() != null) { + FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping); + // 主键为简单字段 + if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) { + Object idVal = esTemplate.getValFromData(mapping, + data, + idFieldItem.getFieldName(), + idFieldItem.getColumn().getColumnName()); - if (logger.isTraceEnabled()) { - logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, id: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index(), - idVal); + if (logger.isTraceEnabled()) { + logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, id: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index(), + idVal); + } + esTemplate.delete(mapping, idVal, null); + } else { + // ------主键带函数, 查询sql获取主键删除------ + // FIXME 删除时反查sql为空记录, 无法获获取 id field 值 + mainTableDelete(config, dml, data); } - esTemplate.delete(mapping, idVal); } else { - // ------主键带函数, 查询sql获取主键删除------ - mainTableDelete(config, dml, data); + FieldItem pkFieldItem = schemaItem.getIdFieldItem(mapping); + if (!pkFieldItem.isMethod() && !pkFieldItem.isBinaryOp()) { + Map esFieldData = new LinkedHashMap<>(); + Object pkVal = esTemplate.getESDataFromDmlData(mapping, data, esFieldData); + + if (logger.isTraceEnabled()) { + logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, pk: {}", + config.getDestination(), + dml.getTable(), + mapping.get_index(), + pkVal); + } + esFieldData.remove(pkFieldItem.getFieldName()); + esFieldData.keySet().forEach(key -> esFieldData.put(key, null)); + esTemplate.delete(mapping, pkVal, esFieldData); + } else { + // ------主键带函数, 查询sql获取主键删除------ + mainTableDelete(config, dml, data); + } } + } // 从表的操作 @@ -478,6 +502,15 @@ public class ESSyncService { } ESSyncUtil.sqlRS(ds, sql, rs -> { try { + Map esFieldData = null; + if (mapping.getPk() != null) { + esFieldData = new LinkedHashMap<>(); + esTemplate.getESDataFromDmlData(mapping, data, esFieldData); + esFieldData.remove(mapping.getPk()); + for (String key : esFieldData.keySet()) { + esFieldData.put(key, null); + } + } while (rs.next()) { Object idVal = esTemplate.getIdValFromRS(mapping, rs); @@ -489,7 +522,7 @@ public class ESSyncService { mapping.get_index(), idVal); } - esTemplate.delete(mapping, idVal); + esTemplate.delete(mapping, idVal, esFieldData); } } catch (Exception e) { throw new RuntimeException(e); @@ -535,13 +568,7 @@ public class ESSyncService { dml.getTable(), mapping.get_index()); } - boolean result = esTemplate.updateByQuery(config, paramsTmp, esFieldData); - if (!result) { - logger.error("Join table update es index by foreign key error, destination:{}, table: {}, index: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index()); - } + esTemplate.updateByQuery(config, paramsTmp, esFieldData); } /** @@ -629,14 +656,7 @@ public class ESSyncService { dml.getTable(), mapping.get_index()); } - boolean result = esTemplate.updateByQuery(config, paramsTmp, esFieldData); - if (!result) { - logger.error( - "Join table update es index by query sql error, destination:{}, table: {}, index: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index()); - } + esTemplate.updateByQuery(config, paramsTmp, esFieldData); } } catch (Exception e) { throw new RuntimeException(e); @@ -738,14 +758,7 @@ public class ESSyncService { dml.getTable(), mapping.get_index()); } - boolean result = esTemplate.updateByQuery(config, paramsTmp, esFieldData); - if (!result) { - logger.error( - "Join table update es index by query whole sql error, destination:{}, table: {}, index: {}", - config.getDestination(), - dml.getTable(), - mapping.get_index()); - } + esTemplate.updateByQuery(config, paramsTmp, esFieldData); } } catch (Exception e) { throw new RuntimeException(e); 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 5c389fae..3230a01e 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 @@ -9,15 +9,18 @@ import java.util.concurrent.ConcurrentMap; import javax.sql.DataSource; +import com.alibaba.fastjson.JSONObject; import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.cluster.metadata.MappingMetaData; import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; +import org.elasticsearch.search.SearchHit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,13 +39,13 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; */ public class ESTemplate { - private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class); + private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class); - private static final int MAX_BATCH_SIZE = 1000; + private static final int MAX_BATCH_SIZE = 1000; - private TransportClient transportClient; + private TransportClient transportClient; - private volatile BulkRequestBuilder bulkRequestBuilder; + private BulkRequestBuilder bulkRequestBuilder; public ESTemplate(TransportClient transportClient){ this.transportClient = transportClient; @@ -56,10 +59,9 @@ public class ESTemplate { /** * 插入数据 * - * @param mapping - * @param pkVal - * @param esFieldData - * @return + * @param mapping 配置对象 + * @param pkVal 主键值 + * @param esFieldData 数据Map */ public void insert(ESMapping mapping, Object pkVal, Map esFieldData) { if (mapping.get_id() != null) { @@ -71,50 +73,42 @@ public class ESTemplate { getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) .setSource(esFieldData)); } - commitBatch(); } else { - // TODO SearchResponse response = - // transportClient.prepareSearch(mapping.get_index()) - // .setTypes(mapping.get_type()) - // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - // .setSize(MAX_BATCH_SIZE) - // .get(); - // for (SearchHit hit : response.getHits()) { - // bulkRequestBuilder - // .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), - // hit.getId())); - // } - // bulkRequestBuilder - // .add(transportClient.prepareIndex(mapping.get_index(), - // mapping.get_type()).setSource(esFieldData)); + SearchResponse response = transportClient.prepareSearch(mapping.get_index()) + .setTypes(mapping.get_type()) + .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + .setSize(10000) + .get(); + for (SearchHit hit : response.getHits()) { + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) + .setDoc(esFieldData)); + } } - // return commitBulkRequest(bulkRequestBuilder); + commitBulk(); } /** * 根据主键更新数据 * - * @param mapping - * @param pkVal - * @param esFieldData - * @return + * @param mapping 配置对象 + * @param pkVal 主键值 + * @param esFieldData 数据Map */ public void update(ESMapping mapping, Object pkVal, Map esFieldData) { append4Update(mapping, pkVal, esFieldData); - commitBatch(); + commitBulk(); } /** * update by query * - * @param config - * @param paramsTmp - * @param esFieldData - * @return + * @param config 配置对象 + * @param paramsTmp sql查询条件 + * @param esFieldData 数据Map */ - public boolean updateByQuery(ESSyncConfig config, Map paramsTmp, Map esFieldData) { + public void updateByQuery(ESSyncConfig config, Map paramsTmp, Map esFieldData) { if (paramsTmp.isEmpty()) { - return false; + return; } ESMapping mapping = config.getEsMapping(); BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); @@ -133,7 +127,7 @@ public class ESTemplate { while (rs.next()) { Object idVal = getIdValFromRS(mapping, rs); append4Update(mapping, idVal, esFieldData); - commitBatch(); + commitBulk(); count++; } } catch (Exception e) { @@ -144,34 +138,30 @@ public class ESTemplate { if (logger.isTraceEnabled()) { logger.trace("Update ES by query effect {} records", syncCount); } - return true; } /** * 通过主键删除数据 * - * @param mapping - * @param pkVal - * @return + * @param mapping 配置对象 + * @param pkVal 主键值 + * @param esFieldData 数据Map */ - public void delete(ESMapping mapping, Object pkVal) { + public void delete(ESMapping mapping, Object pkVal, Map esFieldData) { if (mapping.get_id() != null) { getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); - commitBatch(); } else { - // TODO SearchResponse response = - // transportClient.prepareSearch(mapping.get_index()) - // .setTypes(mapping.get_type()) - // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - // .setSize(MAX_BATCH_SIZE) - // .get(); - // for (SearchHit hit : response.getHits()) { - // bulkRequestBuilder - // .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), - // hit.getId())); - // } + SearchResponse response = transportClient.prepareSearch(mapping.get_index()) + .setTypes(mapping.get_type()) + .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + .setSize(10000) + .get(); + for (SearchHit hit : response.getHits()) { + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) + .setDoc(esFieldData)); + } } - // return commitBulkRequest(bulkRequestBuilder); + commitBulk(); } /** @@ -199,9 +189,12 @@ public class ESTemplate { /** * 如果大于批量则提交批次 */ - private void commitBatch() { + private void commitBulk() { if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) { commit(); + + JSONObject jsonObject = new JSONObject(); + jsonObject.toJavaObject(Object.class); } } @@ -209,25 +202,22 @@ public class ESTemplate { if (mapping.get_id() != null) { if (mapping.isUpsert()) { getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData) - .setDocAsUpsert(true)); + .setDoc(esFieldData) + .setDocAsUpsert(true)); } else { getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData)); + .setDoc(esFieldData)); } } else { - // TODO SearchResponse response = - // transportClient.prepareSearch(mapping.get_index()) - // .setTypes(mapping.get_type()) - // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - // .setSize(MAX_BATCH_SIZE) - // .get(); - // for (SearchHit hit : response.getHits()) { - // bulkRequestBuilder - // .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), - // hit.getId()) - // .setDoc(esFieldData)); - // } + SearchResponse response = transportClient.prepareSearch(mapping.get_index()) + .setTypes(mapping.get_type()) + .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + .setSize(10000) + .get(); + for (SearchHit hit : response.getHits()) { + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) + .setDoc(esFieldData)); + } } } @@ -253,7 +243,7 @@ public class ESTemplate { public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map esFieldData) throws SQLException { SchemaItem schemaItem = mapping.getSchemaItem(); - String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); Object resultIdVal = null; for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); @@ -272,7 +262,7 @@ public class ESTemplate { public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException { SchemaItem schemaItem = mapping.getSchemaItem(); - String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); Object resultIdVal = null; for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); @@ -288,7 +278,7 @@ public class ESTemplate { public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map dmlOld, Map esFieldData) throws SQLException { SchemaItem schemaItem = mapping.getSchemaItem(); - String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + 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)) { @@ -335,7 +325,7 @@ public class ESTemplate { public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, Map esFieldData) { SchemaItem schemaItem = mapping.getSchemaItem(); - String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + 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(); @@ -364,7 +354,7 @@ public class ESTemplate { public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, Map dmlOld, Map esFieldData) { SchemaItem schemaItem = mapping.getSchemaItem(); - String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id(); + 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(); From 476244b26d84e3752801683f1e1e954afbe55498 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 14 Feb 2019 15:41:27 +0800 Subject: [PATCH 6/6] modify --- .../otter/canal/client/adapter/es/support/ESTemplate.java | 4 ---- 1 file changed, 4 deletions(-) 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 3230a01e..5974883d 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 @@ -9,7 +9,6 @@ import java.util.concurrent.ConcurrentMap; import javax.sql.DataSource; -import com.alibaba.fastjson.JSONObject; import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; @@ -192,9 +191,6 @@ public class ESTemplate { private void commitBulk() { if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) { commit(); - - JSONObject jsonObject = new JSONObject(); - jsonObject.toJavaObject(Object.class); } }