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

modify

Browse Source
This commit is contained in:
mcy
2019-03-20 09:10:15 +08:00
parent a2360efff9
commit 5821c1f606
1 changed files with 3 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -202,7 +202,7 @@ public class ESEtlService {
long batchBegin = System.currentTimeMillis(); long batchBegin = System.currentTimeMillis();
while (rs.next()) { while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>(); Map<String, Object> esFieldData = new LinkedHashMap<>();
Map<String, Object> idEsFieldData = new HashMap<>(); Object idVal = null;
for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) { for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) {
String fieldName = fieldItem.getFieldName(); String fieldName = fieldItem.getFieldName();
@@ -212,8 +212,7 @@ public class ESEtlService {
// 如果是主键字段则不插入 // 如果是主键字段则不插入
if (fieldItem.getFieldName().equals(mapping.get_id())) { if (fieldItem.getFieldName().equals(mapping.get_id())) {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
idEsFieldData.put(Util.cleanColumn(fieldName), val);
} else { } else {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
esFieldData.put(Util.cleanColumn(fieldName), val); esFieldData.put(Util.cleanColumn(fieldName), val);
@@ -248,11 +247,6 @@ public class ESEtlService {
}); });
} }
Object idVal = null;
if (mapping.get_id() != null) {
idVal = idEsFieldData.get(mapping.get_id());
}
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()) {
@@ -274,7 +268,7 @@ public class ESEtlService {
bulkRequestBuilder.add(indexRequestBuilder); bulkRequestBuilder.add(indexRequestBuilder);
} }
} else { } else {
idVal = idEsFieldData.get(mapping.getPk()); idVal = esFieldData.get(mapping.getPk());
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(), idVal)) .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))