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..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,11 +144,14 @@ public class ESAdapter implements OuterAdapter { return; } for (Dml dml : dmls) { - sync(dml); + if (!dml.getIsDdl()) { + 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 +177,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 +191,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..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,7 +30,7 @@ public class ESSyncConfig { if (esMapping._type == null) { throw new NullPointerException("esMapping._type"); } - if (esMapping._id == null && esMapping.pk == null) { + if (esMapping._id == null && esMapping.getPk() == null) { throw new NullPointerException("esMapping._id and esMapping.pk"); } if (esMapping.sql == null) { @@ -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 5bb1eb41..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 @@ -1,422 +1,436 @@ -package com.alibaba.otter.canal.client.adapter.es.config; - -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; - -import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; - -/** - * ES 映射配置视图 - * - * @author rewerma 2018-11-01 - * @version 1.0.0 - */ -public class SchemaItem { - - private Map aliasTableItems = new LinkedHashMap<>(); // 别名对应表名 - private Map selectFields = new LinkedHashMap<>(); // 查询字段 - private String sql; - - private volatile Map> tableItemAliases; - private volatile Map> columnFields; - private volatile Boolean allFieldsSimple; - - public void init() { - this.getTableItemAliases(); - this.getColumnFields(); - this.isAllFieldsSimple(); - aliasTableItems.values().forEach(tableItem -> { - tableItem.getRelationTableFields(); - tableItem.getRelationSelectFieldItems(); - }); - } - - public Map getAliasTableItems() { - return aliasTableItems; - } - - public void setAliasTableItems(Map aliasTableItems) { - this.aliasTableItems = aliasTableItems; - } - - public String getSql() { - return sql; - } - - public void setSql(String sql) { - this.sql = sql; - } - - public Map getSelectFields() { - return selectFields; - } - - public void setSelectFields(Map selectFields) { - this.selectFields = selectFields; - } - - public Map> getTableItemAliases() { - if (tableItemAliases == null) { - synchronized (SchemaItem.class) { - if (tableItemAliases == null) { - tableItemAliases = new LinkedHashMap<>(); - aliasTableItems.forEach((alias, tableItem) -> { - List aliases = tableItemAliases - .computeIfAbsent(tableItem.getTableName().toLowerCase(), k -> new ArrayList<>()); - aliases.add(tableItem); - }); - } - } - } - return tableItemAliases; - } - - public Map> getColumnFields() { - if (columnFields == null) { - synchronized (SchemaItem.class) { - if (columnFields == null) { - columnFields = new LinkedHashMap<>(); - getSelectFields() - .forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> { - // TableItem tableItem = getAliasTableItems().get(columnItem.getOwner()); - // if (!tableItem.isSubQuery()) { - List fieldItems = columnFields.computeIfAbsent( - columnItem.getOwner() + "." + columnItem.getColumnName(), - k -> new ArrayList<>()); - fieldItems.add(fieldItem); - // } else { - // tableItem.getSubQueryFields().forEach(subQueryField -> { - // List fieldItems = columnFields.computeIfAbsent( - // columnItem.getOwner() + "." + subQueryField.getColumn().getColumnName(), - // k -> new ArrayList<>()); - // fieldItems.add(fieldItem); - // }); - // } - })); - } - } - } - return columnFields; - } - - public boolean isAllFieldsSimple() { - if (allFieldsSimple == null) { - synchronized (SchemaItem.class) { - if (allFieldsSimple == null) { - allFieldsSimple = true; - - for (FieldItem fieldItem : getSelectFields().values()) { - if (fieldItem.isMethod() || fieldItem.isBinaryOp()) { - allFieldsSimple = false; - break; - } - } - } - } - } - - return allFieldsSimple; - } - - public TableItem getMainTable() { - if (!aliasTableItems.isEmpty()) { - return aliasTableItems.values().iterator().next(); - } else { - return null; - } - } - - public FieldItem getIdFieldItem(ESMapping mapping) { - if (mapping.get_id() != null) { - return getSelectFields().get(mapping.get_id()); - } else { - return getSelectFields().get(mapping.getPk()); - } - } - - public static class TableItem { - - private SchemaItem schemaItem; - - private String schema; - private String tableName; - private String alias; - private String subQuerySql; - private List subQueryFields = new ArrayList<>(); - private List relationFields = new ArrayList<>(); - - private boolean main; - private boolean subQuery; - - private volatile Map> relationTableFields; // 当前表关联条件字段对应主表查询字段 - private volatile List relationSelectFieldItems; // 子表所在主表的查询字段 - - public TableItem(SchemaItem schemaItem){ - this.schemaItem = schemaItem; - } - - public SchemaItem getSchemaItem() { - return schemaItem; - } - - public void setSchemaItem(SchemaItem schemaItem) { - this.schemaItem = schemaItem; - } - - public String getSchema() { - return schema; - } - - public void setSchema(String schema) { - this.schema = schema; - } - - public String getTableName() { - return tableName; - } - - public void setTableName(String tableName) { - this.tableName = tableName; - } - - public String getAlias() { - return alias; - } - - public void setAlias(String alias) { - this.alias = alias; - } - - public String getSubQuerySql() { - return subQuerySql; - } - - public void setSubQuerySql(String subQuerySql) { - this.subQuerySql = subQuerySql; - } - - public boolean isMain() { - return main; - } - - public void setMain(boolean main) { - this.main = main; - } - - public boolean isSubQuery() { - return subQuery; - } - - public void setSubQuery(boolean subQuery) { - this.subQuery = subQuery; - } - - public List getSubQueryFields() { - return subQueryFields; - } - - public void setSubQueryFields(List subQueryFields) { - this.subQueryFields = subQueryFields; - } - - public List getRelationFields() { - return relationFields; - } - - public void setRelationFields(List relationFields) { - this.relationFields = relationFields; - } - - public Map> getRelationTableFields() { - if (relationTableFields == null) { - synchronized (SchemaItem.class) { - if (relationTableFields == null) { - relationTableFields = new LinkedHashMap<>(); - - getRelationFields().forEach(relationFieldsPair -> { - FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); - FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); - FieldItem currentTableRelField = null; - if (getAlias().equals(leftFieldItem.getOwner())) { - currentTableRelField = leftFieldItem; - } else if (getAlias().equals(rightFieldItem.getOwner())) { - currentTableRelField = rightFieldItem; - } - - if (currentTableRelField != null) { - List selectFieldItem = getSchemaItem().getColumnFields() - .get(leftFieldItem.getOwner() + "." + leftFieldItem.getColumn().getColumnName()); - if (selectFieldItem != null && !selectFieldItem.isEmpty()) { - relationTableFields.put(currentTableRelField, selectFieldItem); - } else { - selectFieldItem = getSchemaItem().getColumnFields() - .get(rightFieldItem.getOwner() + "." - + rightFieldItem.getColumn().getColumnName()); - if (selectFieldItem != null && !selectFieldItem.isEmpty()) { - relationTableFields.put(currentTableRelField, selectFieldItem); - } else { - throw new UnsupportedOperationException( - "Relation condition column must in select columns."); - } - } - } - }); - } - } - } - return relationTableFields; - } - - public List getRelationSelectFieldItems() { - if (relationSelectFieldItems == null) { - synchronized (SchemaItem.class) { - if (relationSelectFieldItems == null) { - relationSelectFieldItems = new ArrayList<>(); - for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { - if (fieldItem.getOwners().contains(getAlias())) { - relationSelectFieldItems.add(fieldItem); - } - } - } - } - } - return relationSelectFieldItems; - } - } - - public static class RelationFieldsPair { - - private FieldItem leftFieldItem; - private FieldItem rightFieldItem; - - public RelationFieldsPair(FieldItem leftFieldItem, FieldItem rightFieldItem){ - this.leftFieldItem = leftFieldItem; - this.rightFieldItem = rightFieldItem; - } - - public FieldItem getLeftFieldItem() { - return leftFieldItem; - } - - public void setLeftFieldItem(FieldItem leftFieldItem) { - this.leftFieldItem = leftFieldItem; - } - - public FieldItem getRightFieldItem() { - return rightFieldItem; - } - - public void setRightFieldItem(FieldItem rightFieldItem) { - this.rightFieldItem = rightFieldItem; - } - } - - public static class FieldItem { - - private String fieldName; - private List columnItems = new ArrayList<>(); - private List owners = new ArrayList<>(); - - private boolean method; - private boolean binaryOp; - - public String getFieldName() { - return fieldName; - } - - public void setFieldName(String fieldName) { - this.fieldName = fieldName; - } - - public List getColumnItems() { - return columnItems; - } - - public void setColumnItems(List columnItems) { - this.columnItems = columnItems; - } - - public boolean isMethod() { - return method; - } - - public void setMethod(boolean method) { - this.method = method; - } - - public boolean isBinaryOp() { - return binaryOp; - } - - public void setBinaryOp(boolean binaryOp) { - this.binaryOp = binaryOp; - } - - public List getOwners() { - return owners; - } - - public void setOwners(List owners) { - this.owners = owners; - } - - public void addColumn(ColumnItem columnItem) { - columnItems.add(columnItem); - } - - public ColumnItem getColumn() { - if (!columnItems.isEmpty()) { - return columnItems.get(0); - } else { - return null; - } - } - - public String getOwner() { - if (!owners.isEmpty()) { - return owners.get(0); - } else { - return null; - } - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - FieldItem fieldItem = (FieldItem) o; - - return fieldName != null ? fieldName.equals(fieldItem.fieldName) : fieldItem.fieldName == null; - } - - @Override - public int hashCode() { - return fieldName != null ? fieldName.hashCode() : 0; - } - } - - public static class ColumnItem { - - private String owner; - private String columnName; - - public String getOwner() { - return owner; - } - - public void setOwner(String owner) { - this.owner = owner; - } - - public String getColumnName() { - return columnName; - } - - public void setColumnName(String columnName) { - this.columnName = columnName; - } - } -} +package com.alibaba.otter.canal.client.adapter.es.config; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; + +/** + * ES 映射配置视图 + * + * @author rewerma 2018-11-01 + * @version 1.0.0 + */ +public class SchemaItem { + + private Map aliasTableItems = new LinkedHashMap<>(); // 别名对应表名 + private Map selectFields = new LinkedHashMap<>(); // 查询字段 + private String sql; + + private volatile Map> tableItemAliases; + private volatile Map> columnFields; + private volatile Boolean allFieldsSimple; + + public void init() { + this.getTableItemAliases(); + this.getColumnFields(); + this.isAllFieldsSimple(); + aliasTableItems.values().forEach(tableItem -> { + tableItem.getRelationTableFields(); + tableItem.getRelationSelectFieldItems(); + }); + } + + public Map getAliasTableItems() { + return aliasTableItems; + } + + public void setAliasTableItems(Map aliasTableItems) { + this.aliasTableItems = aliasTableItems; + } + + public String getSql() { + return sql; + } + + public void setSql(String sql) { + this.sql = sql; + } + + public Map getSelectFields() { + return selectFields; + } + + public void setSelectFields(Map selectFields) { + this.selectFields = selectFields; + } + + public String toSql() { + // todo + return null; + } + + public Map> getTableItemAliases() { + if (tableItemAliases == null) { + synchronized (SchemaItem.class) { + if (tableItemAliases == null) { + tableItemAliases = new LinkedHashMap<>(); + aliasTableItems.forEach((alias, tableItem) -> { + List aliases = tableItemAliases + .computeIfAbsent(tableItem.getTableName().toLowerCase(), k -> new ArrayList<>()); + aliases.add(tableItem); + }); + } + } + } + return tableItemAliases; + } + + public Map> getColumnFields() { + if (columnFields == null) { + synchronized (SchemaItem.class) { + if (columnFields == null) { + columnFields = new LinkedHashMap<>(); + getSelectFields() + .forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> { + // TableItem tableItem = getAliasTableItems().get(columnItem.getOwner()); + // if (!tableItem.isSubQuery()) { + List fieldItems = columnFields.computeIfAbsent( + columnItem.getOwner() + "." + columnItem.getColumnName(), + k -> new ArrayList<>()); + fieldItems.add(fieldItem); + // } else { + // tableItem.getSubQueryFields().forEach(subQueryField -> { + // List fieldItems = columnFields.computeIfAbsent( + // columnItem.getOwner() + "." + subQueryField.getColumn().getColumnName(), + // k -> new ArrayList<>()); + // fieldItems.add(fieldItem); + // }); + // } + })); + } + } + } + return columnFields; + } + + public boolean isAllFieldsSimple() { + if (allFieldsSimple == null) { + synchronized (SchemaItem.class) { + if (allFieldsSimple == null) { + allFieldsSimple = true; + + for (FieldItem fieldItem : getSelectFields().values()) { + if (fieldItem.isMethod() || fieldItem.isBinaryOp()) { + allFieldsSimple = false; + break; + } + } + } + } + } + + return allFieldsSimple; + } + + public TableItem getMainTable() { + if (!aliasTableItems.isEmpty()) { + return aliasTableItems.values().iterator().next(); + } else { + return null; + } + } + + public FieldItem getIdFieldItem(ESMapping mapping) { + if (mapping.get_id() != null) { + return getSelectFields().get(mapping.get_id()); + } else { + return getSelectFields().get(mapping.getPk()); + } + } + + public static class TableItem { + + private SchemaItem schemaItem; + + private String schema; + private String tableName; + private String alias; + private String subQuerySql; + private List subQueryFields = new ArrayList<>(); + private List relationFields = new ArrayList<>(); + + private boolean main; + private boolean subQuery; + + private volatile Map> relationTableFields; // 当前表关联条件字段对应主表查询字段 + private volatile List relationSelectFieldItems; // 子表所在主表的查询字段 + + public TableItem(SchemaItem schemaItem){ + this.schemaItem = schemaItem; + } + + public SchemaItem getSchemaItem() { + return schemaItem; + } + + public void setSchemaItem(SchemaItem schemaItem) { + this.schemaItem = schemaItem; + } + + public String getSchema() { + return schema; + } + + public void setSchema(String schema) { + this.schema = schema; + } + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public String getAlias() { + return alias; + } + + public void setAlias(String alias) { + this.alias = alias; + } + + public String getSubQuerySql() { + return subQuerySql; + } + + public void setSubQuerySql(String subQuerySql) { + this.subQuerySql = subQuerySql; + } + + public boolean isMain() { + return main; + } + + public void setMain(boolean main) { + this.main = main; + } + + public boolean isSubQuery() { + return subQuery; + } + + public void setSubQuery(boolean subQuery) { + this.subQuery = subQuery; + } + + public List getSubQueryFields() { + return subQueryFields; + } + + public void setSubQueryFields(List subQueryFields) { + this.subQueryFields = subQueryFields; + } + + public List getRelationFields() { + return relationFields; + } + + public void setRelationFields(List relationFields) { + this.relationFields = relationFields; + } + + public Map> getRelationTableFields() { + if (relationTableFields == null) { + synchronized (SchemaItem.class) { + if (relationTableFields == null) { + relationTableFields = new LinkedHashMap<>(); + + getRelationFields().forEach(relationFieldsPair -> { + FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); + FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); + FieldItem currentTableRelField = null; + if (getAlias().equals(leftFieldItem.getOwner())) { + currentTableRelField = leftFieldItem; + } else if (getAlias().equals(rightFieldItem.getOwner())) { + currentTableRelField = rightFieldItem; + } + + if (currentTableRelField != null) { + List selectFieldItem = getSchemaItem().getColumnFields() + .get(leftFieldItem.getOwner() + "." + leftFieldItem.getColumn().getColumnName()); + if (selectFieldItem != null && !selectFieldItem.isEmpty()) { + relationTableFields.put(currentTableRelField, selectFieldItem); + } else { + selectFieldItem = getSchemaItem().getColumnFields() + .get(rightFieldItem.getOwner() + "." + + rightFieldItem.getColumn().getColumnName()); + if (selectFieldItem != null && !selectFieldItem.isEmpty()) { + relationTableFields.put(currentTableRelField, selectFieldItem); + } else { + throw new UnsupportedOperationException( + "Relation condition column must in select columns."); + } + } + } + }); + } + } + } + return relationTableFields; + } + + public List getRelationSelectFieldItems() { + if (relationSelectFieldItems == null) { + synchronized (SchemaItem.class) { + if (relationSelectFieldItems == null) { + relationSelectFieldItems = new ArrayList<>(); + for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { + if (fieldItem.getOwners().contains(getAlias())) { + relationSelectFieldItems.add(fieldItem); + } + } + } + } + } + return relationSelectFieldItems; + } + } + + public static class RelationFieldsPair { + + private FieldItem leftFieldItem; + private FieldItem rightFieldItem; + + public RelationFieldsPair(FieldItem leftFieldItem, FieldItem rightFieldItem){ + this.leftFieldItem = leftFieldItem; + this.rightFieldItem = rightFieldItem; + } + + public FieldItem getLeftFieldItem() { + return leftFieldItem; + } + + public void setLeftFieldItem(FieldItem leftFieldItem) { + this.leftFieldItem = leftFieldItem; + } + + public FieldItem getRightFieldItem() { + return rightFieldItem; + } + + public void setRightFieldItem(FieldItem rightFieldItem) { + this.rightFieldItem = rightFieldItem; + } + } + + public static class FieldItem { + + private String fieldName; + private String expr; + private List columnItems = new ArrayList<>(); + private List owners = new ArrayList<>(); + + private boolean method; + private boolean binaryOp; + + public String getFieldName() { + return fieldName; + } + + public void setFieldName(String fieldName) { + this.fieldName = fieldName; + } + + public String getExpr() { + return expr; + } + + public void setExpr(String expr) { + this.expr = expr; + } + + public List getColumnItems() { + return columnItems; + } + + public void setColumnItems(List columnItems) { + this.columnItems = columnItems; + } + + public boolean isMethod() { + return method; + } + + public void setMethod(boolean method) { + this.method = method; + } + + public boolean isBinaryOp() { + return binaryOp; + } + + public void setBinaryOp(boolean binaryOp) { + this.binaryOp = binaryOp; + } + + public List getOwners() { + return owners; + } + + public void setOwners(List owners) { + this.owners = owners; + } + + public void addColumn(ColumnItem columnItem) { + columnItems.add(columnItem); + } + + public ColumnItem getColumn() { + if (!columnItems.isEmpty()) { + return columnItems.get(0); + } else { + return null; + } + } + + public String getOwner() { + if (!owners.isEmpty()) { + return owners.get(0); + } else { + return null; + } + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + FieldItem fieldItem = (FieldItem) o; + + return fieldName != null ? fieldName.equals(fieldItem.fieldName) : fieldItem.fieldName == null; + } + + @Override + public int hashCode() { + return fieldName != null ? fieldName.hashCode() : 0; + } + } + + public static class ColumnItem { + + private String owner; + private String columnName; + + public String getOwner() { + return owner; + } + + public void setOwner(String owner) { + this.owner = owner; + } + + public String getColumnName() { + return columnName; + } + + public void setColumnName(String columnName) { + this.columnName = columnName; + } + } +} 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/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..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 @@ -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; @@ -14,7 +17,9 @@ 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; @@ -51,7 +56,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 +98,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); } @@ -158,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; @@ -205,31 +205,27 @@ 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 + 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)); } } @@ -238,11 +234,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); } - if (logger.isDebugEnabled()) { - logger.debug("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", + if (logger.isTraceEnabled()) { + logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", (System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - esBatchBegin), bulkRequestBuilder.numberOfActions(), @@ -259,10 +255,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); } - 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..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 @@ -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()) { @@ -329,33 +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); - } - 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); + 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); } } 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); + } } + } // 从表的操作 @@ -417,14 +441,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 +478,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); @@ -493,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); @@ -504,15 +522,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, esFieldData); } } catch (Exception e) { throw new RuntimeException(e); @@ -558,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); } /** @@ -652,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); @@ -761,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); @@ -799,14 +789,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 +826,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 +834,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 e6442fa1..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 @@ -2,14 +2,13 @@ package com.alibaba.otter.canal.client.adapter.es.support; import java.sql.ResultSet; import java.sql.SQLException; -import java.util.*; +import java.util.LinkedHashMap; +import java.util.Map; 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; @@ -18,18 +17,11 @@ 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; @@ -52,254 +44,131 @@ public class ESTemplate { private TransportClient transportClient; + private 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 + * @param mapping 配置对象 + * @param pkVal 主键值 + * @param esFieldData 数据Map */ - public boolean insert(ESMapping mapping, Object pkVal, Map esFieldData) { - BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); + public void insert(ESMapping mapping, Object pkVal, Map esFieldData) { if (mapping.get_id() != null) { - bulkRequestBuilder - .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) + 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)); + } } else { SearchResponse response = transportClient.prepareSearch(mapping.get_index()) .setTypes(mapping.get_type()) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .setSize(MAX_BATCH_SIZE) + .setSize(10000) .get(); for (SearchHit hit : response.getHits()) { - bulkRequestBuilder - .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) + .setDoc(esFieldData)); } - bulkRequestBuilder - .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()).setSource(esFieldData)); } - return commitBulkRequest(bulkRequestBuilder); + commitBulk(); } /** * 根据主键更新数据 * - * @param mapping - * @param pkVal - * @param esFieldData - * @return + * @param mapping 配置对象 + * @param pkVal 主键值 + * @param esFieldData 数据Map */ - 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)); - } - } + public void update(ESMapping mapping, Object pkVal, Map esFieldData) { + append4Update(mapping, pkVal, esFieldData); + 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(); 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存在分片版本冲突，准备重新执行..."); + // 查询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 { - TimeUnit.SECONDS.sleep(1); - } catch (InterruptedException e) { - // ignore + while (rs.next()) { + Object idVal = getIdValFromRS(mapping, rs); + append4Update(mapping, idVal, esFieldData); + commitBulk(); + count++; + } + } catch (Exception e) { + throw new RuntimeException(e); } - return updateByQuery(mapping, queryBuilder, esFieldData, ++counter); + return count; + }); + 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 boolean delete(ESMapping mapping, Object pkVal) { - BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); + public void delete(ESMapping mapping, Object pkVal, Map esFieldData) { if (mapping.get_id() != null) { - bulkRequestBuilder - .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); + getBulk().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) + .setSize(10000) .get(); for (SearchHit hit : response.getHits()) { - bulkRequestBuilder - .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); + getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) + .setDoc(esFieldData)); } } - return commitBulkRequest(bulkRequestBuilder); + commitBulk(); } /** - * 批量提交 - * - * @param bulkRequestBuilder - * @return + * 提交批次 */ - private static boolean commitBulkRequest(BulkRequestBuilder bulkRequestBuilder) { - if (bulkRequestBuilder.numberOfActions() > 0) { - BulkResponse response = bulkRequestBuilder.execute().actionGet(); + public void commit() { + if (getBulk().numberOfActions() > 0) { + BulkResponse response = getBulk().execute().actionGet(); if (response.hasFailures()) { for (BulkItemResponse itemResponse : response.getItems()) { if (!itemResponse.isFailed()) { @@ -307,16 +176,45 @@ public class ESTemplate { } if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) { - logger.warn(itemResponse.getFailureMessage()); + logger.error(itemResponse.getFailureMessage()); } else { - logger.error("ES sync commit error: {}", itemResponse.getFailureMessage()); + throw new RuntimeException("ES sync commit error" + itemResponse.getFailureMessage()); } } } - - return !response.hasFailures(); } - return true; + } + + /** + * 如果大于批量则提交批次 + */ + private void commitBulk() { + if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) { + commit(); + } + } + + 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 { + 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)); + } + } } public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String 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 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()));