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

Fix Configuration property name '-id' is not valid on canal 1.1.7 alpha 3 #4879 (#4887)

Browse Source 
* Fix Configuration property name '-id' is not valid

* Fix Configuration property name '-id' is not valid
读取原配置文件
This commit is contained in:
cooper lee
2023-09-21 10:53:47 +08:00
committed by GitHub
parent f9f810226d
commit cda330e49e
15 changed files with 198 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
client-adapter/common/pom.xml
View File

@@ -54,6 +54,11 @@
<artifactId>junit</artifactId> <artifactId>junit</artifactId>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
<scope>test</scope>
</dependency>
<dependency> <dependency>
<groupId>com.diffblue</groupId> <groupId>com.diffblue</groupId>
<artifactId>deeptestutils</artifactId> <artifactId>deeptestutils</artifactId>

22
client-adapter/common/src/test/java/com/alibaba/otter/canal/client/adapter/support/YamlUtilsTest.java
View File

@@ -1,14 +1,17 @@
package com.alibaba.otter.canal.client.adapter.support; package com.alibaba.otter.canal.client.adapter.support;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Value;
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.junit.Assert;
import org.junit.Test;
public class YamlUtilsTest { public class YamlUtilsTest {
@Test @Test
public void testLoadConfigToYml() { public void testLoadConfigToYml() {
String configStr="dataSourceKey: defaultDS\n" String configStr="dataSourceKey: defaultDS\n"
@@ -17,6 +20,7 @@ public class YamlUtilsTest {
+ "outerAdapterKey: mysql1\n" + "outerAdapterKey: mysql1\n"
+ "concurrent: true\n" + "concurrent: true\n"
+ "dbMapping:\n" + "dbMapping:\n"
+ " _id: _id\n"
+ " database: mytest\n" + " database: mytest\n"
+ " table: user\n" + " table: user\n"
+ " targetTable: mytest2.user\n" + " targetTable: mytest2.user\n"
@@ -35,6 +39,7 @@ public class YamlUtilsTest {
MappingConfig config = YamlUtils.ymlToObj(null, configStr, MappingConfig.class, null, new Properties()); MappingConfig config = YamlUtils.ymlToObj(null, configStr, MappingConfig.class, null, new Properties());
Assert.assertNotNull(config); Assert.assertNotNull(config);
Assert.assertEquals(config.getDbMapping().getId(), "_id");
Assert.assertEquals(config.getDestination(), "example"); Assert.assertEquals(config.getDestination(), "example");
Assert.assertEquals(config.getOuterAdapterKey(), "mysql1"); Assert.assertEquals(config.getOuterAdapterKey(), "mysql1");
Assert.assertEquals(config.getDbMapping().getDatabase(), "mytest"); Assert.assertEquals(config.getDbMapping().getDatabase(), "mytest");
@@ -104,6 +109,9 @@ public class YamlUtilsTest {
} }
private static class DbMapping { private static class DbMapping {
@Value("${_id}")
private String id ;
private boolean mirrorDb = false; // 是否镜像库 private boolean mirrorDb = false; // 是否镜像库
private String database; // 数据库名或schema名 private String database; // 数据库名或schema名
private String table; // 表名 private String table; // 表名
@@ -225,5 +233,13 @@ public class YamlUtilsTest {
public void setAllMapColumns(Map<String, String> allMapColumns) { public void setAllMapColumns(Map<String, String> allMapColumns) {
this.allMapColumns = allMapColumns; this.allMapColumns = allMapColumns;
} }
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
} }
} }

4
client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/ES6xAdapter.java
View File

@@ -59,12 +59,12 @@ public class ES6xAdapter extends ESAdapter {
public Map<String, Object> count(String task) { public Map<String, Object> count(String task) {
ESSyncConfig config = esSyncConfig.get(task); ESSyncConfig config = esSyncConfig.get(task);
ESSyncConfig.ESMapping mapping = config.getEsMapping(); ESSyncConfig.ESMapping mapping = config.getEsMapping();
SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index(), mapping.get_type()).size(0) SearchResponse response = this.esConnection.new ESSearchRequest(mapping.getIndex(), mapping.getType()).size(0)
.getResponse(); .getResponse();
long rowCount = response.getHits().getTotalHits(); long rowCount = response.getHits().getTotalHits();
Map<String, Object> res = new LinkedHashMap<>(); Map<String, Object> res = new LinkedHashMap<>();
res.put("esIndex", mapping.get_index()); res.put("esIndex", mapping.getIndex());
res.put("count", rowCount); res.put("count", rowCount);
return res; return res;
} }

26
client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/etl/ESEtlService.java
View File

@@ -51,7 +51,7 @@ public class ESEtlService extends AbstractEtlService {
public EtlResult importData(List<String> params) { public EtlResult importData(List<String> params) {
ESMapping mapping = config.getEsMapping(); ESMapping mapping = config.getEsMapping();
logger.info("start etl to import data to index: {}", mapping.get_index()); logger.info("start etl to import data to index: {}", mapping.getIndex());
String sql = mapping.getSql(); String sql = mapping.getSql();
return importData(sql, params); return importData(sql, params);
} }
@@ -78,7 +78,7 @@ public class ESEtlService extends AbstractEtlService {
} }
// 如果是主键字段则不插入 // 如果是主键字段则不插入
if (fieldItem.getFieldName().equals(mapping.get_id())) { if (fieldItem.getFieldName().equals(mapping.getId())) {
idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
} else { } else {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
@@ -117,8 +117,8 @@ public class ESEtlService extends AbstractEtlService {
if (idVal != null) { if (idVal != null) {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
@@ -127,8 +127,8 @@ public class ESEtlService extends AbstractEtlService {
esBulkRequest.add(esUpdateRequest); esBulkRequest.add(esUpdateRequest);
} else { } else {
ESIndexRequest esIndexRequest = this.esConnection.new ES6xIndexRequest(mapping.get_index(), ESIndexRequest esIndexRequest = this.esConnection.new ES6xIndexRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
idVal.toString()).setSource(esFieldData); idVal.toString()).setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esIndexRequest.setRouting(parentVal); esIndexRequest.setRouting(parentVal);
@@ -137,12 +137,12 @@ public class ESEtlService extends AbstractEtlService {
} }
} else { } else {
idVal = esFieldData.get(mapping.getPk()); idVal = esFieldData.get(mapping.getPk());
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)).size(10000); mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)).size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
esBulkRequest.add(esUpdateRequest); esBulkRequest.add(esUpdateRequest);
} }
@@ -160,7 +160,7 @@ public class ESEtlService extends AbstractEtlService {
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.getIndex());
} }
batchBegin = System.currentTimeMillis(); batchBegin = System.currentTimeMillis();
esBulkRequest.resetBulk(); esBulkRequest.resetBulk();
@@ -180,12 +180,12 @@ public class ESEtlService extends AbstractEtlService {
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.getIndex());
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage()); errMsg.add(mapping.getIndex() + " etl failed! ==>" + e.getMessage());
throw new RuntimeException(e); throw new RuntimeException(e);
} }
return count; return count;

70
client-adapter/es6x/src/main/java/com/alibaba/otter/canal/client/adapter/es6x/support/ES6xTemplate.java
View File

@@ -66,19 +66,19 @@ public class ES6xTemplate implements ESTemplate {
@Override @Override
public void insert(ESSyncConfig.ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void insert(ESSyncConfig.ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest updateRequest = esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest updateRequest = esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
updateRequest.setRouting(parentVal); updateRequest.setRouting(parentVal);
} }
getBulk().add(updateRequest); getBulk().add(updateRequest);
} else { } else {
ESIndexRequest indexRequest = esConnection.new ES6xIndexRequest(mapping.get_index(), ESIndexRequest indexRequest = esConnection.new ES6xIndexRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
pkVal.toString()).setSource(esFieldData); pkVal.toString()).setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
indexRequest.setRouting(parentVal); indexRequest.setRouting(parentVal);
@@ -87,13 +87,13 @@ public class ES6xTemplate implements ESTemplate {
} }
commitBulk(); commitBulk();
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000); mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
@@ -150,19 +150,19 @@ public class ES6xTemplate implements ESTemplate {
@Override @Override
public void delete(ESSyncConfig.ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void delete(ESSyncConfig.ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
ESDeleteRequest esDeleteRequest = this.esConnection.new ES6xDeleteRequest(mapping.get_index(), ESDeleteRequest esDeleteRequest = this.esConnection.new ES6xDeleteRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
pkVal.toString()); pkVal.toString());
getBulk().add(esDeleteRequest); getBulk().add(esDeleteRequest);
commitBulk(); commitBulk();
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000); mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
@@ -207,7 +207,7 @@ public class ES6xTemplate implements ESTemplate {
public Object getESDataFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet, public Object getESDataFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet,
Map<String, Object> esFieldData) throws SQLException { Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -216,7 +216,7 @@ public class ES6xTemplate implements ESTemplate {
resultIdVal = value; resultIdVal = value;
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.getId())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
@@ -231,7 +231,7 @@ public class ES6xTemplate implements ESTemplate {
@Override @Override
public Object getIdValFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet) throws SQLException { public Object getIdValFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -248,7 +248,7 @@ public class ES6xTemplate implements ESTemplate {
public Object getESDataFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld, public Object getESDataFromRS(ESSyncConfig.ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
Map<String, Object> esFieldData) throws SQLException { Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
if (fieldItem.getFieldName().equals(idFieldName)) { if (fieldItem.getFieldName().equals(idFieldName)) {
@@ -294,7 +294,7 @@ public class ES6xTemplate implements ESTemplate {
public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping, Map<String, Object> dmlData, public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping, Map<String, Object> dmlData,
Map<String, Object> esFieldData) { Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
String columnName = fieldItem.getColumnItems().iterator().next().getColumnName(); String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
@@ -304,7 +304,7 @@ public class ES6xTemplate implements ESTemplate {
resultIdVal = value; resultIdVal = value;
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.getId())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
@@ -319,7 +319,7 @@ public class ES6xTemplate implements ESTemplate {
public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping,String owner, Map<String, Object> dmlData, public Object getESDataFromDmlData(ESSyncConfig.ESMapping mapping,String owner, Map<String, Object> dmlData,
Map<String, Object> dmlOld, Map<String, Object> esFieldData) { Map<String, Object> dmlOld, Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
ColumnItem columnItem = fieldItem.getColumnItems().iterator().next(); ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
@@ -353,19 +353,19 @@ public class ES6xTemplate implements ESTemplate {
} }
private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
} }
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} else { } else {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
pkVal.toString()).setDoc(esFieldData); pkVal.toString()).setDoc(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
@@ -373,12 +373,12 @@ public class ES6xTemplate implements ESTemplate {
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} }
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex(),
mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000); mapping.getType()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES6xUpdateRequest(mapping.getIndex(),
mapping.get_type(), mapping.getType(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} }
@@ -394,15 +394,15 @@ public class ES6xTemplate implements ESTemplate {
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private String getEsType(ESMapping mapping, String fieldName) { private String getEsType(ESMapping mapping, String fieldName) {
String key = mapping.get_index() + "-" + mapping.get_type(); String key = mapping.getIndex() + "-" + mapping.getType();
Map<String, String> fieldType = esFieldTypes.get(key); Map<String, String> fieldType = esFieldTypes.get(key);
if (fieldType != null) { if (fieldType != null) {
return fieldType.get(fieldName); return fieldType.get(fieldName);
} else { } else {
MappingMetaData mappingMetaData = esConnection.getMapping(mapping.get_index(), mapping.get_type()); MappingMetaData mappingMetaData = esConnection.getMapping(mapping.getIndex(), mapping.getType());
if (mappingMetaData == null) { if (mappingMetaData == null) {
throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.getIndex());
} }
fieldType = new LinkedHashMap<>(); fieldType = new LinkedHashMap<>();

6
client-adapter/es6x/src/test/java/com/alibaba/otter/canal/client/adapter/es6x/test/ConfigLoadTest.java
View File

@@ -29,9 +29,9 @@ public class ConfigLoadTest {
Assert.assertNotNull(config); Assert.assertNotNull(config);
Assert.assertEquals("defaultDS", config.getDataSourceKey()); Assert.assertEquals("defaultDS", config.getDataSourceKey());
ESSyncConfig.ESMapping esMapping = config.getEsMapping(); ESSyncConfig.ESMapping esMapping = config.getEsMapping();
Assert.assertEquals("mytest_user", esMapping.get_index()); Assert.assertEquals("mytest_user", esMapping.getIndex());
Assert.assertEquals("_doc", esMapping.get_type()); Assert.assertEquals("_doc", esMapping.getType());
Assert.assertEquals("id", esMapping.get_id()); Assert.assertEquals("id", esMapping.getId());
Assert.assertNotNull(esMapping.getSql()); Assert.assertNotNull(esMapping.getSql());
// Map<String, List<ESSyncConfig>> dbTableEsSyncConfig = // Map<String, List<ESSyncConfig>> dbTableEsSyncConfig =

4
client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/ES7xAdapter.java
View File

@@ -59,11 +59,11 @@ public class ES7xAdapter extends ESAdapter {
public Map<String, Object> count(String task) { public Map<String, Object> count(String task) {
ESSyncConfig config = esSyncConfig.get(task); ESSyncConfig config = esSyncConfig.get(task);
ESSyncConfig.ESMapping mapping = config.getEsMapping(); ESSyncConfig.ESMapping mapping = config.getEsMapping();
SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index()).size(0).getResponse(); SearchResponse response = this.esConnection.new ESSearchRequest(mapping.getIndex()).size(0).getResponse();
long rowCount = response.getHits().getTotalHits().value; long rowCount = response.getHits().getTotalHits().value;
Map<String, Object> res = new LinkedHashMap<>(); Map<String, Object> res = new LinkedHashMap<>();
res.put("esIndex", mapping.get_index()); res.put("esIndex", mapping.getIndex());
res.put("count", rowCount); res.put("count", rowCount);
return res; return res;
} }

18
client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/etl/ESEtlService.java
View File

@@ -51,7 +51,7 @@ public class ESEtlService extends AbstractEtlService {
public EtlResult importData(List<String> params) { public EtlResult importData(List<String> params) {
ESMapping mapping = config.getEsMapping(); ESMapping mapping = config.getEsMapping();
logger.info("start etl to import data to index: {}", mapping.get_index()); logger.info("start etl to import data to index: {}", mapping.getIndex());
String sql = mapping.getSql(); String sql = mapping.getSql();
return importData(sql, params); return importData(sql, params);
} }
@@ -78,7 +78,7 @@ public class ESEtlService extends AbstractEtlService {
} }
// 如果是主键字段则不插入 // 如果是主键字段则不插入
if (fieldItem.getFieldName().equals(mapping.get_id())) { if (fieldItem.getFieldName().equals(mapping.getId())) {
idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
} else { } else {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
@@ -118,7 +118,7 @@ public class ESEtlService extends AbstractEtlService {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest( ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(
mapping.get_index(), mapping.getIndex(),
idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
@@ -128,7 +128,7 @@ public class ESEtlService extends AbstractEtlService {
esBulkRequest.add(esUpdateRequest); esBulkRequest.add(esUpdateRequest);
} else { } else {
ESIndexRequest esIndexRequest = this.esConnection.new ES7xIndexRequest( ESIndexRequest esIndexRequest = this.esConnection.new ES7xIndexRequest(
mapping.get_index(), mapping.getIndex(),
idVal.toString()).setSource(esFieldData); idVal.toString()).setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esIndexRequest.setRouting(parentVal); esIndexRequest.setRouting(parentVal);
@@ -137,13 +137,13 @@ public class ESEtlService extends AbstractEtlService {
} }
} else { } else {
idVal = esFieldData.get(mapping.getPk()); idVal = esFieldData.get(mapping.getPk());
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest( ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(
mapping.get_index(), mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
esBulkRequest.add(esUpdateRequest); esBulkRequest.add(esUpdateRequest);
} }
@@ -162,7 +162,7 @@ public class ESEtlService extends AbstractEtlService {
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.getIndex());
} }
batchBegin = System.currentTimeMillis(); batchBegin = System.currentTimeMillis();
esBulkRequest.resetBulk(); esBulkRequest.resetBulk();
@@ -182,12 +182,12 @@ public class ESEtlService extends AbstractEtlService {
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.getIndex());
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage()); errMsg.add(mapping.getIndex() + " etl failed! ==>" + e.getMessage());
throw new RuntimeException(e); throw new RuntimeException(e);
} }
return count; return count;

48
client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ES7xTemplate.java
View File

@@ -66,17 +66,17 @@ public class ES7xTemplate implements ESTemplate {
@Override @Override
public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.get_index(), ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.getIndex(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
updateRequest.setRouting(parentVal); updateRequest.setRouting(parentVal);
} }
getBulk().add(updateRequest); getBulk().add(updateRequest);
} else { } else {
ESIndexRequest indexRequest = esConnection.new ES7xIndexRequest(mapping.get_index(), pkVal.toString()) ESIndexRequest indexRequest = esConnection.new ES7xIndexRequest(mapping.getIndex(), pkVal.toString())
.setSource(esFieldData); .setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
indexRequest.setRouting(parentVal); indexRequest.setRouting(parentVal);
@@ -85,13 +85,13 @@ public class ES7xTemplate implements ESTemplate {
} }
commitBulk(); commitBulk();
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
@@ -148,18 +148,18 @@ public class ES7xTemplate implements ESTemplate {
@Override @Override
public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
ESDeleteRequest esDeleteRequest = this.esConnection.new ES7xDeleteRequest(mapping.get_index(), ESDeleteRequest esDeleteRequest = this.esConnection.new ES7xDeleteRequest(mapping.getIndex(),
pkVal.toString()); pkVal.toString());
getBulk().add(esDeleteRequest); getBulk().add(esDeleteRequest);
commitBulk(); commitBulk();
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
@@ -204,7 +204,7 @@ public class ES7xTemplate implements ESTemplate {
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet,
Map<String, Object> esFieldData) throws SQLException { Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -213,7 +213,7 @@ public class ES7xTemplate implements ESTemplate {
resultIdVal = value; resultIdVal = value;
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.getId())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
@@ -228,7 +228,7 @@ public class ES7xTemplate implements ESTemplate {
@Override @Override
public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException { public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -245,7 +245,7 @@ public class ES7xTemplate implements ESTemplate {
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld, public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
Map<String, Object> esFieldData) throws SQLException { Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
if (fieldItem.getFieldName().equals(idFieldName)) { if (fieldItem.getFieldName().equals(idFieldName)) {
@@ -289,7 +289,7 @@ public class ES7xTemplate implements ESTemplate {
@Override @Override
public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> esFieldData) { public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
String columnName = fieldItem.getColumnItems().iterator().next().getColumnName(); String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
@@ -299,7 +299,7 @@ public class ES7xTemplate implements ESTemplate {
resultIdVal = value; resultIdVal = value;
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.getId())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
@@ -314,7 +314,7 @@ public class ES7xTemplate implements ESTemplate {
public Object getESDataFromDmlData(ESMapping mapping,String owner, Map<String, Object> dmlData, Map<String, Object> dmlOld, public Object getESDataFromDmlData(ESMapping mapping,String owner, Map<String, Object> dmlData, Map<String, Object> dmlOld,
Map<String, Object> esFieldData) { Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
ColumnItem columnItem = fieldItem.getColumnItems().iterator().next(); ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
@@ -348,17 +348,17 @@ public class ES7xTemplate implements ESTemplate {
} }
private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
} }
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} else { } else {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
pkVal.toString()).setDoc(esFieldData); pkVal.toString()).setDoc(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
@@ -366,12 +366,12 @@ public class ES7xTemplate implements ESTemplate {
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} }
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} }
@@ -387,15 +387,15 @@ public class ES7xTemplate implements ESTemplate {
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private String getEsType(ESMapping mapping, String fieldName) { private String getEsType(ESMapping mapping, String fieldName) {
String key = mapping.get_index() + "-" + mapping.get_type(); String key = mapping.getIndex() + "-" + mapping.getType();
Map<String, String> fieldType = esFieldTypes.get(key); Map<String, String> fieldType = esFieldTypes.get(key);
if (fieldType != null) { if (fieldType != null) {
return fieldType.get(fieldName); return fieldType.get(fieldName);
} else { } else {
MappingMetaData mappingMetaData = esConnection.getMapping(mapping.get_index()); MappingMetaData mappingMetaData = esConnection.getMapping(mapping.getIndex());
if (mappingMetaData == null) { if (mappingMetaData == null) {
throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.getIndex());
} }
fieldType = new LinkedHashMap<>(); fieldType = new LinkedHashMap<>();

4
client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/ES8xAdapter.java
View File

@@ -53,11 +53,11 @@ public class ES8xAdapter extends ESAdapter {
public Map<String, Object> count(String task) { public Map<String, Object> count(String task) {
ESSyncConfig config = esSyncConfig.get(task); ESSyncConfig config = esSyncConfig.get(task);
ESSyncConfig.ESMapping mapping = config.getEsMapping(); ESSyncConfig.ESMapping mapping = config.getEsMapping();
SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index()).size(0).getResponse(); SearchResponse response = this.esConnection.new ESSearchRequest(mapping.getIndex()).size(0).getResponse();
long rowCount = response.getHits().getTotalHits().value; long rowCount = response.getHits().getTotalHits().value;
Map<String, Object> res = new LinkedHashMap<>(); Map<String, Object> res = new LinkedHashMap<>();
res.put("esIndex", mapping.get_index()); res.put("esIndex", mapping.getIndex());
res.put("count", rowCount); res.put("count", rowCount);
return res; return res;
} }

18
client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java
View File

@@ -51,7 +51,7 @@ public class ESEtlService extends AbstractEtlService {
public EtlResult importData(List<String> params) { public EtlResult importData(List<String> params) {
ESMapping mapping = config.getEsMapping(); ESMapping mapping = config.getEsMapping();
logger.info("start etl to import data to index: {}", mapping.get_index()); logger.info("start etl to import data to index: {}", mapping.getIndex());
String sql = mapping.getSql(); String sql = mapping.getSql();
return importData(sql, params); return importData(sql, params);
} }
@@ -78,7 +78,7 @@ public class ESEtlService extends AbstractEtlService {
} }
// 如果是主键字段则不插入 // 如果是主键字段则不插入
if (fieldItem.getFieldName().equals(mapping.get_id())) { if (fieldItem.getFieldName().equals(mapping.getId())) {
idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
} else { } else {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
@@ -118,7 +118,7 @@ public class ESEtlService extends AbstractEtlService {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest( ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(
mapping.get_index(), mapping.getIndex(),
idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
@@ -128,7 +128,7 @@ public class ESEtlService extends AbstractEtlService {
esBulkRequest.add(esUpdateRequest); esBulkRequest.add(esUpdateRequest);
} else { } else {
ESIndexRequest esIndexRequest = this.esConnection.new ES8xIndexRequest( ESIndexRequest esIndexRequest = this.esConnection.new ES8xIndexRequest(
mapping.get_index(), mapping.getIndex(),
idVal.toString()).setSource(esFieldData); idVal.toString()).setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esIndexRequest.setRouting(parentVal); esIndexRequest.setRouting(parentVal);
@@ -137,13 +137,13 @@ public class ESEtlService extends AbstractEtlService {
} }
} else { } else {
idVal = esFieldData.get(mapping.getPk()); idVal = esFieldData.get(mapping.getPk());
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest( ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(
mapping.get_index(), mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
esBulkRequest.add(esUpdateRequest); esBulkRequest.add(esUpdateRequest);
} }
@@ -162,7 +162,7 @@ public class ESEtlService extends AbstractEtlService {
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.getIndex());
} }
batchBegin = System.currentTimeMillis(); batchBegin = System.currentTimeMillis();
esBulkRequest.resetBulk(); esBulkRequest.resetBulk();
@@ -182,12 +182,12 @@ public class ESEtlService extends AbstractEtlService {
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.getIndex());
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage()); errMsg.add(mapping.getIndex() + " etl failed! ==>" + e.getMessage());
throw new RuntimeException(e); throw new RuntimeException(e);
} }
return count; return count;

48
client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java
View File

@@ -62,17 +62,17 @@ public class ES8xTemplate implements ESTemplate {
@Override @Override
public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest updateRequest = esConnection.new ES8xUpdateRequest(mapping.get_index(), ESUpdateRequest updateRequest = esConnection.new ES8xUpdateRequest(mapping.getIndex(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
updateRequest.setRouting(parentVal); updateRequest.setRouting(parentVal);
} }
getBulk().add(updateRequest); getBulk().add(updateRequest);
} else { } else {
ESIndexRequest indexRequest = esConnection.new ES8xIndexRequest(mapping.get_index(), pkVal.toString()) ESIndexRequest indexRequest = esConnection.new ES8xIndexRequest(mapping.getIndex(), pkVal.toString())
.setSource(esFieldData); .setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
indexRequest.setRouting(parentVal); indexRequest.setRouting(parentVal);
@@ -81,13 +81,13 @@ public class ES8xTemplate implements ESTemplate {
} }
commitBulk(); commitBulk();
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
@@ -144,18 +144,18 @@ public class ES8xTemplate implements ESTemplate {
@Override @Override
public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
ESDeleteRequest esDeleteRequest = this.esConnection.new ES8xDeleteRequest(mapping.get_index(), ESDeleteRequest esDeleteRequest = this.esConnection.new ES8xDeleteRequest(mapping.getIndex(),
pkVal.toString()); pkVal.toString());
getBulk().add(esDeleteRequest); getBulk().add(esDeleteRequest);
commitBulk(); commitBulk();
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
@@ -200,7 +200,7 @@ public class ES8xTemplate implements ESTemplate {
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet,
Map<String, Object> esFieldData) throws SQLException { Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -209,7 +209,7 @@ public class ES8xTemplate implements ESTemplate {
resultIdVal = value; resultIdVal = value;
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.getId())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
@@ -224,7 +224,7 @@ public class ES8xTemplate implements ESTemplate {
@Override @Override
public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException { public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
@@ -241,7 +241,7 @@ public class ES8xTemplate implements ESTemplate {
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld, public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
Map<String, Object> esFieldData) throws SQLException { Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
if (fieldItem.getFieldName().equals(idFieldName)) { if (fieldItem.getFieldName().equals(idFieldName)) {
@@ -286,7 +286,7 @@ public class ES8xTemplate implements ESTemplate {
public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData,
Map<String, Object> esFieldData) { Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
String columnName = fieldItem.getColumnItems().iterator().next().getColumnName(); String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
@@ -296,7 +296,7 @@ public class ES8xTemplate implements ESTemplate {
resultIdVal = value; resultIdVal = value;
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.getId())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
@@ -311,7 +311,7 @@ public class ES8xTemplate implements ESTemplate {
public Object getESDataFromDmlData(ESMapping mapping, String owner, Map<String, Object> dmlData, public Object getESDataFromDmlData(ESMapping mapping, String owner, Map<String, Object> dmlData,
Map<String, Object> dmlOld, Map<String, Object> esFieldData) { Map<String, Object> dmlOld, Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem(); SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
ColumnItem columnItem = fieldItem.getColumnItems().iterator().next(); ColumnItem columnItem = fieldItem.getColumnItems().iterator().next();
@@ -345,17 +345,17 @@ public class ES8xTemplate implements ESTemplate {
} }
private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing"); String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
} }
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} else { } else {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
pkVal.toString()).setDoc(esFieldData); pkVal.toString()).setDoc(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
@@ -363,12 +363,12 @@ public class ES8xTemplate implements ESTemplate {
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} }
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.getIndex())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000); .size(10000);
SearchResponse response = esSearchRequest.getResponse(); SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.getIndex(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} }
@@ -384,15 +384,15 @@ public class ES8xTemplate implements ESTemplate {
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private String getEsType(ESMapping mapping, String fieldName) { private String getEsType(ESMapping mapping, String fieldName) {
String key = mapping.get_index() + "-" + mapping.get_type(); String key = mapping.getIndex() + "-" + mapping.getType();
Map<String, String> fieldType = esFieldTypes.get(key); Map<String, String> fieldType = esFieldTypes.get(key);
if (fieldType != null) { if (fieldType != null) {
return fieldType.get(fieldName); return fieldType.get(fieldName);
} else { } else {
MappingMetadata mappingMetaData = esConnection.getMapping(mapping.get_index()); MappingMetadata mappingMetaData = esConnection.getMapping(mapping.getIndex());
if (mappingMetaData == null) { if (mappingMetaData == null) {
throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.getIndex());
} }
fieldType = new LinkedHashMap<>(); fieldType = new LinkedHashMap<>();

42
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/ESSyncConfig.java
View File

@@ -6,6 +6,7 @@ import java.util.List;
import java.util.Map; import java.util.Map;
import com.alibaba.otter.canal.client.adapter.support.AdapterConfig; import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
import org.springframework.beans.factory.annotation.Value;
/** /**
* ES 映射配置 * ES 映射配置
@@ -28,13 +29,13 @@ public class ESSyncConfig implements AdapterConfig {
private String esVersion = "es6"; private String esVersion = "es6";
public void validate() { public void validate() {
if (esMapping._index == null) { if (esMapping.index == null) {
throw new NullPointerException("esMapping._index"); throw new NullPointerException("esMapping._index");
} }
if ("es6".equals(esVersion) && esMapping._type == null) { if ("es6".equals(esVersion) && esMapping.type == null) {
throw new NullPointerException("esMapping._type"); throw new NullPointerException("esMapping._type");
} }
if (esMapping._id == null && esMapping.getPk() == null) { if (esMapping.id == null && esMapping.getPk() == null) {
throw new NullPointerException("esMapping._id or esMapping.pk"); throw new NullPointerException("esMapping._id or esMapping.pk");
} }
if (esMapping.sql == null) { if (esMapping.sql == null) {
@@ -96,9 +97,14 @@ public class ESSyncConfig implements AdapterConfig {
public static class ESMapping implements AdapterMapping { public static class ESMapping implements AdapterMapping {
private String _index; @Value("${_index}")
private String _type; private String index;
private String _id;
@Value("${_type}")
private String type;
@Value("${_id}")
private String id;
private boolean upsert = false; private boolean upsert = false;
private String pk; private String pk;
private Map<String, RelationMapping> relations = new LinkedHashMap<>(); private Map<String, RelationMapping> relations = new LinkedHashMap<>();
@@ -114,28 +120,28 @@ public class ESSyncConfig implements AdapterConfig {
private SchemaItem schemaItem; // sql解析结果模型 private SchemaItem schemaItem; // sql解析结果模型
public String get_index() { public String getIndex() {
return _index; return index;
} }
public void set_index(String _index) { public void setIndex(String index) {
this._index = _index; this.index = index;
} }
public String get_type() { public String getType() {
return _type; return type;
} }
public void set_type(String _type) { public void setType(String type) {
this._type = _type; this.type = type;
} }
public String get_id() { public String getId() {
return _id; return id;
} }
public void set_id(String _id) { public void setId(String id) {
this._id = _id; this.id = id;
} }
public boolean isUpsert() { public boolean isUpsert() {

4
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/config/SchemaItem.java
View File

@@ -136,8 +136,8 @@ public class SchemaItem {
} }
public FieldItem getIdFieldItem(ESSyncConfig.ESMapping mapping) { public FieldItem getIdFieldItem(ESSyncConfig.ESMapping mapping) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
return getSelectFields().get(mapping.get_id()); return getSelectFields().get(mapping.getId());
} else { } else {
return getSelectFields().get(mapping.getPk()); return getSelectFields().get(mapping.getPk());
} }

50
client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/service/ESSyncService.java
View File

@@ -54,13 +54,13 @@ public class ESSyncService {
for (ESSyncConfig config : esSyncConfigs) { for (ESSyncConfig config : esSyncConfigs) {
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Prepared to sync index: {}, destination: {}", logger.trace("Prepared to sync index: {}, destination: {}",
config.getEsMapping().get_index(), config.getEsMapping().getIndex(),
dml.getDestination()); dml.getDestination());
} }
this.sync(config, dml); this.sync(config, dml);
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Sync completed: {}, destination: {}", logger.trace("Sync completed: {}, destination: {}",
config.getEsMapping().get_index(), config.getEsMapping().getIndex(),
dml.getDestination()); dml.getDestination());
} }
} }
@@ -73,7 +73,7 @@ public class ESSyncService {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
StringBuilder configIndexes = new StringBuilder(); StringBuilder configIndexes = new StringBuilder();
esSyncConfigs esSyncConfigs
.forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" ")); .forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().getIndex()).append(" "));
logger.debug("DML: {} \nAffected indexes: {}", logger.debug("DML: {} \nAffected indexes: {}",
JSON.toJSONString(dml, JSONWriter.Feature.WriteNulls), JSON.toJSONString(dml, JSONWriter.Feature.WriteNulls),
configIndexes.toString()); configIndexes.toString());
@@ -105,10 +105,10 @@ public class ESSyncService {
logger.trace("Sync elapsed time: {} ms,destination: {}, es index: {}", logger.trace("Sync elapsed time: {} ms,destination: {}, es index: {}",
(System.currentTimeMillis() - begin), (System.currentTimeMillis() - begin),
dml.getDestination(), dml.getDestination(),
config.getEsMapping().get_index()); config.getEsMapping().getIndex());
} }
} catch (Throwable e) { } catch (Throwable e) {
logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().get_index(), dml); logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().getIndex(), dml);
throw new RuntimeException(e); throw new RuntimeException(e);
} }
} }
@@ -210,7 +210,7 @@ public class ESSyncService {
// ------主表 查询sql来更新------ // ------主表 查询sql来更新------
if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
ESMapping mapping = config.getEsMapping(); ESMapping mapping = config.getEsMapping();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.getId() == null ? mapping.getPk() : mapping.getId();
FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName); FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName);
boolean idFieldSimple = true; boolean idFieldSimple = true;
@@ -336,7 +336,7 @@ public class ESSyncService {
// ------是主表------ // ------是主表------
if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
if (mapping.get_id() != null) { if (mapping.getId() != null) {
FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping); FieldItem idFieldItem = schemaItem.getIdFieldItem(mapping);
// 主键为简单字段 // 主键为简单字段
if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) { if (!idFieldItem.isMethod() && !idFieldItem.isBinaryOp()) {
@@ -349,7 +349,7 @@ public class ESSyncService {
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.getIndex(),
idVal); idVal);
} }
esTemplate.delete(mapping, idVal, null); esTemplate.delete(mapping, idVal, null);
@@ -368,7 +368,7 @@ public class ESSyncService {
logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, pk: {}", logger.trace("Main table delete es index, destination:{}, table: {}, index: {}, pk: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
pkVal); pkVal);
} }
esFieldData.remove(pkFieldItem.getFieldName()); esFieldData.remove(pkFieldItem.getFieldName());
@@ -438,7 +438,7 @@ public class ESSyncService {
logger.trace("Single table insert to es index, destination:{}, table: {}, index: {}, id: {}", logger.trace("Single table insert to es index, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
idVal); idVal);
} }
esTemplate.insert(mapping, idVal, esFieldData); esTemplate.insert(mapping, idVal, esFieldData);
@@ -461,7 +461,7 @@ public class ESSyncService {
logger.trace("Main table insert to es index by query sql, destination:{}, table: {}, index: {}, sql: {}", logger.trace("Main table insert to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
sql.replace("\n", " ")); sql.replace("\n", " "));
} }
Util.sqlRS(ds, sql, rs -> { Util.sqlRS(ds, sql, rs -> {
@@ -475,7 +475,7 @@ public class ESSyncService {
"Main table insert to es index by query sql, destination:{}, table: {}, index: {}, id: {}", "Main table insert to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
idVal); idVal);
} }
esTemplate.insert(mapping, idVal, esFieldData); esTemplate.insert(mapping, idVal, esFieldData);
@@ -497,7 +497,7 @@ public class ESSyncService {
logger.trace("Main table delete es index by query sql, destination:{}, table: {}, index: {}, sql: {}", logger.trace("Main table delete es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
sql.replace("\n", " ")); sql.replace("\n", " "));
} }
Util.sqlRS(ds, sql, rs -> { Util.sqlRS(ds, sql, rs -> {
@@ -519,7 +519,7 @@ public class ESSyncService {
"Main table delete to es index by query sql, destination:{}, table: {}, index: {}, id: {}", "Main table delete to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
idVal); idVal);
} }
esTemplate.delete(mapping, idVal, esFieldData); esTemplate.delete(mapping, idVal, esFieldData);
@@ -554,7 +554,7 @@ public class ESSyncService {
String fieldName = fieldItem.getFieldName(); String fieldName = fieldItem.getFieldName();
// 判断是否是主键 // 判断是否是主键
if (fieldName.equals(mapping.get_id())) { if (fieldName.equals(mapping.getId())) {
fieldName = "_id"; fieldName = "_id";
} }
paramsTmp.put(fieldName, value); paramsTmp.put(fieldName, value);
@@ -566,7 +566,7 @@ public class ESSyncService {
logger.trace("Join table update es index by foreign key, destination:{}, table: {}, index: {}", logger.trace("Join table update es index by foreign key, destination:{}, table: {}, index: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index()); mapping.getIndex());
} }
esTemplate.updateByQuery(config, paramsTmp, esFieldData); esTemplate.updateByQuery(config, paramsTmp, esFieldData);
} }
@@ -617,7 +617,7 @@ public class ESSyncService {
logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}, sql: {}", logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
sql.toString().replace("\n", " ")); sql.toString().replace("\n", " "));
} }
Util.sqlRS(ds, sql.toString(), values, rs -> { Util.sqlRS(ds, sql.toString(), values, rs -> {
@@ -661,7 +661,7 @@ public class ESSyncService {
entry.getKey().getColumn().getColumnName()); entry.getKey().getColumn().getColumnName());
String fieldName = fieldItem.getFieldName(); String fieldName = fieldItem.getFieldName();
// 判断是否是主键 // 判断是否是主键
if (fieldName.equals(mapping.get_id())) { if (fieldName.equals(mapping.getId())) {
fieldName = "_id"; fieldName = "_id";
} }
paramsTmp.put(fieldName, value); paramsTmp.put(fieldName, value);
@@ -673,7 +673,7 @@ public class ESSyncService {
logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}", logger.trace("Join table update es index by query sql, destination:{}, table: {}, index: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index()); mapping.getIndex());
} }
esTemplate.updateByQuery(config, paramsTmp, esFieldData); esTemplate.updateByQuery(config, paramsTmp, esFieldData);
} }
@@ -721,7 +721,7 @@ public class ESSyncService {
logger.trace("Join table update es index by query whole sql, destination:{}, table: {}, index: {}, sql: {}", logger.trace("Join table update es index by query whole sql, destination:{}, table: {}, index: {}, sql: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
sql.toString().replace("\n", " ")); sql.toString().replace("\n", " "));
} }
Util.sqlRS(ds, sql.toString(), rs -> { Util.sqlRS(ds, sql.toString(), rs -> {
@@ -775,7 +775,7 @@ public class ESSyncService {
.getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName()); .getValFromRS(mapping, rs, fieldItem.getFieldName(), fieldItem.getFieldName());
String fieldName = fieldItem.getFieldName(); String fieldName = fieldItem.getFieldName();
// 判断是否是主键 // 判断是否是主键
if (fieldName.equals(mapping.get_id())) { if (fieldName.equals(mapping.getId())) {
fieldName = "_id"; fieldName = "_id";
} }
paramsTmp.put(fieldName, value); paramsTmp.put(fieldName, value);
@@ -787,7 +787,7 @@ public class ESSyncService {
"Join table update es index by query whole sql, destination:{}, table: {}, index: {}", "Join table update es index by query whole sql, destination:{}, table: {}, index: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index()); mapping.getIndex());
} }
esTemplate.updateByQuery(config, paramsTmp, esFieldData); esTemplate.updateByQuery(config, paramsTmp, esFieldData);
} }
@@ -817,7 +817,7 @@ public class ESSyncService {
logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}", logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
idVal); idVal);
} }
esTemplate.update(mapping, idVal, esFieldData); esTemplate.update(mapping, idVal, esFieldData);
@@ -840,7 +840,7 @@ public class ESSyncService {
logger.trace("Main table update to es index by query sql, destination:{}, table: {}, index: {}, sql: {}", logger.trace("Main table update to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
sql.replace("\n", " ")); sql.replace("\n", " "));
} }
Util.sqlRS(ds, sql, rs -> { Util.sqlRS(ds, sql, rs -> {
@@ -854,7 +854,7 @@ public class ESSyncService {
"Main table update to es index by query sql, destination:{}, table: {}, index: {}, id: {}", "Main table update to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.getIndex(),
idVal); idVal);
} }
esTemplate.update(mapping, idVal, esFieldData); esTemplate.update(mapping, idVal, esFieldData);