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

es父子文档索引ETL支持

Browse Source
This commit is contained in:
mcy
2019-03-13 17:19:08 +08:00
parent 4df60a0fb5
commit e03705e211
1 changed files with 46 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,9 +1,7 @@
package com.alibaba.otter.canal.client.adapter.es.service; package com.alibaba.otter.canal.client.adapter.es.service;
import java.util.ArrayList; import java.sql.SQLException;
import java.util.LinkedHashMap; import java.util.*;
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;
@@ -13,13 +11,14 @@ import java.util.regex.Pattern;
import javax.sql.DataSource; import javax.sql.DataSource;
import org.apache.commons.lang.StringUtils;
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;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType; import org.elasticsearch.action.update.UpdateRequestBuilder;
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;
@@ -199,21 +198,58 @@ public class ESEtlService {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName); Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
esFieldData.put(fieldName, val); esFieldData.put(fieldName, val);
} }
if (!mapping.getRelations().isEmpty()) {
mapping.getRelations().forEach((relationField, relationMapping) -> {
Map<String, Object> relations = new HashMap<>();
relations.put("name", relationMapping.getName());
if (StringUtils.isNotEmpty(relationMapping.getParent())) {
FieldItem parentFieldItem = mapping.getSchemaItem()
.getSelectFields()
.get(relationMapping.getParent());
Object parentVal;
try {
parentVal = esTemplate.getValFromRS(mapping,
rs,
parentFieldItem.getFieldName(),
parentFieldItem.getFieldName());
} catch (SQLException e) {
throw new RuntimeException(e);
}
if (parentVal != null) {
relations.put("parent", parentVal.toString());
esFieldData.put("$parent_routing", parentVal.toString());
}
}
esFieldData.put(relationField, relations);
});
}
Object idVal = null; Object idVal = null;
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
idVal = rs.getObject(mapping.get_id()); idVal = rs.getObject(mapping.get_id());
} }
if (idVal != null) { if (idVal != null) {
String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
bulkRequestBuilder.add(transportClient UpdateRequestBuilder updateRequestBuilder = transportClient
.prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString()) .prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString())
.setDoc(esFieldData) .setDoc(esFieldData)
.setDocAsUpsert(true)); .setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) {
updateRequestBuilder.setRouting(parentVal);
}
bulkRequestBuilder.add(updateRequestBuilder);
} else { } else {
bulkRequestBuilder.add(transportClient IndexRequestBuilder indexRequestBuilder = transportClient
.prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString()) .prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString())
.setSource(esFieldData)); .setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) {
indexRequestBuilder.setRouting(parentVal);
}
bulkRequestBuilder.add(indexRequestBuilder);
} }
} else { } else {
idVal = rs.getObject(mapping.getPk()); idVal = rs.getObject(mapping.getPk());