diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml
index e62fc386..f018c490 100644
--- a/client-adapter/common/pom.xml
+++ b/client-adapter/common/pom.xml
@@ -14,7 +14,7 @@
com.alibaba.otter
canal.protocol
- 1.1.2-SNAPSHOT
+ ${canal_version}
joda-time
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java
index f08c6435..3d16a067 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java
@@ -19,6 +19,12 @@ public class CanalClientConfig {
private Boolean flatMessage = true; // 是否已flatMessage模式传输, 只适用于mq模式
+ private Integer batchSize; // 批大小
+
+ private Integer retry; // 重试次数
+
+ private Long timeout; // 消费超时时间
+
private List mqTopics; // mq topic 列表
private List canalInstances; // tcp 模式下 canal 实例列表, 与mq模式不能共存!!
@@ -63,6 +69,30 @@ public class CanalClientConfig {
this.flatMessage = flatMessage;
}
+ public Integer getBatchSize() {
+ return batchSize;
+ }
+
+ public void setBatchSize(Integer batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ public Integer getRetry() {
+ return retry;
+ }
+
+ public void setRetry(Integer retry) {
+ this.retry = retry;
+ }
+
+ public Long getTimeout() {
+ return timeout;
+ }
+
+ public void setTimeout(Long timeout) {
+ this.timeout = timeout;
+ }
+
public List getCanalInstances() {
return canalInstances;
}
@@ -73,9 +103,9 @@ public class CanalClientConfig {
public static class CanalInstance {
- private String instance; // 实例名
+ private String instance; // 实例名
- private List groups; // 适配器分组列表
+ private List groups; // 适配器分组列表
public String getInstance() {
return instance;
@@ -112,9 +142,9 @@ public class CanalClientConfig {
public static class MQTopic {
- private String mqMode; // mq模式 kafka or rocketMQ
+ private String mqMode; // mq模式 kafka or rocketMQ
- private String topic; // topic名
+ private String topic; // topic名
private List groups = new ArrayList<>(); // 分组列表
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java
index 5b5c1bbb..24d43a0f 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java
@@ -292,10 +292,13 @@ public class ExtensionLoader {
private String getJarDirectoryPath() {
URL url = Thread.currentThread().getContextClassLoader().getResource("");
- if (url == null) {
- throw new IllegalStateException("failed to get class loader resource");
+ String dirtyPath;
+ if (url != null) {
+ dirtyPath = url.toString();
+ } else {
+ File file = new File("");
+ dirtyPath = file.getAbsolutePath();
}
- String dirtyPath = url.toString();
String jarPath = dirtyPath.replaceAll("^.*file:/", ""); // removes
// file:/ and
// everything
diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml
new file mode 100644
index 00000000..bff1bffd
--- /dev/null
+++ b/client-adapter/elasticsearch/pom.xml
@@ -0,0 +1,77 @@
+
+
+
+ canal.client-adapter
+ com.alibaba.otter
+ 1.1.2-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
+
+
+ org.yaml
+ snakeyaml
+ 1.19
+ provided
+
+
+ com.alibaba.fastsql
+ fastsql
+ 2.0.0_preview_644
+
+
+ org.elasticsearch
+ elasticsearch
+ 6.2.3
+
+
+ org.elasticsearch.client
+ transport
+ 6.2.3
+
+
+
+ junit
+ junit
+ 4.12
+ test
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-assembly-plugin
+ 2.4
+
+
+ jar-with-dependencies
+
+
+
+
+ make-assembly
+ package
+
+ single
+
+
+
+
+
+
+
+
\ No newline at end of file
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
new file mode 100644
index 00000000..7d10d702
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
@@ -0,0 +1,151 @@
+package com.alibaba.otter.canal.client.adapter.es;
+
+import java.net.InetAddress;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.transport.client.PreBuiltTransportClient;
+
+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.service.ESEtlService;
+import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService;
+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 TransportClient transportClient;
+
+ private ESSyncService esSyncService;
+
+ public TransportClient getTransportClient() {
+ return transportClient;
+ }
+
+ public ESSyncService getEsSyncService() {
+ return esSyncService;
+ }
+
+ @Override
+ public void init(OuterAdapterConfig configuration) {
+ try {
+ ESSyncConfigLoader.load();
+
+ Map properties = configuration.getProperties();
+ Settings.Builder settingBuilder = Settings.builder();
+ properties.forEach(settingBuilder::put);
+ Settings settings = settingBuilder.build();
+ transportClient = new PreBuiltTransportClient(settings);
+ String[] hostArray = configuration.getHosts().split(",");
+ for (String host : hostArray) {
+ int i = host.indexOf(":");
+ transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)),
+ Integer.parseInt(host.substring(i + 1))));
+ }
+ ESTemplate esTemplate = new ESTemplate(transportClient);
+ esSyncService = new ESSyncService(esTemplate);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void sync(Dml dml) {
+ esSyncService.sync(dml);
+ }
+
+ @Override
+ public EtlResult etl(String task, List params) {
+ EtlResult etlResult = new EtlResult();
+ ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().get(task);
+ if (config != null) {
+ DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
+ ESEtlService esEtlService = new ESEtlService(transportClient, config);
+ if (dataSource != null) {
+ return esEtlService.importData(params, false);
+ } else {
+ etlResult.setSucceeded(false);
+ etlResult.setErrorMessage("DataSource not found");
+ return etlResult;
+ }
+ } else {
+ StringBuilder resultMsg = new StringBuilder();
+ boolean resSuccess = true;
+ // ds不为空说明传入的是datasourceKey
+ for (ESSyncConfig configTmp : ESSyncConfigLoader.getEsSyncConfig().values()) {
+ // 取所有的destination为task的配置
+ if (configTmp.getDestination().equals(task)) {
+ ESEtlService esEtlService = new ESEtlService(transportClient, configTmp);
+ EtlResult etlRes = esEtlService.importData(params, false);
+ 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 = ESSyncConfigLoader.getEsSyncConfig().get(task);
+ ESMapping mapping = config.getEsMapping();
+ SearchResponse response = transportClient.prepareSearch(mapping.get_index())
+ .setTypes(mapping.get_type())
+ .setSize(0)
+ .get();
+
+ 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 (transportClient != null) {
+ transportClient.close();
+ }
+ }
+
+ @Override
+ public String getDestination(String task) {
+ ESSyncConfig config = ESSyncConfigLoader.getEsSyncConfig().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
new file mode 100644
index 00000000..9d4a9da2
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
@@ -0,0 +1,184 @@
+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;
+
+/**
+ * ES 映射配置
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class ESSyncConfig {
+
+ private String dataSourceKey; // 数据源key
+
+ 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.pk == null) {
+ throw new NullPointerException("esMapping._id and 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 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 static class ESMapping {
+
+ private String _index;
+ private String _type;
+ private String _id;
+ private String pk;
+ private String parent;
+ 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 String getPk() {
+ return pk;
+ }
+
+ public void setPk(String pk) {
+ this.pk = pk;
+ }
+
+ public String getParent() {
+ return parent;
+ }
+
+ public void setParent(String parent) {
+ this.parent = parent;
+ }
+
+ 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 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;
+ }
+ }
+}
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
new file mode 100644
index 00000000..cb1cc274
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java
@@ -0,0 +1,127 @@
+package com.alibaba.otter.canal.client.adapter.es.config;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.Yaml;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+
+/**
+ * ES 配置装载器
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class ESSyncConfigLoader {
+
+ private static Logger logger = LoggerFactory
+ .getLogger(ESSyncConfigLoader.class);
+
+ private static final String BASE_PATH = "es";
+
+ private static volatile Map esSyncConfig = new LinkedHashMap<>(); // 文件名对应配置
+ private static volatile Map> dbTableEsSyncConfig = new LinkedHashMap<>(); // schema-table对应配置
+
+ public static Map getEsSyncConfig() {
+ return esSyncConfig;
+ }
+
+ public static Map> getDbTableEsSyncConfig() {
+ return dbTableEsSyncConfig;
+ }
+
+ public static synchronized void load() {
+ logger.info("## Start loading mapping config ... ");
+ Collection configs = AdapterConfigs.get("es");
+ if (configs == null) {
+ return;
+ }
+ for (String c : configs) {
+ if (c == null) {
+ continue;
+ }
+ c = c.trim();
+ if (c.equals("") || c.startsWith("#")) {
+ continue;
+ }
+
+ ESSyncConfig config;
+ String configContent = null;
+
+ if (c.endsWith(".yml")) {
+ configContent = readConfigContent(BASE_PATH + "/" + c);
+ }
+
+ config = new Yaml().loadAs(configContent, ESSyncConfig.class);
+
+ try {
+ config.validate();
+ 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 -> {
+ List esSyncConfigs = dbTableEsSyncConfig
+ .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new ArrayList<>());
+ esSyncConfigs.add(config);
+ });
+ } catch (Exception e) {
+ throw new RuntimeException("ERROR Config: " + c, e);
+ }
+ esSyncConfig.put(c, config);
+ }
+
+ logger.info("## Mapping config loaded");
+ }
+
+ private static String readConfigContent(String config) {
+ InputStream in = null;
+ try {
+ // 先取本地文件,再取类路径
+ File configFile = new File("config/" + config);
+ if (configFile.exists()) {
+ in = new FileInputStream(configFile);
+ } else {
+ in = ESSyncConfigLoader.class.getClassLoader().getResourceAsStream(config);
+ }
+ if (in == null) {
+ throw new RuntimeException("Config file: " + config + " not found.");
+ }
+
+ byte[] bytes = new byte[in.available()];
+ in.read(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new RuntimeException("Read yml config error ", e);
+ } finally {
+ try {
+ if (in != null) {
+ in.close();
+ }
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+ }
+}
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
new file mode 100644
index 00000000..fc6e076c
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
@@ -0,0 +1,422 @@
+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;
+ }
+ }
+}
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
new file mode 100644
index 00000000..0c2d7b2c
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java
@@ -0,0 +1,210 @@
+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.SQLBinaryOpExpr;
+import com.alibaba.fastsql.sql.ast.expr.SQLIdentifierExpr;
+import com.alibaba.fastsql.sql.ast.expr.SQLMethodInvokeExpr;
+import com.alibaba.fastsql.sql.ast.expr.SQLPropertyExpr;
+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());
+ 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());
+ }
+ 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.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);
+ }
+ }
+
+ /**
+ * 解析表
+ *
+ * @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");
+ }
+ }
+}
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
new file mode 100644
index 00000000..51f0a924
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
@@ -0,0 +1,286 @@
+package com.alibaba.otter.canal.client.adapter.es.service;
+
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import javax.sql.DataSource;
+
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.SearchHit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.druid.pool.DruidDataSource;
+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.ESSyncUtil;
+import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
+import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.EtlResult;
+import com.google.common.base.Joiner;
+
+/**
+ * ES ETL Service
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class ESEtlService {
+
+ private static Logger logger = LoggerFactory.getLogger(ESEtlService.class);
+
+ private TransportClient transportClient;
+ private ESTemplate esTemplate;
+ private ESSyncConfig config;
+
+ public ESEtlService(TransportClient transportClient, ESSyncConfig config){
+ this.transportClient = transportClient;
+ this.esTemplate = new ESTemplate(transportClient);
+ this.config = config;
+ }
+
+ public EtlResult importData(List params, boolean bulk) {
+ EtlResult etlResult = new EtlResult();
+ AtomicLong impCount = new AtomicLong();
+ List errMsg = new ArrayList<>();
+ String esIndex = "";
+ if (config == null) {
+ logger.warn("esSycnCofnig is null, etl go end !");
+ etlResult.setErrorMessage("esSycnCofnig is null, etl go end !");
+ return etlResult;
+ }
+
+ ESMapping mapping = config.getEsMapping();
+
+ esIndex = mapping.get_index();
+ DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(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);
+
+ logger.info("etl from db: {}, to es index: {}", schema, esIndex);
+ long start = System.currentTimeMillis();
+ try {
+ String sql = mapping.getSql();
+
+ // 拼接条件
+ if (mapping.getEtlCondition() != null && params != null) {
+ String etlCondition = mapping.getEtlCondition();
+ int size = params.size();
+ for (int i = 0; i < size; i++) {
+ etlCondition = etlCondition.replace("{" + i + "}", params.get(i));
+ }
+
+ sql += " " + etlCondition;
+ }
+
+ if (logger.isDebugEnabled()) {
+ 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);
+ }
+ 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);
+ }
+ } else {
+ logger.info("自动ETL,无需统计记录总条数,直接进行ETL, index: {}", esIndex);
+ executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
+ }
+
+ logger.info("数据全量导入完成,一共导入 {} 条数据, 耗时: {}", impCount.get(), System.currentTimeMillis() - start);
+ etlResult.setResultMessage("导入ES索引 " + esIndex + " 数据:" + impCount.get() + " 条");
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ errMsg.add(esIndex + " etl failed! ==>" + e.getMessage());
+ }
+ if (errMsg.isEmpty()) {
+ etlResult.setSucceeded(true);
+ } else {
+ etlResult.setErrorMessage(Joiner.on("\n").join(errMsg));
+ }
+ return etlResult;
+ }
+
+ private void processFailBulkResponse(BulkResponse bulkResponse, boolean hasParent) {
+ 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());
+ }
+ }
+ }
+
+ private boolean executeSqlImport(DataSource ds, String sql, ESMapping mapping, AtomicLong impCount,
+ List errMsg) {
+ try {
+ ESSyncUtil.sqlRS(ds, sql, rs -> {
+ int count = 0;
+ try {
+ BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
+
+ long batchBegin = System.currentTimeMillis();
+ while (rs.next()) {
+ Map esFieldData = new LinkedHashMap<>();
+ for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) {
+
+ // 如果是主键字段则不插入
+ if (fieldItem.getFieldName().equals(mapping.get_id())) {
+ continue;
+ }
+
+ String fieldName = fieldItem.getFieldName();
+ if (mapping.getSkips().contains(fieldName)) {
+ continue;
+ }
+
+ Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
+ esFieldData.put(fieldName, val);
+ }
+ Object idVal = null;
+ if (mapping.get_id() != null) {
+ idVal = rs.getObject(mapping.get_id());
+ }
+
+ if (idVal != null) {
+ if (mapping.getParent() == null) {
+ 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
+ }
+ }
+
+ if (bulkRequestBuilder.numberOfActions() % mapping.getCommitBatch() == 0
+ && bulkRequestBuilder.numberOfActions() > 0) {
+ long esBatchBegin = System.currentTimeMillis();
+ BulkResponse rp = bulkRequestBuilder.execute().actionGet();
+ if (rp.hasFailures()) {
+ this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent()));
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
+ (System.currentTimeMillis() - batchBegin),
+ (System.currentTimeMillis() - esBatchBegin),
+ bulkRequestBuilder.numberOfActions(),
+ mapping.get_index());
+ }
+ batchBegin = System.currentTimeMillis();
+ bulkRequestBuilder = transportClient.prepareBulk();
+ }
+ count++;
+ impCount.incrementAndGet();
+ }
+
+ if (bulkRequestBuilder.numberOfActions() > 0) {
+ long esBatchBegin = System.currentTimeMillis();
+ BulkResponse rp = bulkRequestBuilder.execute().actionGet();
+ if (rp.hasFailures()) {
+ this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent()));
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
+ (System.currentTimeMillis() - batchBegin),
+ (System.currentTimeMillis() - esBatchBegin),
+ bulkRequestBuilder.numberOfActions(),
+ mapping.get_index());
+ }
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage());
+ throw new RuntimeException(e);
+ }
+ return count;
+ });
+
+ return true;
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ return false;
+ }
+ }
+}
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
new file mode 100644
index 00000000..3927c6ab
--- /dev/null
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
@@ -0,0 +1,863 @@
+package com.alibaba.otter.canal.client.adapter.es.service;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
+import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfigLoader;
+import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
+import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem;
+import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
+import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.TableItem;
+import com.alibaba.otter.canal.client.adapter.es.support.ESSyncUtil;
+import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate;
+import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+
+/**
+ * ES 同步 Service
+ *
+ * @author rewerma 2018-11-01
+ * @version 1.0.0
+ */
+public class ESSyncService {
+
+ private static Logger logger = LoggerFactory.getLogger(ESSyncService.class);
+
+ private ESTemplate esTemplate;
+
+ public ESSyncService(ESTemplate esTemplate){
+ this.esTemplate = esTemplate;
+ }
+
+ public void sync(Dml dml) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("DML: {}", dml.toString());
+ }
+ long begin = System.currentTimeMillis();
+ String database = dml.getDatabase();
+ String table = dml.getTable();
+ List esSyncConfigs = ESSyncConfigLoader.getDbTableEsSyncConfig().get(database + "-" + table);
+ if (esSyncConfigs != null) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Destination: {}, database:{}, table:{}, type:{}, effect index count: {}",
+ dml.getDestination(),
+ dml.getDatabase(),
+ dml.getTable(),
+ dml.getType(),
+ esSyncConfigs.size());
+ }
+
+ for (ESSyncConfig config : esSyncConfigs) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Prepared to sync index: {}, destination: {}",
+ config.getEsMapping().get_index(),
+ dml.getDestination());
+ }
+ this.sync(config, dml);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Sync completed: {}, destination: {}",
+ config.getEsMapping().get_index(),
+ dml.getDestination());
+ }
+ }
+ if (logger.isTraceEnabled()) {
+ logger.trace("Sync elapsed time: {} ms, effect index count:{}, destination: {}",
+ (System.currentTimeMillis() - begin),
+ esSyncConfigs.size(),
+ dml.getDestination());
+ }
+ }
+ }
+
+ public void sync(ESSyncConfig config, Dml dml) {
+ try {
+ // 如果是按时间戳定时更新则返回
+ if (config.getEsMapping().isSyncByTimestamp()) {
+ return;
+ }
+
+ long begin = System.currentTimeMillis();
+
+ String type = dml.getType();
+ if (type != null && type.equalsIgnoreCase("INSERT")) {
+ insert(config, dml);
+ } else if (type != null && type.equalsIgnoreCase("UPDATE")) {
+ update(config, dml);
+ } else if (type != null && type.equalsIgnoreCase("DELETE")) {
+ delete(config, dml);
+ }
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("Sync elapsed time: {} ms,destination: {}, es index: {}",
+ (System.currentTimeMillis() - begin),
+ dml.getDestination(),
+ config.getEsMapping().get_index());
+ }
+ } catch (Exception e) {
+ logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().get_index(), dml);
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * 插入操作dml
+ *
+ * @param config es配置
+ * @param dml dml数据
+ */
+ private void insert(ESSyncConfig config, Dml dml) {
+ List