chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1523 from rewerma/master

Browse Source 
es adapter新增upsert属性
This commit is contained in:
agapple
2019-02-25 14:50:58 +08:00
committed by GitHub
parent 0fff514876 fcc1fc78ae
commit 1473a4a6b5
11 changed files with 878 additions and 976 deletions
Download Patch File Download Diff File Expand all files Collapse all files

311
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java
View File

@@ -1,156 +1,155 @@
package com.alibaba.otter.canal.client.adapter.config; package com.alibaba.otter.canal.client.adapter.config;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import java.util.Properties; import java.util.Properties;
import org.springframework.util.PropertyPlaceholderHelper; import org.springframework.util.PropertyPlaceholderHelper;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
import com.alibaba.otter.canal.client.adapter.config.bind.PropertiesConfigurationFactory; import com.alibaba.otter.canal.client.adapter.config.bind.PropertiesConfigurationFactory;
import com.alibaba.otter.canal.client.adapter.config.common.*; import com.alibaba.otter.canal.client.adapter.config.common.*;
/** /**
* 将yaml内容绑定到指定对象, 遵循spring yml的绑定规范 * 将yaml内容绑定到指定对象, 遵循spring yml的绑定规范
* *
* @author reweerma 2019-2-1 上午9:14:02 * @author reweerma 2019-2-1 上午9:14:02
* @version 1.0.0 * @version 1.0.0
*/ */
public class YmlConfigBinder { public class YmlConfigBinder {
/** /**
* 将当前内容绑定到指定对象 * 将当前内容绑定到指定对象
* *
* @param content yml内容 * @param content yml内容
* @param clazz 指定对象类型 * @param clazz 指定对象类型
* @return 对象 * @return 对象
*/ */
public static <T> T bindYmlToObj(String content, Class<T> clazz) { public static <T> T bindYmlToObj(String content, Class<T> clazz) {
return bindYmlToObj(null, content, clazz, null); return bindYmlToObj(null, content, clazz, null);
} }
/** /**
* 将当前内容绑定到指定对象并指定内容编码格式 * 将当前内容绑定到指定对象并指定内容编码格式
* *
* @param content yml内容 * @param content yml内容
* @param clazz 指定对象类型 * @param clazz 指定对象类型
* @param charset yml内容编码格式 * @param charset yml内容编码格式
* @return 对象 * @return 对象
*/ */
public static <T> T bindYmlToObj(String content, Class<T> clazz, String charset) { public static <T> T bindYmlToObj(String content, Class<T> clazz, String charset) {
return bindYmlToObj(null, content, clazz, charset); return bindYmlToObj(null, content, clazz, charset);
} }
/** /**
* 将当前内容指定前缀部分绑定到指定对象 * 将当前内容指定前缀部分绑定到指定对象
* *
* @param prefix 指定前缀 * @param prefix 指定前缀
* @param content yml内容 * @param content yml内容
* @param clazz 指定对象类型 * @param clazz 指定对象类型
* @return 对象 * @return 对象
*/ */
public static <T> T bindYmlToObj(String prefix, String content, Class<T> clazz) { public static <T> T bindYmlToObj(String prefix, String content, Class<T> clazz) {
return bindYmlToObj(prefix, content, clazz, null); return bindYmlToObj(prefix, content, clazz, null);
} }
/** /**
* 将当前内容指定前缀部分绑定到指定对象并指定内容编码格式 * 将当前内容指定前缀部分绑定到指定对象并指定内容编码格式
* *
* @param prefix 指定前缀 * @param prefix 指定前缀
* @param content yml内容 * @param content yml内容
* @param clazz 指定对象类型 * @param clazz 指定对象类型
* @param charset yml内容编码格式 * @param charset yml内容编码格式
* @return 对象 * @return 对象
*/ */
public static <T> T bindYmlToObj(String prefix, String content, Class<T> clazz, String charset) { public static <T> T bindYmlToObj(String prefix, String content, Class<T> clazz, String charset) {
return bindYmlToObj(prefix, content, clazz, charset, null); return bindYmlToObj(prefix, content, clazz, charset, null);
} }
/** /**
* 将当前内容指定前缀部分绑定到指定对象并用环境变量中的属性替换占位符, 例: 当前内容有属性 zkServers: * 将当前内容指定前缀部分绑定到指定对象并用环境变量中的属性替换占位符, 例: 当前内容有属性 zkServers: ${zookeeper.servers}
* ${zookeeper.servers} 在envProperties中有属性 zookeeper.servers: * 在envProperties中有属性 zookeeper.servers:
* 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 则当前内容 zkServers 会被替换为 * 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 则当前内容 zkServers 会被替换为
* zkServers: 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 注: 假设绑定的类中 * zkServers: 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 注: 假设绑定的类中
* zkServers 属性是 List<String> 对象, 则会自动映射成List * zkServers 属性是 List<String> 对象, 则会自动映射成List
* *
* @param prefix 指定前缀 * @param prefix 指定前缀
* @param content yml内容 * @param content yml内容
* @param clazz 指定对象类型 * @param clazz 指定对象类型
* @param charset yml内容编码格式 * @param charset yml内容编码格式
* @return 对象 * @return 对象
*/ */
public static <T> T bindYmlToObj(String prefix, String content, Class<T> clazz, String charset, public static <T> T bindYmlToObj(String prefix, String content, Class<T> clazz, String charset,
Properties baseProperties) { Properties baseProperties) {
try { try {
byte[] contentBytes; byte[] contentBytes;
if (charset == null) { if (charset == null) {
contentBytes = content.getBytes(); contentBytes = content.getBytes();
} else { } else {
contentBytes = content.getBytes(charset); contentBytes = content.getBytes(charset);
} }
YamlPropertySourceLoader propertySourceLoader = new YamlPropertySourceLoader(); YamlPropertySourceLoader propertySourceLoader = new YamlPropertySourceLoader();
Resource configResource = new ByteArrayResource(contentBytes); Resource configResource = new ByteArrayResource(contentBytes);
PropertySource<?> propertySource = propertySourceLoader.load("manualBindConfig", configResource, null); PropertySource<?> propertySource = propertySourceLoader.load("manualBindConfig", configResource, null);
if (propertySource == null) { if (propertySource == null) {
return null; return null;
} }
Properties properties = new Properties(); Properties properties = new Properties();
Map<String, Object> propertiesRes = new LinkedHashMap<>(); Map<String, Object> propertiesRes = new LinkedHashMap<>();
if (!StringUtils.isEmpty(prefix) && !prefix.endsWith(".")) { if (!StringUtils.isEmpty(prefix) && !prefix.endsWith(".")) {
prefix = prefix + "."; prefix = prefix + ".";
} }
properties.putAll((Map<?, ?>) propertySource.getSource()); properties.putAll((Map<?, ?>) propertySource.getSource());
if (baseProperties != null) { if (baseProperties != null) {
baseProperties.putAll(properties); baseProperties.putAll(properties);
properties = baseProperties; properties = baseProperties;
} }
for (Object o : ((Map<?, ?>) propertySource.getSource()).entrySet()) { for (Map.Entry<?, ?> entry : ((Map<?, ?>) propertySource.getSource()).entrySet()) {
Map.Entry<?, ?> entry = (Map.Entry<?, ?>) o; String key = (String) entry.getKey();
String key = (String) entry.getKey(); Object value = entry.getValue();
Object value = entry.getValue();
if (prefix != null) {
if (prefix != null) { if (key != null && key.startsWith(prefix)) {
if (key != null && key.startsWith(prefix)) { key = key.substring(prefix.length());
key = key.substring(prefix.length()); } else {
} else { continue;
continue; }
} }
}
if (value instanceof String && ((String) value).contains("${") && ((String) value).contains("}")) {
if (value != null && value.toString().contains("${")) { PropertyPlaceholderHelper propertyPlaceholderHelper = new PropertyPlaceholderHelper("${", "}");
PropertyPlaceholderHelper propertyPlaceholderHelper = new PropertyPlaceholderHelper("${", "}"); value = propertyPlaceholderHelper.replacePlaceholders((String) value, properties);
value = propertyPlaceholderHelper.replacePlaceholders(value.toString(), properties); }
}
propertiesRes.put(key, value);
propertiesRes.put(key, value); }
}
if (propertiesRes.isEmpty()) {
if (propertiesRes.isEmpty()) { return null;
return null; }
}
propertySource = new MapPropertySource(propertySource.getName(), propertiesRes);
propertySource = new MapPropertySource(propertySource.getName(), propertiesRes);
T target = clazz.newInstance();
T target = clazz.newInstance();
MutablePropertySources propertySources = new MutablePropertySources();
MutablePropertySources propertySources = new MutablePropertySources(); propertySources.addFirst(propertySource);
propertySources.addFirst(propertySource);
PropertiesConfigurationFactory<Object> factory = new PropertiesConfigurationFactory<Object>(target);
PropertiesConfigurationFactory<Object> factory = new PropertiesConfigurationFactory<Object>(target); factory.setPropertySources(propertySources);
factory.setPropertySources(propertySources); factory.setIgnoreInvalidFields(true);
factory.setIgnoreInvalidFields(true); factory.setIgnoreUnknownFields(true);
factory.setIgnoreUnknownFields(true);
factory.bindPropertiesToTarget();
factory.bindPropertiesToTarget();
return target;
return target; } catch (Exception e) {
} catch (Exception e) { throw new RuntimeException(e);
throw new RuntimeException(e); }
} }
} }
}

11
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
View File

@@ -144,11 +144,14 @@ public class ESAdapter implements OuterAdapter {
return; return;
} }
for (Dml dml : dmls) { for (Dml dml : dmls) {
sync(dml); if (!dml.getIsDdl()) {
sync(dml);
}
} }
esSyncService.commit(); // 批次统一提交
} }
public void sync(Dml dml) { private void sync(Dml dml) {
String database = dml.getDatabase(); String database = dml.getDatabase();
String table = dml.getTable(); String table = dml.getTable();
Map<String, ESSyncConfig> configMap; Map<String, ESSyncConfig> configMap;
@@ -174,7 +177,7 @@ public class ESAdapter implements OuterAdapter {
DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
ESEtlService esEtlService = new ESEtlService(transportClient, config); ESEtlService esEtlService = new ESEtlService(transportClient, config);
if (dataSource != null) { if (dataSource != null) {
return esEtlService.importData(params, false); return esEtlService.importData(params);
} else { } else {
etlResult.setSucceeded(false); etlResult.setSucceeded(false);
etlResult.setErrorMessage("DataSource not found"); etlResult.setErrorMessage("DataSource not found");
@@ -188,7 +191,7 @@ public class ESAdapter implements OuterAdapter {
// 取所有的destination为task的配置 // 取所有的destination为task的配置
if (configTmp.getDestination().equals(task)) { if (configTmp.getDestination().equals(task)) {
ESEtlService esEtlService = new ESEtlService(transportClient, configTmp); ESEtlService esEtlService = new ESEtlService(transportClient, configTmp);
EtlResult etlRes = esEtlService.importData(params, false); EtlResult etlRes = esEtlService.importData(params);
if (!etlRes.getSucceeded()) { if (!etlRes.getSucceeded()) {
resSuccess = false; resSuccess = false;
resultMsg.append(etlRes.getErrorMessage()).append("\n"); resultMsg.append(etlRes.getErrorMessage()).append("\n");

21
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
View File

@@ -30,7 +30,7 @@ public class ESSyncConfig {
if (esMapping._type == null) { if (esMapping._type == null) {
throw new NullPointerException("esMapping._type"); throw new NullPointerException("esMapping._type");
} }
if (esMapping._id == null && esMapping.pk == null) { if (esMapping._id == null && esMapping.getPk() == null) {
throw new NullPointerException("esMapping._id and esMapping.pk"); throw new NullPointerException("esMapping._id and esMapping.pk");
} }
if (esMapping.sql == null) { if (esMapping.sql == null) {
@@ -83,8 +83,9 @@ public class ESSyncConfig {
private String _index; private String _index;
private String _type; private String _type;
private String _id; private String _id;
private boolean upsert = false;
private String pk; private String pk;
private String parent; // private String parent;
private String sql; private String sql;
// 对象字段, 例: objFields: // 对象字段, 例: objFields:
// - _labels: array:; // - _labels: array:;
@@ -121,6 +122,14 @@ public class ESSyncConfig {
this._id = _id; this._id = _id;
} }
public boolean isUpsert() {
return upsert;
}
public void setUpsert(boolean upsert) {
this.upsert = upsert;
}
public String getPk() { public String getPk() {
return pk; return pk;
} }
@@ -129,14 +138,6 @@ public class ESSyncConfig {
this.pk = pk; this.pk = pk;
} }
public String getParent() {
return parent;
}
public void setParent(String parent) {
this.parent = parent;
}
public Map<String, String> getObjFields() { public Map<String, String> getObjFields() {
return objFields; return objFields;
} }

858
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
View File

@@ -1,422 +1,436 @@
package com.alibaba.otter.canal.client.adapter.es.config; package com.alibaba.otter.canal.client.adapter.es.config;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
/** /**
* ES 映射配置视图 * ES 映射配置视图
* *
* @author rewerma 2018-11-01 * @author rewerma 2018-11-01
* @version 1.0.0 * @version 1.0.0
*/ */
public class SchemaItem { public class SchemaItem {
private Map<String, TableItem> aliasTableItems = new LinkedHashMap<>(); // 别名对应表名 private Map<String, TableItem> aliasTableItems = new LinkedHashMap<>(); // 别名对应表名
private Map<String, FieldItem> selectFields = new LinkedHashMap<>(); // 查询字段 private Map<String, FieldItem> selectFields = new LinkedHashMap<>(); // 查询字段
private String sql; private String sql;
private volatile Map<String, List<TableItem>> tableItemAliases; private volatile Map<String, List<TableItem>> tableItemAliases;
private volatile Map<String, List<FieldItem>> columnFields; private volatile Map<String, List<FieldItem>> columnFields;
private volatile Boolean allFieldsSimple; private volatile Boolean allFieldsSimple;
public void init() { public void init() {
this.getTableItemAliases(); this.getTableItemAliases();
this.getColumnFields(); this.getColumnFields();
this.isAllFieldsSimple(); this.isAllFieldsSimple();
aliasTableItems.values().forEach(tableItem -> { aliasTableItems.values().forEach(tableItem -> {
tableItem.getRelationTableFields(); tableItem.getRelationTableFields();
tableItem.getRelationSelectFieldItems(); tableItem.getRelationSelectFieldItems();
}); });
} }
public Map<String, TableItem> getAliasTableItems() { public Map<String, TableItem> getAliasTableItems() {
return aliasTableItems; return aliasTableItems;
} }
public void setAliasTableItems(Map<String, TableItem> aliasTableItems) { public void setAliasTableItems(Map<String, TableItem> aliasTableItems) {
this.aliasTableItems = aliasTableItems; this.aliasTableItems = aliasTableItems;
} }
public String getSql() { public String getSql() {
return sql; return sql;
} }
public void setSql(String sql) { public void setSql(String sql) {
this.sql = sql; this.sql = sql;
} }
public Map<String, FieldItem> getSelectFields() { public Map<String, FieldItem> getSelectFields() {
return selectFields; return selectFields;
} }
public void setSelectFields(Map<String, FieldItem> selectFields) { public void setSelectFields(Map<String, FieldItem> selectFields) {
this.selectFields = selectFields; this.selectFields = selectFields;
} }
public Map<String, List<TableItem>> getTableItemAliases() { public String toSql() {
if (tableItemAliases == null) { // todo
synchronized (SchemaItem.class) { return null;
if (tableItemAliases == null) { }
tableItemAliases = new LinkedHashMap<>();
aliasTableItems.forEach((alias, tableItem) -> { public Map<String, List<TableItem>> getTableItemAliases() {
List<TableItem> aliases = tableItemAliases if (tableItemAliases == null) {
.computeIfAbsent(tableItem.getTableName().toLowerCase(), k -> new ArrayList<>()); synchronized (SchemaItem.class) {
aliases.add(tableItem); if (tableItemAliases == null) {
}); tableItemAliases = new LinkedHashMap<>();
} aliasTableItems.forEach((alias, tableItem) -> {
} List<TableItem> aliases = tableItemAliases
} .computeIfAbsent(tableItem.getTableName().toLowerCase(), k -> new ArrayList<>());
return tableItemAliases; aliases.add(tableItem);
} });
}
public Map<String, List<FieldItem>> getColumnFields() { }
if (columnFields == null) { }
synchronized (SchemaItem.class) { return tableItemAliases;
if (columnFields == null) { }
columnFields = new LinkedHashMap<>();
getSelectFields() public Map<String, List<FieldItem>> getColumnFields() {
.forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> { if (columnFields == null) {
// TableItem tableItem = getAliasTableItems().get(columnItem.getOwner()); synchronized (SchemaItem.class) {
// if (!tableItem.isSubQuery()) { if (columnFields == null) {
List<FieldItem> fieldItems = columnFields.computeIfAbsent( columnFields = new LinkedHashMap<>();
columnItem.getOwner() + "." + columnItem.getColumnName(), getSelectFields()
k -> new ArrayList<>()); .forEach((fieldName, fieldItem) -> fieldItem.getColumnItems().forEach(columnItem -> {
fieldItems.add(fieldItem); // TableItem tableItem = getAliasTableItems().get(columnItem.getOwner());
// } else { // if (!tableItem.isSubQuery()) {
// tableItem.getSubQueryFields().forEach(subQueryField -> { List<FieldItem> fieldItems = columnFields.computeIfAbsent(
// List<FieldItem> fieldItems = columnFields.computeIfAbsent( columnItem.getOwner() + "." + columnItem.getColumnName(),
// columnItem.getOwner() + "." + subQueryField.getColumn().getColumnName(), k -> new ArrayList<>());
// k -> new ArrayList<>()); fieldItems.add(fieldItem);
// fieldItems.add(fieldItem); // } else {
// }); // tableItem.getSubQueryFields().forEach(subQueryField -> {
// } // List<FieldItem> 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) { return columnFields;
if (allFieldsSimple == null) { }
allFieldsSimple = true;
public boolean isAllFieldsSimple() {
for (FieldItem fieldItem : getSelectFields().values()) { if (allFieldsSimple == null) {
if (fieldItem.isMethod() || fieldItem.isBinaryOp()) { synchronized (SchemaItem.class) {
allFieldsSimple = false; if (allFieldsSimple == null) {
break; 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(); return allFieldsSimple;
} else { }
return null;
} public TableItem getMainTable() {
} if (!aliasTableItems.isEmpty()) {
return aliasTableItems.values().iterator().next();
public FieldItem getIdFieldItem(ESMapping mapping) { } else {
if (mapping.get_id() != null) { return null;
return getSelectFields().get(mapping.get_id()); }
} else { }
return getSelectFields().get(mapping.getPk());
} public FieldItem getIdFieldItem(ESMapping mapping) {
} if (mapping.get_id() != null) {
return getSelectFields().get(mapping.get_id());
public static class TableItem { } else {
return getSelectFields().get(mapping.getPk());
private SchemaItem schemaItem; }
}
private String schema;
private String tableName; public static class TableItem {
private String alias;
private String subQuerySql; private SchemaItem schemaItem;
private List<FieldItem> subQueryFields = new ArrayList<>();
private List<RelationFieldsPair> relationFields = new ArrayList<>(); private String schema;
private String tableName;
private boolean main; private String alias;
private boolean subQuery; private String subQuerySql;
private List<FieldItem> subQueryFields = new ArrayList<>();
private volatile Map<FieldItem, List<FieldItem>> relationTableFields; // 当前表关联条件字段对应主表查询字段 private List<RelationFieldsPair> relationFields = new ArrayList<>();
private volatile List<FieldItem> relationSelectFieldItems; // 子表所在主表的查询字段
private boolean main;
public TableItem(SchemaItem schemaItem){ private boolean subQuery;
this.schemaItem = schemaItem;
} private volatile Map<FieldItem, List<FieldItem>> relationTableFields; // 当前表关联条件字段对应主表查询字段
private volatile List<FieldItem> relationSelectFieldItems; // 子表所在主表的查询字段
public SchemaItem getSchemaItem() {
return schemaItem; public TableItem(SchemaItem schemaItem){
} this.schemaItem = schemaItem;
}
public void setSchemaItem(SchemaItem schemaItem) {
this.schemaItem = schemaItem; public SchemaItem getSchemaItem() {
} return schemaItem;
}
public String getSchema() {
return schema; public void setSchemaItem(SchemaItem schemaItem) {
} this.schemaItem = schemaItem;
}
public void setSchema(String schema) {
this.schema = schema; public String getSchema() {
} return schema;
}
public String getTableName() {
return tableName; public void setSchema(String schema) {
} this.schema = schema;
}
public void setTableName(String tableName) {
this.tableName = tableName; public String getTableName() {
} return tableName;
}
public String getAlias() {
return alias; public void setTableName(String tableName) {
} this.tableName = tableName;
}
public void setAlias(String alias) {
this.alias = alias; public String getAlias() {
} return alias;
}
public String getSubQuerySql() {
return subQuerySql; public void setAlias(String alias) {
} this.alias = alias;
}
public void setSubQuerySql(String subQuerySql) {
this.subQuerySql = subQuerySql; public String getSubQuerySql() {
} return subQuerySql;
}
public boolean isMain() {
return main; public void setSubQuerySql(String subQuerySql) {
} this.subQuerySql = subQuerySql;
}
public void setMain(boolean main) {
this.main = main; public boolean isMain() {
} return main;
}
public boolean isSubQuery() {
return subQuery; public void setMain(boolean main) {
} this.main = main;
}
public void setSubQuery(boolean subQuery) {
this.subQuery = subQuery; public boolean isSubQuery() {
} return subQuery;
}
public List<FieldItem> getSubQueryFields() {
return subQueryFields; public void setSubQuery(boolean subQuery) {
} this.subQuery = subQuery;
}
public void setSubQueryFields(List<FieldItem> subQueryFields) {
this.subQueryFields = subQueryFields; public List<FieldItem> getSubQueryFields() {
} return subQueryFields;
}
public List<RelationFieldsPair> getRelationFields() {
return relationFields; public void setSubQueryFields(List<FieldItem> subQueryFields) {
} this.subQueryFields = subQueryFields;
}
public void setRelationFields(List<RelationFieldsPair> relationFields) {
this.relationFields = relationFields; public List<RelationFieldsPair> getRelationFields() {
} return relationFields;
}
public Map<FieldItem, List<FieldItem>> getRelationTableFields() {
if (relationTableFields == null) { public void setRelationFields(List<RelationFieldsPair> relationFields) {
synchronized (SchemaItem.class) { this.relationFields = relationFields;
if (relationTableFields == null) { }
relationTableFields = new LinkedHashMap<>();
public Map<FieldItem, List<FieldItem>> getRelationTableFields() {
getRelationFields().forEach(relationFieldsPair -> { if (relationTableFields == null) {
FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem(); synchronized (SchemaItem.class) {
FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem(); if (relationTableFields == null) {
FieldItem currentTableRelField = null; relationTableFields = new LinkedHashMap<>();
if (getAlias().equals(leftFieldItem.getOwner())) {
currentTableRelField = leftFieldItem; getRelationFields().forEach(relationFieldsPair -> {
} else if (getAlias().equals(rightFieldItem.getOwner())) { FieldItem leftFieldItem = relationFieldsPair.getLeftFieldItem();
currentTableRelField = rightFieldItem; FieldItem rightFieldItem = relationFieldsPair.getRightFieldItem();
} FieldItem currentTableRelField = null;
if (getAlias().equals(leftFieldItem.getOwner())) {
if (currentTableRelField != null) { currentTableRelField = leftFieldItem;
List<FieldItem> selectFieldItem = getSchemaItem().getColumnFields() } else if (getAlias().equals(rightFieldItem.getOwner())) {
.get(leftFieldItem.getOwner() + "." + leftFieldItem.getColumn().getColumnName()); currentTableRelField = rightFieldItem;
if (selectFieldItem != null && !selectFieldItem.isEmpty()) { }
relationTableFields.put(currentTableRelField, selectFieldItem);
} else { if (currentTableRelField != null) {
selectFieldItem = getSchemaItem().getColumnFields() List<FieldItem> selectFieldItem = getSchemaItem().getColumnFields()
.get(rightFieldItem.getOwner() + "." .get(leftFieldItem.getOwner() + "." + leftFieldItem.getColumn().getColumnName());
+ rightFieldItem.getColumn().getColumnName()); if (selectFieldItem != null && !selectFieldItem.isEmpty()) {
if (selectFieldItem != null && !selectFieldItem.isEmpty()) { relationTableFields.put(currentTableRelField, selectFieldItem);
relationTableFields.put(currentTableRelField, selectFieldItem); } else {
} else { selectFieldItem = getSchemaItem().getColumnFields()
throw new UnsupportedOperationException( .get(rightFieldItem.getOwner() + "."
"Relation condition column must in select columns."); + 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<FieldItem> getRelationSelectFieldItems() { }
if (relationSelectFieldItems == null) { }
synchronized (SchemaItem.class) { return relationTableFields;
if (relationSelectFieldItems == null) { }
relationSelectFieldItems = new ArrayList<>();
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { public List<FieldItem> getRelationSelectFieldItems() {
if (fieldItem.getOwners().contains(getAlias())) { if (relationSelectFieldItems == null) {
relationSelectFieldItems.add(fieldItem); 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 { }
return relationSelectFieldItems;
private FieldItem leftFieldItem; }
private FieldItem rightFieldItem; }
public RelationFieldsPair(FieldItem leftFieldItem, FieldItem rightFieldItem){ public static class RelationFieldsPair {
this.leftFieldItem = leftFieldItem;
this.rightFieldItem = rightFieldItem; private FieldItem leftFieldItem;
} private FieldItem rightFieldItem;
public FieldItem getLeftFieldItem() { public RelationFieldsPair(FieldItem leftFieldItem, FieldItem rightFieldItem){
return leftFieldItem; this.leftFieldItem = leftFieldItem;
} this.rightFieldItem = rightFieldItem;
}
public void setLeftFieldItem(FieldItem leftFieldItem) {
this.leftFieldItem = leftFieldItem; public FieldItem getLeftFieldItem() {
} return leftFieldItem;
}
public FieldItem getRightFieldItem() {
return rightFieldItem; public void setLeftFieldItem(FieldItem leftFieldItem) {
} this.leftFieldItem = leftFieldItem;
}
public void setRightFieldItem(FieldItem rightFieldItem) {
this.rightFieldItem = rightFieldItem; public FieldItem getRightFieldItem() {
} return rightFieldItem;
} }
public static class FieldItem { public void setRightFieldItem(FieldItem rightFieldItem) {
this.rightFieldItem = rightFieldItem;
private String fieldName; }
private List<ColumnItem> columnItems = new ArrayList<>(); }
private List<String> owners = new ArrayList<>();
public static class FieldItem {
private boolean method;
private boolean binaryOp; private String fieldName;
private String expr;
public String getFieldName() { private List<ColumnItem> columnItems = new ArrayList<>();
return fieldName; private List<String> owners = new ArrayList<>();
}
private boolean method;
public void setFieldName(String fieldName) { private boolean binaryOp;
this.fieldName = fieldName;
} public String getFieldName() {
return fieldName;
public List<ColumnItem> getColumnItems() { }
return columnItems;
} public void setFieldName(String fieldName) {
this.fieldName = fieldName;
public void setColumnItems(List<ColumnItem> columnItems) { }
this.columnItems = columnItems;
} public String getExpr() {
return expr;
public boolean isMethod() { }
return method;
} public void setExpr(String expr) {
this.expr = expr;
public void setMethod(boolean method) { }
this.method = method;
} public List<ColumnItem> getColumnItems() {
return columnItems;
public boolean isBinaryOp() { }
return binaryOp;
} public void setColumnItems(List<ColumnItem> columnItems) {
this.columnItems = columnItems;
public void setBinaryOp(boolean binaryOp) { }
this.binaryOp = binaryOp;
} public boolean isMethod() {
return method;
public List<String> getOwners() { }
return owners;
} public void setMethod(boolean method) {
this.method = method;
public void setOwners(List<String> owners) { }
this.owners = owners;
} public boolean isBinaryOp() {
return binaryOp;
public void addColumn(ColumnItem columnItem) { }
columnItems.add(columnItem);
} public void setBinaryOp(boolean binaryOp) {
this.binaryOp = binaryOp;
public ColumnItem getColumn() { }
if (!columnItems.isEmpty()) {
return columnItems.get(0); public List<String> getOwners() {
} else { return owners;
return null; }
}
} public void setOwners(List<String> owners) {
this.owners = owners;
public String getOwner() { }
if (!owners.isEmpty()) {
return owners.get(0); public void addColumn(ColumnItem columnItem) {
} else { columnItems.add(columnItem);
return null; }
}
} public ColumnItem getColumn() {
if (!columnItems.isEmpty()) {
@Override return columnItems.get(0);
public boolean equals(Object o) { } else {
if (this == o) return true; return null;
if (o == null || getClass() != o.getClass()) return false; }
}
FieldItem fieldItem = (FieldItem) o;
public String getOwner() {
return fieldName != null ? fieldName.equals(fieldItem.fieldName) : fieldItem.fieldName == null; if (!owners.isEmpty()) {
} return owners.get(0);
} else {
@Override return null;
public int hashCode() { }
return fieldName != null ? fieldName.hashCode() : 0; }
}
} @Override
public boolean equals(Object o) {
public static class ColumnItem { if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
private String owner;
private String columnName; FieldItem fieldItem = (FieldItem) o;
public String getOwner() { return fieldName != null ? fieldName.equals(fieldItem.fieldName) : fieldItem.fieldName == null;
return owner; }
}
@Override
public void setOwner(String owner) { public int hashCode() {
this.owner = owner; return fieldName != null ? fieldName.hashCode() : 0;
} }
}
public String getColumnName() {
return columnName; public static class ColumnItem {
}
private String owner;
public void setColumnName(String columnName) { private String columnName;
this.columnName = 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;
}
}
}

15
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SqlParser.java
View File

@@ -25,7 +25,7 @@ import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.TableItem;
/** /**
* ES同步指定sql格式解析 * ES同步指定sql格式解析
* *
* @author rewerma 2018-10-26 下午03:45:49 * @author rewerma 2018-10-26 下午03:45:49
* @version 1.0.0 * @version 1.0.0
*/ */
@@ -33,7 +33,7 @@ public class SqlParser {
/** /**
* 解析sql * 解析sql
* *
* @param sql sql * @param sql sql
* @return 视图对象 * @return 视图对象
*/ */
@@ -66,7 +66,7 @@ public class SqlParser {
/** /**
* 归集字段 * 归集字段
* *
* @param sqlSelectQueryBlock sqlSelectQueryBlock * @param sqlSelectQueryBlock sqlSelectQueryBlock
* @return 字段属性列表 * @return 字段属性列表
*/ */
@@ -74,6 +74,7 @@ public class SqlParser {
return sqlSelectQueryBlock.getSelectList().stream().map(selectItem -> { return sqlSelectQueryBlock.getSelectList().stream().map(selectItem -> {
FieldItem fieldItem = new FieldItem(); FieldItem fieldItem = new FieldItem();
fieldItem.setFieldName(selectItem.getAlias()); fieldItem.setFieldName(selectItem.getAlias());
fieldItem.setExpr(selectItem.toString());
visitColumn(selectItem.getExpr(), fieldItem); visitColumn(selectItem.getExpr(), fieldItem);
return fieldItem; return fieldItem;
}).collect(Collectors.toList()); }).collect(Collectors.toList());
@@ -81,7 +82,7 @@ public class SqlParser {
/** /**
* 解析字段 * 解析字段
* *
* @param expr sql expr * @param expr sql expr
* @param fieldItem 字段属性 * @param fieldItem 字段属性
*/ */
@@ -91,6 +92,7 @@ public class SqlParser {
SQLIdentifierExpr identifierExpr = (SQLIdentifierExpr) expr; SQLIdentifierExpr identifierExpr = (SQLIdentifierExpr) expr;
if (fieldItem.getFieldName() == null) { if (fieldItem.getFieldName() == null) {
fieldItem.setFieldName(identifierExpr.getName()); fieldItem.setFieldName(identifierExpr.getName());
fieldItem.setExpr(identifierExpr.toString());
} }
ColumnItem columnItem = new ColumnItem(); ColumnItem columnItem = new ColumnItem();
columnItem.setColumnName(identifierExpr.getName()); columnItem.setColumnName(identifierExpr.getName());
@@ -101,6 +103,7 @@ public class SqlParser {
SQLPropertyExpr sqlPropertyExpr = (SQLPropertyExpr) expr; SQLPropertyExpr sqlPropertyExpr = (SQLPropertyExpr) expr;
if (fieldItem.getFieldName() == null) { if (fieldItem.getFieldName() == null) {
fieldItem.setFieldName(sqlPropertyExpr.getName()); fieldItem.setFieldName(sqlPropertyExpr.getName());
fieldItem.setExpr(sqlPropertyExpr.toString());
} }
fieldItem.getOwners().add(sqlPropertyExpr.getOwnernName()); fieldItem.getOwners().add(sqlPropertyExpr.getOwnernName());
ColumnItem columnItem = new ColumnItem(); ColumnItem columnItem = new ColumnItem();
@@ -123,7 +126,7 @@ public class SqlParser {
/** /**
* 解析表 * 解析表
* *
* @param schemaItem 视图对象 * @param schemaItem 视图对象
* @param sqlTableSource sqlTableSource * @param sqlTableSource sqlTableSource
* @param tableItems 表对象列表 * @param tableItems 表对象列表
@@ -178,7 +181,7 @@ public class SqlParser {
/** /**
* 解析on条件 * 解析on条件
* *
* @param expr sql expr * @param expr sql expr
* @param tableItem 表对象 * @param tableItem 表对象
*/ */

142
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java
View File

@@ -1,6 +1,9 @@
package com.alibaba.otter.canal.client.adapter.es.service; package com.alibaba.otter.canal.client.adapter.es.service;
import java.util.*; import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.Future;
@@ -14,7 +17,9 @@ import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHit;
@@ -51,7 +56,7 @@ public class ESEtlService {
this.config = config; this.config = config;
} }
public EtlResult importData(List<String> params, boolean bulk) { public EtlResult importData(List<String> params) {
EtlResult etlResult = new EtlResult(); EtlResult etlResult = new EtlResult();
AtomicLong impCount = new AtomicLong(); AtomicLong impCount = new AtomicLong();
List<String> errMsg = new ArrayList<>(); List<String> errMsg = new ArrayList<>();
@@ -93,54 +98,49 @@ public class ESEtlService {
logger.debug("etl sql : {}", mapping.getSql()); logger.debug("etl sql : {}", mapping.getSql());
} }
if (bulk) { // 获取总数
// 获取总数 String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> {
long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> { Long count = null;
Long count = null; try {
try { if (rs.next()) {
if (rs.next()) { count = ((Number) rs.getObject(1)).longValue();
count = ((Number) rs.getObject(1)).longValue();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} }
return count == null ? 0L : count; } catch (Exception e) {
}); logger.error(e.getMessage(), e);
// 当大于1万条记录时开启多线程
if (cnt >= 10000) {
int threadCount = 3; // TODO 从配置读取默认为3
long perThreadCnt = cnt / threadCount;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
List<Future<Boolean>> 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<Boolean> future = executor
.submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
futures.add(future);
}
for (Future<Boolean> future : futures) {
future.get();
}
executor.shutdown();
} else {
executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
} }
return count == null ? 0L : count;
});
// 当大于1万条记录时开启多线程
if (cnt >= 10000) {
int threadCount = 3; // 从配置读取默认为3
long perThreadCnt = cnt / threadCount;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
List<Future<Boolean>> 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<Boolean> future = executor
.submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
futures.add(future);
}
for (Future<Boolean> future : futures) {
future.get();
}
executor.shutdown();
} else { } else {
logger.info("自动ETL，无需统计记录总条数，直接进行ETL, index: {}", esIndex);
executeSqlImport(dataSource, sql, mapping, impCount, errMsg); executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
} }
@@ -158,7 +158,7 @@ public class ESEtlService {
return etlResult; return etlResult;
} }
private void processFailBulkResponse(BulkResponse bulkResponse, boolean hasParent) { private void processFailBulkResponse(BulkResponse bulkResponse) {
for (BulkItemResponse response : bulkResponse.getItems()) { for (BulkItemResponse response : bulkResponse.getItems()) {
if (!response.isFailed()) { if (!response.isFailed()) {
continue; continue;
@@ -205,31 +205,27 @@ public class ESEtlService {
} }
if (idVal != null) { if (idVal != null) {
if (mapping.getParent() == null) { if (mapping.isUpsert()) {
bulkRequestBuilder.add(transportClient
.prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString())
.setDoc(esFieldData)
.setDocAsUpsert(true));
} else {
bulkRequestBuilder.add(transportClient bulkRequestBuilder.add(transportClient
.prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString()) .prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString())
.setSource(esFieldData)); .setSource(esFieldData));
} else {
// ignore
} }
} else { } else {
idVal = rs.getObject(mapping.getPk()); idVal = rs.getObject(mapping.getPk());
if (mapping.getParent() == null) { SearchResponse response = transportClient.prepareSearch(mapping.get_index())
// 删除pk对应的数据 .setTypes(mapping.get_type())
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
.setTypes(mapping.get_type()) .setSize(10000)
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) .get();
.get(); for (SearchHit hit : response.getHits()) {
for (SearchHit hit : response.getHits()) { bulkRequestBuilder.add(
bulkRequestBuilder.add(transportClient transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); .setDoc(esFieldData));
}
bulkRequestBuilder
.add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type())
.setSource(esFieldData));
} else {
// ignore
} }
} }
@@ -238,11 +234,11 @@ public class ESEtlService {
long esBatchBegin = System.currentTimeMillis(); long esBatchBegin = System.currentTimeMillis();
BulkResponse rp = bulkRequestBuilder.execute().actionGet(); BulkResponse rp = bulkRequestBuilder.execute().actionGet();
if (rp.hasFailures()) { if (rp.hasFailures()) {
this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent())); this.processFailBulkResponse(rp);
} }
if (logger.isDebugEnabled()) { if (logger.isTraceEnabled()) {
logger.debug("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
bulkRequestBuilder.numberOfActions(), bulkRequestBuilder.numberOfActions(),
@@ -259,10 +255,10 @@ public class ESEtlService {
long esBatchBegin = System.currentTimeMillis(); long esBatchBegin = System.currentTimeMillis();
BulkResponse rp = bulkRequestBuilder.execute().actionGet(); BulkResponse rp = bulkRequestBuilder.execute().actionGet();
if (rp.hasFailures()) { if (rp.hasFailures()) {
this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent())); this.processFailBulkResponse(rp);
} }
if (logger.isDebugEnabled()) { if (logger.isTraceEnabled()) {
logger.debug("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
bulkRequestBuilder.numberOfActions(), bulkRequestBuilder.numberOfActions(),

162
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
View File

@@ -71,7 +71,12 @@ public class ESSyncService {
dml.getDestination()); dml.getDestination());
} }
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); StringBuilder configIndexes = new StringBuilder();
esSyncConfigs
.forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
logger.debug("DML: {} \nEffect indexes: {}",
JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue),
configIndexes.toString());
} }
} }
} }
@@ -92,6 +97,8 @@ public class ESSyncService {
update(config, dml); update(config, dml);
} else if (type != null && type.equalsIgnoreCase("DELETE")) { } else if (type != null && type.equalsIgnoreCase("DELETE")) {
delete(config, dml); delete(config, dml);
} else {
return;
} }
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
@@ -329,33 +336,50 @@ public class ESSyncService {
// ------是主表------ // ------是主表------
if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping); if (mapping.get_id() != null) {
// 主键为简单字段 FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping);
if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) { // 主键为简单字段
Object idVal = esTemplate.getValFromData(mapping, if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) {
data, Object idVal = esTemplate.getValFromData(mapping,
idFieldItem.getFieldName(), data,
idFieldItem.getColumn().getColumnName()); idFieldItem.getFieldName(),
idFieldItem.getColumn().getColumnName());
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, id: {}", logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.delete(mapping, idVal); esTemplate.delete(mapping, idVal, null);
if (!result) { } else {
logger.error("Main table delete es index error, destination:{}, table: {}, index: {}, id: {}", // ------主键带函数, 查询sql获取主键删除------
config.getDestination(), // FIXME 删除时反查sql为空记录, 无法获获取 id field 值
dml.getTable(), mainTableDelete(config, dml, data);
mapping.get_index(),
idVal);
} }
} else { } else {
// ------主键带函数, 查询sql获取主键删除------ FieldItem pkFieldItem = schemaItem.getIdFieldItem(mapping);
mainTableDelete(config, dml, data); if (!pkFieldItem.isMethod() && !pkFieldItem.isBinaryOp()) {
Map<String, Object> esFieldData = new LinkedHashMap<>();
Object pkVal = esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
if (logger.isTraceEnabled()) {
logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, pk: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
pkVal);
}
esFieldData.remove(pkFieldItem.getFieldName());
esFieldData.keySet().forEach(key -> esFieldData.put(key, null));
esTemplate.delete(mapping, pkVal, esFieldData);
} else {
// ------主键带函数, 查询sql获取主键删除------
mainTableDelete(config, dml, data);
}
} }
} }
// 从表的操作 // 从表的操作
@@ -417,14 +441,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.insert(mapping, idVal, esFieldData); esTemplate.insert(mapping, idVal, esFieldData);
if (!result) {
logger.error("Single table insert to es index error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
/** /**
@@ -461,15 +478,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.insert(mapping, idVal, esFieldData); esTemplate.insert(mapping, idVal, esFieldData);
if (!result) {
logger.error(
"Main table insert to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -493,6 +502,15 @@ public class ESSyncService {
} }
ESSyncUtil.sqlRS(ds, sql, rs -> { ESSyncUtil.sqlRS(ds, sql, rs -> {
try { try {
Map<String, Object> esFieldData = null;
if (mapping.getPk() != null) {
esFieldData = new LinkedHashMap<>();
esTemplate.getESDataFromDmlData(mapping, data, esFieldData);
esFieldData.remove(mapping.getPk());
for (String key : esFieldData.keySet()) {
esFieldData.put(key, null);
}
}
while (rs.next()) { while (rs.next()) {
Object idVal = esTemplate.getIdValFromRS(mapping, rs); Object idVal = esTemplate.getIdValFromRS(mapping, rs);
@@ -504,15 +522,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.delete(mapping, idVal); esTemplate.delete(mapping, idVal, esFieldData);
if (!result) {
logger.error(
"Main table delete to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -558,13 +568,7 @@ public class ESSyncService {
dml.getTable(), dml.getTable(),
mapping.get_index()); mapping.get_index());
} }
boolean result = esTemplate.updateByQuery(config, paramsTmp, esFieldData); esTemplate.updateByQuery(config, paramsTmp, esFieldData);
if (!result) {
logger.error("Join table update es index by foreign key error, destination:{}, table: {}, index: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index());
}
} }
/** /**
@@ -652,14 +656,7 @@ public class ESSyncService {
dml.getTable(), dml.getTable(),
mapping.get_index()); mapping.get_index());
} }
boolean result = esTemplate.updateByQuery(config, paramsTmp, esFieldData); esTemplate.updateByQuery(config, paramsTmp, esFieldData);
if (!result) {
logger.error(
"Join table update es index by query sql error, destination:{}, table: {}, index: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index());
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -761,14 +758,7 @@ public class ESSyncService {
dml.getTable(), dml.getTable(),
mapping.get_index()); mapping.get_index());
} }
boolean result = esTemplate.updateByQuery(config, paramsTmp, esFieldData); esTemplate.updateByQuery(config, paramsTmp, esFieldData);
if (!result) {
logger.error(
"Join table update es index by query whole sql error, destination:{}, table: {}, index: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index());
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -799,14 +789,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.update(mapping, idVal, esFieldData); esTemplate.update(mapping, idVal, esFieldData);
if (!result) {
logger.error("Main table update to es index error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
/** /**
@@ -843,15 +826,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.update(mapping, idVal, esFieldData); esTemplate.update(mapping, idVal, esFieldData);
if (!result) {
logger.error(
"Main table update to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -859,4 +834,11 @@ public class ESSyncService {
return 0; return 0;
}); });
} }
/**
* 提交批次
*/
public void commit() {
esTemplate.commit();
}
} }

312
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
View File

@@ -2,14 +2,13 @@ package com.alibaba.otter.canal.client.adapter.es.support;
import java.sql.ResultSet; import java.sql.ResultSet;
import java.sql.SQLException; import java.sql.SQLException;
import java.util.*; import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import javax.sql.DataSource; import javax.sql.DataSource;
import com.alibaba.fastjson.JSON;
import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.bulk.BulkResponse;
@@ -18,18 +17,11 @@ import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.MappingMetaData; import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryAction;
import org.elasticsearch.index.reindex.UpdateByQueryRequestBuilder;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHit;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
@@ -52,254 +44,135 @@ public class ESTemplate {
private TransportClient transportClient; private TransportClient transportClient;
private BulkRequestBuilder bulkRequestBuilder;
public ESTemplate(TransportClient transportClient){ public ESTemplate(TransportClient transportClient){
this.transportClient = transportClient; this.transportClient = transportClient;
this.bulkRequestBuilder = transportClient.prepareBulk();
}
public BulkRequestBuilder getBulk() {
return bulkRequestBuilder;
} }
/** /**
* 插入数据 * 插入数据
* *
* @param mapping * @param mapping 配置对象
* @param pkVal * @param pkVal 主键值
* @param esFieldData * @param esFieldData 数据Map
* @return
*/ */
public boolean insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
bulkRequestBuilder if (mapping.isUpsert()) {
.add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData)
.setDocAsUpsert(true));
} else {
getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setSource(esFieldData)); .setSource(esFieldData));
}
commitBulk();
} else { } else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type()) .setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.setSize(MAX_BATCH_SIZE) .setSize(10000)
.get(); .get();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
bulkRequestBuilder getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
.add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); .setDoc(esFieldData));
commitBulk();
} }
bulkRequestBuilder
.add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()).setSource(esFieldData));
} }
return commitBulkRequest(bulkRequestBuilder);
} }
/** /**
* 根据主键更新数据 * 根据主键更新数据
* *
* @param mapping * @param mapping 配置对象
* @param pkVal * @param pkVal 主键值
* @param esFieldData * @param esFieldData 数据Map
* @return
*/ */
public boolean update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); append4Update(mapping, pkVal, esFieldData);
append4Update(bulkRequestBuilder, mapping, pkVal, esFieldData); commitBulk();
return commitBulkRequest(bulkRequestBuilder);
}
public void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal,
Map<String, Object> esFieldData) {
if (mapping.get_id() != null) {
bulkRequestBuilder
.add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData));
} else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.setSize(MAX_BATCH_SIZE)
.get();
for (SearchHit hit : response.getHits()) {
bulkRequestBuilder
.add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
.setDoc(esFieldData));
}
}
} }
/** /**
* update by query * update by query
* *
* @param config * @param config 配置对象
* @param paramsTmp * @param paramsTmp sql查询条件
* @param esFieldData * @param esFieldData 数据Map
* @return
*/ */
public boolean updateByQuery(ESSyncConfig config, Map<String, Object> paramsTmp, Map<String, Object> esFieldData) { public void updateByQuery(ESSyncConfig config, Map<String, Object> paramsTmp, Map<String, Object> esFieldData) {
if (paramsTmp.isEmpty()) { if (paramsTmp.isEmpty()) {
return false; return;
} }
ESMapping mapping = config.getEsMapping(); ESMapping mapping = config.getEsMapping();
BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value))); paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value)));
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) // 查询sql批量更新
.setTypes(mapping.get_type()) DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
.setSize(0) StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE ");
.setQuery(queryBuilder) paramsTmp.forEach(
.get(); (fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND "));
long count = response.getHits().getTotalHits(); int len = sql.length();
// 如果更新量大于Max, 查询sql批量更新 sql.delete(len - 4, len);
if (count > MAX_BATCH_SIZE) { Integer syncCount = (Integer) ESSyncUtil.sqlRS(ds, sql.toString(), rs -> {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); int count = 0;
DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
// 查询sql更新
StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE ");
paramsTmp.forEach(
(fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND "));
int len = sql.length();
sql.delete(len - 4, len);
ESSyncUtil.sqlRS(ds, sql.toString(), rs -> {
int exeCount = 1;
try {
BulkRequestBuilder bulkRequestBuilderTmp = bulkRequestBuilder;
while (rs.next()) {
Object idVal = getIdValFromRS(mapping, rs);
append4Update(bulkRequestBuilderTmp, mapping, idVal, esFieldData);
if (exeCount % mapping.getCommitBatch() == 0 && bulkRequestBuilderTmp.numberOfActions() > 0) {
commitBulkRequest(bulkRequestBuilderTmp);
bulkRequestBuilderTmp = transportClient.prepareBulk();
}
exeCount++;
}
if (bulkRequestBuilder.numberOfActions() > 0) {
commitBulkRequest(bulkRequestBuilderTmp);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return 0;
});
return true;
} else {
return updateByQuery(mapping, queryBuilder, esFieldData, 1);
}
}
private boolean updateByQuery(ESMapping mapping, QueryBuilder queryBuilder, Map<String, Object> esFieldData,
int counter) {
if (CollectionUtils.isEmpty(esFieldData)) {
return true;
}
StringBuilder sb = new StringBuilder();
esFieldData.forEach((key, value) -> {
if (value instanceof Map) {
Map<?, ?> mapValue = (Map<?, ?>) value;
if (mapValue.containsKey("lon") && mapValue.containsKey("lat") && mapValue.size() == 2) {
sb.append("ctx._source")
.append("['")
.append(key)
.append("']")
.append(" = [")
.append(mapValue.get("lon"))
.append(", ")
.append(mapValue.get("lat"))
.append("];");
} else {
sb.append("ctx._source").append("[\"").append(key).append("\"]").append(" = ");
sb.append(JSON.toJSONString(value));
sb.append(";");
}
} else if (value instanceof List) {
sb.append("ctx._source").append("[\"").append(key).append("\"]").append(" = ");
sb.append(JSON.toJSONString(value));
sb.append(";");
} else if (value instanceof String) {
sb.append("ctx._source")
.append("['")
.append(key)
.append("']")
.append(" = '")
.append(value)
.append("';");
} else {
sb.append("ctx._source").append("['").append(key).append("']").append(" = ").append(value).append(";");
}
});
String scriptLine = sb.toString();
if (logger.isTraceEnabled()) {
logger.trace(scriptLine);
}
UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(transportClient);
updateByQuery.source(mapping.get_index())
.abortOnVersionConflict(false)
.filter(queryBuilder)
.script(new Script(ScriptType.INLINE, "painless", scriptLine, Collections.emptyMap()));
BulkByScrollResponse response = updateByQuery.get();
if (logger.isTraceEnabled()) {
logger.trace("updateByQuery response: {}", response.getStatus());
}
if (!CollectionUtils.isEmpty(response.getSearchFailures())) {
logger.error("script update_for_search has search error: " + response.getBulkFailures());
return false;
}
if (!CollectionUtils.isEmpty(response.getBulkFailures())) {
logger.error("script update_for_search has update error: " + response.getBulkFailures());
return false;
}
if (response.getStatus().getVersionConflicts() > 0) {
if (counter >= 3) {
logger.error("第 {} 次执行updateByQuery, 依旧存在分片版本冲突，不再继续重试。", counter);
return false;
}
logger.warn("本次updateByQuery存在分片版本冲突，准备重新执行...");
try { try {
TimeUnit.SECONDS.sleep(1); while (rs.next()) {
} catch (InterruptedException e) { Object idVal = getIdValFromRS(mapping, rs);
// ignore append4Update(mapping, idVal, esFieldData);
commitBulk();
count++;
}
} catch (Exception e) {
throw new RuntimeException(e);
} }
return updateByQuery(mapping, queryBuilder, esFieldData, ++counter); return count;
});
if (logger.isTraceEnabled()) {
logger.trace("Update ES by query effect {} records", syncCount);
} }
return true;
} }
/** /**
* 通过主键删除数据 * 通过主键删除数据
* *
* @param mapping * @param mapping 配置对象
* @param pkVal * @param pkVal 主键值
* @return * @param esFieldData 数据Map
*/ */
public boolean delete(ESMapping mapping, Object pkVal) { public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
bulkRequestBuilder getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString()));
.add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); commitBulk();
} else { } else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type()) .setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.setSize(MAX_BATCH_SIZE) .setSize(10000)
.get(); .get();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
bulkRequestBuilder getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
.add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); .setDoc(esFieldData));
commitBulk();
} }
} }
return commitBulkRequest(bulkRequestBuilder);
} }
/** /**
* 批量提交 * 提交批次
*
* @param bulkRequestBuilder
* @return
*/ */
private static boolean commitBulkRequest(BulkRequestBuilder bulkRequestBuilder) { public void commit() {
if (bulkRequestBuilder.numberOfActions() > 0) { if (getBulk().numberOfActions() > 0) {
BulkResponse response = bulkRequestBuilder.execute().actionGet(); BulkResponse response = getBulk().execute().actionGet();
if (response.hasFailures()) { if (response.hasFailures()) {
for (BulkItemResponse itemResponse : response.getItems()) { for (BulkItemResponse itemResponse : response.getItems()) {
if (!itemResponse.isFailed()) { if (!itemResponse.isFailed()) {
@@ -307,16 +180,45 @@ public class ESTemplate {
} }
if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) { if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
logger.warn(itemResponse.getFailureMessage()); logger.error(itemResponse.getFailureMessage());
} else { } else {
logger.error("ES sync commit error: {}", itemResponse.getFailureMessage()); throw new RuntimeException("ES sync commit error" + itemResponse.getFailureMessage());
} }
} }
} }
return !response.hasFailures();
} }
return true; }
/**
* 如果大于批量数则提交批次
*/
private void commitBulk() {
if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {
commit();
}
}
private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) {
if (mapping.isUpsert()) {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData)
.setDocAsUpsert(true));
} else {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData));
}
} else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.setSize(10000)
.get();
for (SearchHit hit : response.getHits()) {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
.setDoc(esFieldData));
}
}
} }
public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName, public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,

1
client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml
View File

@@ -5,6 +5,7 @@ esMapping:
_index: mytest_user _index: mytest_user
_type: _doc _type: _doc
_id: _id _id: _id
upsert: true
# pk: id # pk: id
sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name, sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,
a.c_time as _c_time from user a a.c_time as _c_time from user a

2
client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/SqlParseTest.java
View File

@@ -29,7 +29,7 @@ public class SqlParseTest {
Assert.assertFalse(tableItem.isMain()); Assert.assertFalse(tableItem.isMain());
Assert.assertTrue(tableItem.isSubQuery()); Assert.assertTrue(tableItem.isSubQuery());
// 通过字段名找 FieldItem // 通过字段名找 FieldItem
List<FieldItem> fieldItems = schemaItem.getColumnFields().get(tableItem.getAlias() + ".label".toLowerCase()); List<FieldItem> fieldItems = schemaItem.getColumnFields().get(tableItem.getAlias() + ".labels".toLowerCase());
fieldItems.forEach( fieldItems.forEach(
fieldItem -> Assert.assertEquals("c.labels", fieldItem.getOwner() + "." + fieldItem.getFieldName())); fieldItem -> Assert.assertEquals("c.labels", fieldItem.getOwner() + "." + fieldItem.getFieldName()));

19
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java
View File

@@ -52,16 +52,17 @@ public class BatchExecutor implements Closeable {
} }
public void execute(String sql, List<Map<String, ?>> values) throws SQLException { public void execute(String sql, List<Map<String, ?>> values) throws SQLException {
PreparedStatement pstmt = getConn().prepareStatement(sql); try (PreparedStatement pstmt = getConn().prepareStatement(sql)) {
int len = values.size(); int len = values.size();
for (int i = 0; i < len; i++) { for (int i = 0; i < len; i++) {
int type = (Integer) values.get(i).get("type"); int type = (Integer) values.get(i).get("type");
Object value = values.get(i).get("value"); Object value = values.get(i).get("value");
SyncUtil.setPStmt(type, pstmt, value, i + 1); SyncUtil.setPStmt(type, pstmt, value, i + 1);
} }
pstmt.execute(); pstmt.execute();
idx.incrementAndGet(); idx.incrementAndGet();
}
} }
public void commit() throws SQLException { public void commit() throws SQLException {