diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml
deleted file mode 100644
index ddc96604..00000000
--- a/client-adapter/elasticsearch/pom.xml
+++ /dev/null
@@ -1,105 +0,0 @@
-
-
-
- canal.client-adapter
- com.alibaba.otter
- 1.1.5-SNAPSHOT
-
- 4.0.0
- com.alibaba.otter
- client-adapter.elasticsearch
- jar
- canal client adapter elasticsearch module for otter ${project.version}
-
-
-
- com.alibaba.otter
- client-adapter.common
- ${project.version}
- provided
-
-
- com.alibaba.fastsql
- fastsql
- 2.0.0_preview_855
-
-
- org.elasticsearch
- elasticsearch
- 6.4.3
-
-
- org.elasticsearch.client
- transport
- 6.4.3
-
-
- org.elasticsearch.client
- elasticsearch-rest-client
- 6.4.3
-
-
- org.elasticsearch.client
- elasticsearch-rest-high-level-client
- 6.4.3
-
-
-
- junit
- junit
- 4.12
- test
-
-
- mysql
- mysql-connector-java
- 5.1.40
- test
-
-
-
-
-
-
- org.apache.maven.plugins
- maven-assembly-plugin
- 2.4
-
-
- jar-with-dependencies
-
-
-
-
- make-assembly
- package
-
- single
-
-
-
-
-
- maven-antrun-plugin
-
-
- package
-
- run
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
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
deleted file mode 100644
index 90e13ef8..00000000
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
+++ /dev/null
@@ -1,256 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.es;
-
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import javax.sql.DataSource;
-
-import org.apache.commons.lang.StringUtils;
-import org.elasticsearch.action.search.SearchResponse;
-
-import com.alibaba.druid.pool.DruidDataSource;
-import com.alibaba.otter.canal.client.adapter.OuterAdapter;
-import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
-import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
-import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader;
-import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
-import com.alibaba.otter.canal.client.adapter.es.config.SqlParser;
-import com.alibaba.otter.canal.client.adapter.es.monitor.ESConfigMonitor;
-import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService;
-import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService;
-import com.alibaba.otter.canal.client.adapter.es.support.ESConnection;
-import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
-import com.alibaba.otter.canal.client.adapter.support.*;
-
-/**
- * ES外部适配器
- *
- * @author rewerma 2018-10-20
- * @version 1.0.0
- */
-@SPI("es")
-public class ESAdapter implements OuterAdapter {
-
- private Map esSyncConfig = new ConcurrentHashMap<>(); // 文件名对应配置
- private Map> dbTableEsSyncConfig = new ConcurrentHashMap<>(); // schema-table对应配置
-
- private ESConnection esConnection;
-
- private ESSyncService esSyncService;
-
- private ESConfigMonitor esConfigMonitor;
-
- private Properties envProperties;
-
- public ESSyncService getEsSyncService() {
- return esSyncService;
- }
-
- public Map getEsSyncConfig() {
- return esSyncConfig;
- }
-
- public Map> getDbTableEsSyncConfig() {
- return dbTableEsSyncConfig;
- }
-
- public ESConnection getEsConnection() {
- return esConnection;
- }
-
- public void setEsConnection(ESConnection esConnection) {
- this.esConnection = esConnection;
- }
-
- @Override
- public void init(OuterAdapterConfig configuration, Properties envProperties) {
- try {
- this.envProperties = envProperties;
- Map esSyncConfigTmp = ESSyncConfigLoader.load(envProperties);
- // 过滤不匹配的key的配置
- esSyncConfigTmp.forEach((key, config) -> {
- if ((config.getOuterAdapterKey() == null && configuration.getKey() == null)
- || (config.getOuterAdapterKey() != null
- && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
- esSyncConfig.put(key, config);
- }
- });
-
- for (Map.Entry entry : esSyncConfig.entrySet()) {
- String configName = entry.getKey();
- ESSyncConfig config = entry.getValue();
- addSyncConfigToCache(configName, config);
- }
-
- Map properties = configuration.getProperties();
-
- String[] hostArray = configuration.getHosts().split(",");
- String mode = properties.get("mode");
- if ("rest".equalsIgnoreCase(mode) || "http".equalsIgnoreCase(mode)) {
- esConnection = new ESConnection(hostArray, properties, ESConnection.ESClientMode.REST);
- } else {
- esConnection = new ESConnection(hostArray, properties, ESConnection.ESClientMode.TRANSPORT);
- }
-
- ESTemplate esTemplate = new ESTemplate(esConnection);
- esSyncService = new ESSyncService(esTemplate);
-
- esConfigMonitor = new ESConfigMonitor();
- esConfigMonitor.init(this, envProperties);
- } catch (Throwable e) {
- throw new RuntimeException(e);
- }
- }
-
- public void addSyncConfigToCache(String configName, ESSyncConfig config) {
- Properties envProperties = this.envProperties;
- SchemaItem schemaItem = SqlParser.parse(config.getEsMapping().getSql());
- config.getEsMapping().setSchemaItem(schemaItem);
-
- DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
- if (dataSource == null || dataSource.getUrl() == null) {
- throw new RuntimeException("No data source found: " + config.getDataSourceKey());
- }
- Pattern pattern = Pattern.compile(".*:(.*)://.*/(.*)\\?.*$");
- Matcher matcher = pattern.matcher(dataSource.getUrl());
- if (!matcher.find()) {
- throw new RuntimeException("Not found the schema of jdbc-url: " + config.getDataSourceKey());
- }
- String schema = matcher.group(2);
-
- schemaItem.getAliasTableItems().values().forEach(tableItem -> {
- Map esSyncConfigMap;
- if (envProperties != null
- && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
- esSyncConfigMap = dbTableEsSyncConfig
- .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "-"
- + StringUtils.trimToEmpty(config.getGroupId()) + "_" + schema + "-"
- + tableItem.getTableName(),
- k -> new ConcurrentHashMap<>());
- } else {
- esSyncConfigMap = dbTableEsSyncConfig
- .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "_" + schema + "-"
- + tableItem.getTableName(),
- k -> new ConcurrentHashMap<>());
- }
-
- esSyncConfigMap.put(configName, config);
- });
- }
-
-
- @Override
- public void sync(List dmls) {
- if (dmls == null || dmls.isEmpty()) {
- return;
- }
- for (Dml dml : dmls) {
- if (!dml.getIsDdl()) {
- sync(dml);
- }
- }
- esSyncService.commit(); // 批次统一提交
-
- }
-
- private void sync(Dml dml) {
- String database = dml.getDatabase();
- String table = dml.getTable();
- Map configMap;
- if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
- configMap = dbTableEsSyncConfig
- .get(StringUtils.trimToEmpty(dml.getDestination()) + "-" + StringUtils.trimToEmpty(dml.getGroupId())
- + "_" + database + "-" + table);
- } else {
- configMap = dbTableEsSyncConfig
- .get(StringUtils.trimToEmpty(dml.getDestination()) + "_" + database + "-" + table);
- }
-
- if (configMap != null && !configMap.values().isEmpty()) {
- esSyncService.sync(configMap.values(), dml);
- }
- }
-
- @Override
- public EtlResult etl(String task, List params) {
- EtlResult etlResult = new EtlResult();
- ESSyncConfig config = esSyncConfig.get(task);
- if (config != null) {
- DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
- ESEtlService esEtlService = new ESEtlService(esConnection, config);
- if (dataSource != null) {
- return esEtlService.importData(params);
- } else {
- etlResult.setSucceeded(false);
- etlResult.setErrorMessage("DataSource not found");
- return etlResult;
- }
- } else {
- StringBuilder resultMsg = new StringBuilder();
- boolean resSuccess = true;
- for (ESSyncConfig configTmp : esSyncConfig.values()) {
- // 取所有的destination为task的配置
- if (configTmp.getDestination().equals(task)) {
- ESEtlService esEtlService = new ESEtlService(esConnection, configTmp);
- EtlResult etlRes = esEtlService.importData(params);
- if (!etlRes.getSucceeded()) {
- resSuccess = false;
- resultMsg.append(etlRes.getErrorMessage()).append("\n");
- } else {
- resultMsg.append(etlRes.getResultMessage()).append("\n");
- }
- }
- }
- if (resultMsg.length() > 0) {
- etlResult.setSucceeded(resSuccess);
- if (resSuccess) {
- etlResult.setResultMessage(resultMsg.toString());
- } else {
- etlResult.setErrorMessage(resultMsg.toString());
- }
- return etlResult;
- }
- }
- etlResult.setSucceeded(false);
- etlResult.setErrorMessage("Task not found");
- return etlResult;
- }
-
- @Override
- public Map count(String task) {
- ESSyncConfig config = esSyncConfig.get(task);
- ESMapping mapping = config.getEsMapping();
- SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index(), mapping.get_type()).size(0)
- .getResponse();
-
- long rowCount = response.getHits().getTotalHits();
- Map res = new LinkedHashMap<>();
- res.put("esIndex", mapping.get_index());
- res.put("count", rowCount);
- return res;
- }
-
- @Override
- public void destroy() {
- if (esConfigMonitor != null) {
- esConfigMonitor.destroy();
- }
- if (esConnection != null) {
- esConnection.close();
- }
- }
-
- @Override
- public String getDestination(String task) {
- ESSyncConfig config = esSyncConfig.get(task);
- if (config != null) {
- return config.getDestination();
- }
- return null;
- }
-}
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
deleted file mode 100644
index 9cb2ed30..00000000
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
+++ /dev/null
@@ -1,241 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.es.config;
-
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
-
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * ES 映射配置
- *
- * @author rewerma 2018-11-01
- * @version 1.0.0
- */
-public class ESSyncConfig implements AdapterConfig {
-
- private String dataSourceKey; // 数据源key
-
- private String outerAdapterKey; // adapter key
-
- private String groupId; // group id
-
- private String destination; // canal destination
-
- private ESMapping esMapping;
-
- public void validate() {
- if (esMapping._index == null) {
- throw new NullPointerException("esMapping._index");
- }
- if (esMapping._type == null) {
- throw new NullPointerException("esMapping._type");
- }
- if (esMapping._id == null && esMapping.getPk() == null) {
- throw new NullPointerException("esMapping._id or esMapping.pk");
- }
- if (esMapping.sql == null) {
- throw new NullPointerException("esMapping.sql");
- }
- }
-
- public String getDataSourceKey() {
- return dataSourceKey;
- }
-
- public void setDataSourceKey(String dataSourceKey) {
- this.dataSourceKey = dataSourceKey;
- }
-
- public String getOuterAdapterKey() {
- return outerAdapterKey;
- }
-
- public void setOuterAdapterKey(String outerAdapterKey) {
- this.outerAdapterKey = outerAdapterKey;
- }
-
- public String getGroupId() {
- return groupId;
- }
-
- public void setGroupId(String groupId) {
- this.groupId = groupId;
- }
-
- public String getDestination() {
- return destination;
- }
-
- public void setDestination(String destination) {
- this.destination = destination;
- }
-
- public ESMapping getEsMapping() {
- return esMapping;
- }
-
- public void setEsMapping(ESMapping esMapping) {
- this.esMapping = esMapping;
- }
-
- public ESMapping getMapping(){
- return esMapping;
- }
-
- public static class ESMapping implements AdapterMapping{
-
- private String _index;
- private String _type;
- private String _id;
- private boolean upsert = false;
- private String pk;
- private Map relations = new LinkedHashMap<>();
- private String sql;
- // 对象字段, 例: objFields:
- // - _labels: array:;
- private Map objFields = new LinkedHashMap<>();
- private List skips = new ArrayList<>();
- private int commitBatch = 1000;
- private String etlCondition;
- private boolean syncByTimestamp = false; // 是否按时间戳定时同步
- private Long syncInterval; // 同步时间间隔
-
- private SchemaItem schemaItem; // sql解析结果模型
-
- public String get_index() {
- return _index;
- }
-
- public void set_index(String _index) {
- this._index = _index;
- }
-
- public String get_type() {
- return _type;
- }
-
- public void set_type(String _type) {
- this._type = _type;
- }
-
- public String get_id() {
- return _id;
- }
-
- public void set_id(String _id) {
- this._id = _id;
- }
-
- public boolean isUpsert() {
- return upsert;
- }
-
- public void setUpsert(boolean upsert) {
- this.upsert = upsert;
- }
-
- public String getPk() {
- return pk;
- }
-
- public void setPk(String pk) {
- this.pk = pk;
- }
-
- public Map getObjFields() {
- return objFields;
- }
-
- public void setObjFields(Map objFields) {
- this.objFields = objFields;
- }
-
- public List getSkips() {
- return skips;
- }
-
- public void setSkips(List skips) {
- this.skips = skips;
- }
-
- public Map getRelations() {
- return relations;
- }
-
- public void setRelations(Map relations) {
- this.relations = relations;
- }
-
- public String getSql() {
- return sql;
- }
-
- public void setSql(String sql) {
- this.sql = sql;
- }
-
- public int getCommitBatch() {
- return commitBatch;
- }
-
- public void setCommitBatch(int commitBatch) {
- this.commitBatch = commitBatch;
- }
-
- public String getEtlCondition() {
- return etlCondition;
- }
-
- public void setEtlCondition(String etlCondition) {
- this.etlCondition = etlCondition;
- }
-
- public Long getSyncInterval() {
- return syncInterval;
- }
-
- public void setSyncInterval(Long syncInterval) {
- this.syncInterval = syncInterval;
- }
-
- public boolean isSyncByTimestamp() {
- return syncByTimestamp;
- }
-
- public void setSyncByTimestamp(boolean syncByTimestamp) {
- this.syncByTimestamp = syncByTimestamp;
- }
-
- public SchemaItem getSchemaItem() {
- return schemaItem;
- }
-
- public void setSchemaItem(SchemaItem schemaItem) {
- this.schemaItem = schemaItem;
- }
- }
-
- public static class RelationMapping {
-
- private String name;
- private String parent;
-
- public String getName() {
- return name;
- }
-
- public void setName(String name) {
- this.name = name;
- }
-
- public String getParent() {
- return parent;
- }
-
- public void setParent(String parent) {
- this.parent = parent;
- }
- }
-}
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java
deleted file mode 100644
index 51216665..00000000
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java
+++ /dev/null
@@ -1,45 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.es.config;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Properties;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
-import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
-
-/**
- * ES 配置装载器
- *
- * @author rewerma 2018-11-01
- * @version 1.0.0
- */
-public class ESSyncConfigLoader {
-
- private static Logger logger = LoggerFactory.getLogger(ESSyncConfigLoader.class);
-
- public static synchronized Map load(Properties envProperties) {
- logger.info("## Start loading es mapping config ... ");
-
- Map esSyncConfig = new LinkedHashMap<>();
-
- Map configContentMap = MappingConfigsLoader.loadConfigs("es");
- configContentMap.forEach((fileName, content) -> {
- ESSyncConfig config = YmlConfigBinder.bindYmlToObj(null, content, ESSyncConfig.class, null, envProperties);
- if (config == null) {
- return;
- }
- try {
- config.validate();
- } catch (Exception e) {
- throw new RuntimeException("ERROR Config: " + fileName + " " + e.getMessage(), e);
- }
- esSyncConfig.put(fileName, config);
- });
-
- logger.info("## ES mapping config loaded");
- return esSyncConfig;
- }
-}
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
deleted file mode 100644
index 0faf2de4..00000000
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
+++ /dev/null
@@ -1,440 +0,0 @@
-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()) {
- //当数据列并非原始列时，columnName是空的，例如concat('px',id)
- if(columnItem.getColumnName() != null) {
- 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) {
- List relationSelectFieldItemsTmp = new ArrayList<>();
- for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
- if (fieldItem.getOwners().contains(getAlias())) {
- relationSelectFieldItemsTmp.add(fieldItem);
- }
- }
- relationSelectFieldItems = relationSelectFieldItemsTmp;
- }
- }
- }
- 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
deleted file mode 100644
index 6f81f532..00000000
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java
+++ /dev/null
@@ -1,276 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.es.config;
-
-import static com.alibaba.fastsql.sql.ast.expr.SQLBinaryOperator.BooleanAnd;
-import static com.alibaba.fastsql.sql.ast.expr.SQLBinaryOperator.Equality;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import com.alibaba.fastsql.sql.SQLUtils;
-import com.alibaba.fastsql.sql.ast.SQLExpr;
-import com.alibaba.fastsql.sql.ast.expr.*;
-import com.alibaba.fastsql.sql.ast.statement.*;
-import com.alibaba.fastsql.sql.dialect.mysql.ast.statement.MySqlSelectQueryBlock;
-import com.alibaba.fastsql.sql.dialect.mysql.parser.MySqlStatementParser;
-import com.alibaba.fastsql.sql.parser.ParserException;
-import com.alibaba.fastsql.sql.parser.SQLStatementParser;
-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.es.config.SchemaItem.RelationFieldsPair;
-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
- */
-public class SqlParser {
-
- /**
- * 解析sql
- *
- * @param sql sql
- * @return 视图对象
- */
- public static SchemaItem parse(String sql) {
- try {
- SQLStatementParser parser = new MySqlStatementParser(sql);
- SQLSelectStatement statement = (SQLSelectStatement) parser.parseStatement();
- MySqlSelectQueryBlock sqlSelectQueryBlock = (MySqlSelectQueryBlock) statement.getSelect().getQuery();
-
- SchemaItem schemaItem = new SchemaItem();
- schemaItem.setSql(SQLUtils.toMySqlString(sqlSelectQueryBlock));
- SQLTableSource sqlTableSource = sqlSelectQueryBlock.getFrom();
- List tableItems = new ArrayList<>();
- SqlParser.visitSelectTable(schemaItem, sqlTableSource, tableItems, null);
- tableItems.forEach(tableItem -> schemaItem.getAliasTableItems().put(tableItem.getAlias(), tableItem));
-
- List fieldItems = collectSelectQueryFields(sqlSelectQueryBlock);
- fieldItems.forEach(fieldItem -> schemaItem.getSelectFields().put(fieldItem.getFieldName(), fieldItem));
-
- schemaItem.init();
-
- if (schemaItem.getAliasTableItems().isEmpty() || schemaItem.getSelectFields().isEmpty()) {
- throw new ParserException("Parse sql error");
- }
- return schemaItem;
- } catch (Exception e) {
- throw new ParserException();
- }
- }
-
- /**
- * 归集字段
- *
- * @param sqlSelectQueryBlock sqlSelectQueryBlock
- * @return 字段属性列表
- */
- private static List collectSelectQueryFields(MySqlSelectQueryBlock sqlSelectQueryBlock) {
- 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());
- }
-
- /**
- * 解析字段
- *
- * @param expr sql expr
- * @param fieldItem 字段属性
- */
- private static void visitColumn(SQLExpr expr, FieldItem fieldItem) {
- if (expr instanceof SQLIdentifierExpr) {
- // 无owner
- SQLIdentifierExpr identifierExpr = (SQLIdentifierExpr) expr;
- if (fieldItem.getFieldName() == null) {
- fieldItem.setFieldName(identifierExpr.getName());
- fieldItem.setExpr(identifierExpr.toString());
- }
- ColumnItem columnItem = new ColumnItem();
- columnItem.setColumnName(identifierExpr.getName());
- fieldItem.getOwners().add(null);
- fieldItem.addColumn(columnItem);
- } else if (expr instanceof SQLPropertyExpr) {
- // 有owner
- 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();
- columnItem.setColumnName(sqlPropertyExpr.getName());
- columnItem.setOwner(sqlPropertyExpr.getOwnernName());
- fieldItem.addColumn(columnItem);
- } else if (expr instanceof SQLMethodInvokeExpr) {
- SQLMethodInvokeExpr methodInvokeExpr = (SQLMethodInvokeExpr) expr;
- fieldItem.setMethod(true);
- for (SQLExpr sqlExpr : methodInvokeExpr.getArguments()) {
- visitColumn(sqlExpr, fieldItem);
- }
- } else if (expr instanceof SQLBinaryOpExpr) {
- SQLBinaryOpExpr sqlBinaryOpExpr = (SQLBinaryOpExpr) expr;
- fieldItem.setBinaryOp(true);
- visitColumn(sqlBinaryOpExpr.getLeft(), fieldItem);
- visitColumn(sqlBinaryOpExpr.getRight(), fieldItem);
- } else if (expr instanceof SQLCaseExpr) {
- SQLCaseExpr sqlCaseExpr = (SQLCaseExpr) expr;
- fieldItem.setMethod(true);
- sqlCaseExpr.getItems().forEach(item -> visitColumn(item.getConditionExpr(), fieldItem));
- } else if (expr instanceof SQLCharExpr) {
- SQLCharExpr sqlCharExpr = (SQLCharExpr) expr;
- String owner = null;
- String columnName = null;
- if (sqlCharExpr.getParent() instanceof SQLCaseExpr.Item) {
- owner = ((SQLPropertyExpr) ((SQLCaseExpr.Item) sqlCharExpr.getParent()).getValueExpr()).getOwnernName();
- columnName = ((SQLPropertyExpr) ((SQLCaseExpr.Item) sqlCharExpr.getParent()).getValueExpr()).getName();
- }
- if (fieldItem.getFieldName() == null) {
- fieldItem.setFieldName(columnName);
- fieldItem.setExpr(sqlCharExpr.toString());
- }
- ColumnItem columnItem = new ColumnItem();
- columnItem.setColumnName(columnName);
- columnItem.setOwner(owner);
- fieldItem.getOwners().add(owner);
- fieldItem.addColumn(columnItem);
- }
- }
-
- /**
- * 解析表
- *
- * @param schemaItem 视图对象
- * @param sqlTableSource sqlTableSource
- * @param tableItems 表对象列表
- * @param tableItemTmp 表对象(临时)
- */
- private static void visitSelectTable(SchemaItem schemaItem, SQLTableSource sqlTableSource,
- List tableItems, TableItem tableItemTmp) {
- if (sqlTableSource instanceof SQLExprTableSource) {
- SQLExprTableSource sqlExprTableSource = (SQLExprTableSource) sqlTableSource;
- TableItem tableItem;
- if (tableItemTmp != null) {
- tableItem = tableItemTmp;
- } else {
- tableItem = new TableItem(schemaItem);
- }
- tableItem.setSchema(sqlExprTableSource.getSchema());
- tableItem.setTableName(sqlExprTableSource.getTableName());
- if (tableItem.getAlias() == null) {
- tableItem.setAlias(sqlExprTableSource.getAlias());
- }
- if (tableItems.isEmpty()) {
- // 第一张表为主表
- tableItem.setMain(true);
- }
- tableItems.add(tableItem);
- } else if (sqlTableSource instanceof SQLJoinTableSource) {
- SQLJoinTableSource sqlJoinTableSource = (SQLJoinTableSource) sqlTableSource;
- SQLTableSource leftTableSource = sqlJoinTableSource.getLeft();
- visitSelectTable(schemaItem, leftTableSource, tableItems, null);
- SQLTableSource rightTableSource = sqlJoinTableSource.getRight();
- TableItem rightTableItem = new TableItem(schemaItem);
- // 解析on条件字段
- visitOnCondition(sqlJoinTableSource.getCondition(), rightTableItem);
- visitSelectTable(schemaItem, rightTableSource, tableItems, rightTableItem);
-
- } else if (sqlTableSource instanceof SQLSubqueryTableSource) {
- SQLSubqueryTableSource subQueryTableSource = (SQLSubqueryTableSource) sqlTableSource;
- MySqlSelectQueryBlock sqlSelectQuery = (MySqlSelectQueryBlock) subQueryTableSource.getSelect().getQuery();
- TableItem tableItem;
- if (tableItemTmp != null) {
- tableItem = tableItemTmp;
- } else {
- tableItem = new TableItem(schemaItem);
- }
- tableItem.setAlias(subQueryTableSource.getAlias());
- tableItem.setSubQuerySql(SQLUtils.toMySqlString(sqlSelectQuery));
- tableItem.setSubQuery(true);
- tableItem.setSubQueryFields(collectSelectQueryFields(sqlSelectQuery));
- visitSelectTable(schemaItem, sqlSelectQuery.getFrom(), tableItems, tableItem);
- }
- }
-
- /**
- * 解析on条件
- *
- * @param expr sql expr
- * @param tableItem 表对象
- */
- private static void visitOnCondition(SQLExpr expr, TableItem tableItem) {
- if (!(expr instanceof SQLBinaryOpExpr)) {
- throw new UnsupportedOperationException();
- }
- SQLBinaryOpExpr sqlBinaryOpExpr = (SQLBinaryOpExpr) expr;
- if (sqlBinaryOpExpr.getOperator() == BooleanAnd) {
- visitOnCondition(sqlBinaryOpExpr.getLeft(), tableItem);
- visitOnCondition(sqlBinaryOpExpr.getRight(), tableItem);
- } else if (sqlBinaryOpExpr.getOperator() == Equality) {
- FieldItem leftFieldItem = new FieldItem();
- visitColumn(sqlBinaryOpExpr.getLeft(), leftFieldItem);
- if (leftFieldItem.getColumnItems().size() != 1 || leftFieldItem.isMethod() || leftFieldItem.isBinaryOp()) {
- throw new UnsupportedOperationException("Unsupported for complex of on-condition");
- }
- FieldItem rightFieldItem = new FieldItem();
- visitColumn(sqlBinaryOpExpr.getRight(), rightFieldItem);
- if (rightFieldItem.getColumnItems().size() != 1 || rightFieldItem.isMethod()
- || rightFieldItem.isBinaryOp()) {
- throw new UnsupportedOperationException("Unsupported for complex of on-condition");
- }
- tableItem.getRelationFields().add(new RelationFieldsPair(leftFieldItem, rightFieldItem));
- } else {
- throw new UnsupportedOperationException("Unsupported for complex of on-condition");
- }
- }
-
- public static MySqlSelectQueryBlock parseSQLSelectQueryBlock(String sql) {
- if (sql == null || "".equals(sql)) {
- return null;
- }
- SQLStatementParser parser = new MySqlStatementParser(sql);
- SQLSelectStatement statement = (SQLSelectStatement) parser.parseStatement();
- return (MySqlSelectQueryBlock) statement.getSelect().getQuery();
- }
-
- public static String parse4SQLSelectItem(MySqlSelectQueryBlock sqlSelectQueryBlock) {
- List selectItems = sqlSelectQueryBlock.getSelectList();
- StringBuilder subSql = new StringBuilder();
- int i = 0;
- for (SQLSelectItem sqlSelectItem : selectItems) {
- if (i != 0) {
- subSql.append(",");
- } else {
- i++;
- }
- subSql.append(SQLUtils.toMySqlString(sqlSelectItem));
- }
- return subSql.toString();
- }
-
- public static String parse4FromTableSource(MySqlSelectQueryBlock sqlSelectQueryBlock) {
- SQLTableSource sqlTableSource = sqlSelectQueryBlock.getFrom();
- return SQLUtils.toMySqlString(sqlTableSource);
- }
-
- public static String parse4WhereItem(MySqlSelectQueryBlock sqlSelectQueryBlock) {
- SQLExpr sqlExpr = sqlSelectQueryBlock.getWhere();
- if (sqlExpr != null) {
- return SQLUtils.toMySqlString(sqlExpr);
- }
- return null;
- }
-
- public static String parse4GroupBy(MySqlSelectQueryBlock sqlSelectQueryBlock) {
- SQLSelectGroupByClause expr = sqlSelectQueryBlock.getGroupBy();
- if (expr != null) {
- return SQLUtils.toMySqlString(expr);
- }
- return null;
- }
-}
diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java
deleted file mode 100644
index cc41e728..00000000
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/monitor/ESConfigMonitor.java
+++ /dev/null
@@ -1,147 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.es.monitor;
-
-import java.io.File;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import org.apache.commons.io.filefilter.FileFilterUtils;
-import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
-import org.apache.commons.io.monitor.FileAlterationMonitor;
-import org.apache.commons.io.monitor.FileAlterationObserver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.alibaba.druid.pool.DruidDataSource;
-import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
-import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
-import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
-import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
-import com.alibaba.otter.canal.client.adapter.es.config.SqlParser;
-import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
-import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-
-public class ESConfigMonitor {
-
- private static final Logger logger = LoggerFactory.getLogger(ESConfigMonitor.class);
-
- private static final String adapterName = "es";
-
- private ESAdapter esAdapter;
-
- private Properties envProperties;
-
- private FileAlterationMonitor fileMonitor;
-
- public void init(ESAdapter esAdapter, Properties envProperties) {
- this.esAdapter = esAdapter;
- this.envProperties = envProperties;
- File confDir = Util.getConfDirPath(adapterName);
- try {
- FileAlterationObserver observer = new FileAlterationObserver(confDir,
- FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
- FileListener listener = new FileListener();
- observer.addListener(listener);
- fileMonitor = new FileAlterationMonitor(3000, observer);
- fileMonitor.start();
-
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- public void destroy() {
- try {
- fileMonitor.stop();
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- private class FileListener extends FileAlterationListenerAdaptor {
-
- @Override
- public void onFileCreate(File file) {
- super.onFileCreate(file);
- try {
- // 加载新增的配置文件
- String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
- ESSyncConfig config = YmlConfigBinder
- .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
- if (config != null) {
- config.validate();
- addConfigToCache(file, config);
- logger.info("Add a new es mapping config: {} to canal adapter", file.getName());
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- @Override
- public void onFileChange(File file) {
- super.onFileChange(file);
-
- try {
- if (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
- // 加载配置文件
- String configContent = MappingConfigsLoader
- .loadConfig(adapterName + File.separator + file.getName());
- if (configContent == null) {
- onFileDelete(file);
- return;
- }
- ESSyncConfig config = YmlConfigBinder
- .bindYmlToObj(null, configContent, ESSyncConfig.class, null, envProperties);
- if (config == null) {
- return;
- }
- config.validate();
- if (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
- deleteConfigFromCache(file);
- }
- addConfigToCache(file, config);
-
- logger.info("Change a es mapping config: {} of canal adapter", file.getName());
- }else{
- onFileCreate(file);
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- @Override
- public void onFileDelete(File file) {
- super.onFileDelete(file);
-
- try {
- if (esAdapter.getEsSyncConfig().containsKey(file.getName())) {
- deleteConfigFromCache(file);
-
- logger.info("Delete a es mapping config: {} of canal adapter", file.getName());
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- }
- }
-
- private void addConfigToCache(File file, ESSyncConfig config) {
- esAdapter.getEsSyncConfig().put(file.getName(), config);
- esAdapter.addSyncConfigToCache(file.getName(),config);
- }
-
- private void deleteConfigFromCache(File file) {
- esAdapter.getEsSyncConfig().remove(file.getName());
- for (Map configMap : esAdapter.getDbTableEsSyncConfig().values()) {
- if (configMap != null) {
- configMap.remove(file.getName());
- }
- }
-
- }
- }
-}
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
deleted file mode 100644
index 218a32e8..00000000
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
+++ /dev/null
@@ -1,217 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.es.service;
-
-import java.sql.SQLException;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.sql.DataSource;
-
-import org.apache.commons.lang.StringUtils;
-import org.elasticsearch.action.bulk.BulkItemResponse;
-import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.index.query.QueryBuilders;
-import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.search.SearchHit;
-
-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.FieldItem;
-import com.alibaba.otter.canal.client.adapter.es.support.ESConnection;
-import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESBulkRequest;
-import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESIndexRequest;
-import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESSearchRequest;
-import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESUpdateRequest;
-import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
-import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService;
-import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
-import com.alibaba.otter.canal.client.adapter.support.EtlResult;
-import com.alibaba.otter.canal.client.adapter.support.Util;
-
-/**
- * ES ETL Service
- *
- * @author rewerma 2018-11-01
- * @version 1.0.0
- */
-public class ESEtlService extends AbstractEtlService {
-
- private ESConnection esConnection;
- private ESTemplate esTemplate;
- private ESSyncConfig config;
-
- public ESEtlService(ESConnection esConnection, ESSyncConfig config){
- super("ES", config);
- this.esConnection = esConnection;
- this.esTemplate = new ESTemplate(esConnection);
- this.config = config;
- }
-
- public EtlResult importData(List params) {
- ESMapping mapping = config.getEsMapping();
- logger.info("start etl to import data to index: {}", mapping.get_index());
- String sql = mapping.getSql();
- return importData(sql, params);
- }
-
- private void processFailBulkResponse(BulkResponse bulkResponse) {
- for (BulkItemResponse response : bulkResponse.getItems()) {
- if (!response.isFailed()) {
- continue;
- }
-
- if (response.getFailure().getStatus() == RestStatus.NOT_FOUND) {
- logger.warn(response.getFailureMessage());
- } else {
- logger.error("全量导入数据有误 {}", response.getFailureMessage());
- throw new RuntimeException("全量数据 etl 异常: " + response.getFailureMessage());
- }
- }
- }
-
- protected boolean executeSqlImport(DataSource ds, String sql, List