diff --git a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java index 705d3be2..5c8464c5 100644 --- a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java +++ b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/etl/ESEtlService.java @@ -1,5 +1,19 @@ package com.alibaba.otter.canal.client.adapter.es8x.etl; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + +import javax.sql.DataSource; + +import org.apache.commons.lang.StringUtils; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; + import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem; @@ -15,18 +29,6 @@ import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService; import com.alibaba.otter.canal.client.adapter.support.AdapterConfig; import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.Util; -import org.apache.commons.lang.StringUtils; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.search.SearchHit; - -import javax.sql.DataSource; -import java.sql.SQLException; -import java.util.HashMap; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.atomic.AtomicLong; /** * ES ETL Service @@ -37,10 +39,10 @@ import java.util.concurrent.atomic.AtomicLong; public class ESEtlService extends AbstractEtlService { private ESConnection esConnection; - private ESTemplate esTemplate; + private ESTemplate esTemplate; private ESSyncConfig config; - public ESEtlService(ESConnection esConnection, ESSyncConfig config) { + public ESEtlService(ESConnection esConnection, ESSyncConfig config){ super("ES", config); this.esConnection = esConnection; this.esTemplate = new ES8xTemplate(esConnection); @@ -91,14 +93,14 @@ public class ESEtlService extends AbstractEtlService { relations.put("name", relationMapping.getName()); if (StringUtils.isNotEmpty(relationMapping.getParent())) { FieldItem parentFieldItem = mapping.getSchemaItem() - .getSelectFields() - .get(relationMapping.getParent()); + .getSelectFields() + .get(relationMapping.getParent()); Object parentVal; try { parentVal = esTemplate.getValFromRS(mapping, - rs, - parentFieldItem.getFieldName(), - parentFieldItem.getFieldName()); + rs, + parentFieldItem.getFieldName(), + parentFieldItem.getFieldName()); } catch (SQLException e) { throw new RuntimeException(e); } @@ -116,8 +118,8 @@ public class ESEtlService extends AbstractEtlService { String parentVal = (String) esFieldData.remove("$parent_routing"); if (mapping.isUpsert()) { ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest( - mapping.get_index(), - idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); + mapping.get_index(), + idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); if (StringUtils.isNotEmpty(parentVal)) { esUpdateRequest.setRouting(parentVal); @@ -126,8 +128,8 @@ public class ESEtlService extends AbstractEtlService { esBulkRequest.add(esUpdateRequest); } else { ESIndexRequest esIndexRequest = this.esConnection.new ES8xIndexRequest( - mapping.get_index(), - idVal.toString()).setSource(esFieldData); + mapping.get_index(), + idVal.toString()).setSource(esFieldData); if (StringUtils.isNotEmpty(parentVal)) { esIndexRequest.setRouting(parentVal); } @@ -136,19 +138,19 @@ public class ESEtlService extends AbstractEtlService { } else { idVal = esFieldData.get(mapping.getPk()); ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) - .size(10000); + .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) + .size(10000); SearchResponse response = esSearchRequest.getResponse(); for (SearchHit hit : response.getHits()) { ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest( - mapping.get_index(), - hit.getId()).setDoc(esFieldData); + mapping.get_index(), + hit.getId()).setDoc(esFieldData); esBulkRequest.add(esUpdateRequest); } } if (esBulkRequest.numberOfActions() % mapping.getCommitBatch() == 0 - && esBulkRequest.numberOfActions() > 0) { + && esBulkRequest.numberOfActions() > 0) { long esBatchBegin = System.currentTimeMillis(); ESBulkResponse rp = esBulkRequest.bulk(); if (rp.hasFailures()) { @@ -157,10 +159,10 @@ public class ESEtlService extends AbstractEtlService { if (logger.isTraceEnabled()) { logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", - (System.currentTimeMillis() - batchBegin), - (System.currentTimeMillis() - esBatchBegin), - esBulkRequest.numberOfActions(), - mapping.get_index()); + (System.currentTimeMillis() - batchBegin), + (System.currentTimeMillis() - esBatchBegin), + esBulkRequest.numberOfActions(), + mapping.get_index()); } batchBegin = System.currentTimeMillis(); esBulkRequest.resetBulk(); @@ -177,10 +179,10 @@ public class ESEtlService extends AbstractEtlService { } if (logger.isTraceEnabled()) { logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", - (System.currentTimeMillis() - batchBegin), - (System.currentTimeMillis() - esBatchBegin), - esBulkRequest.numberOfActions(), - mapping.get_index()); + (System.currentTimeMillis() - batchBegin), + (System.currentTimeMillis() - esBatchBegin), + esBulkRequest.numberOfActions(), + mapping.get_index()); } } } catch (Exception e) { diff --git a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java index 6988b964..efced987 100644 --- a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java +++ b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ES8xTemplate.java @@ -1,5 +1,22 @@ package com.alibaba.otter.canal.client.adapter.es8x.support; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import javax.sql.DataSource; + +import org.apache.commons.lang.StringUtils; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.cluster.metadata.MappingMetadata; +import org.elasticsearch.index.query.BoolQueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem; @@ -15,37 +32,22 @@ import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.es8x.support.ESConnection.ESSearchRequest; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Util; -import org.apache.commons.lang.StringUtils; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.cluster.metadata.MappingMetadata; -import org.elasticsearch.index.query.BoolQueryBuilder; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.search.SearchHit; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.sql.DataSource; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; public class ES8xTemplate implements ESTemplate { - private static final Logger logger = LoggerFactory - .getLogger(ESTemplate.class); + private static final Logger logger = LoggerFactory + .getLogger(ESTemplate.class); - private static final int MAX_BATCH_SIZE = 1000; + private static final int MAX_BATCH_SIZE = 1000; - private ESConnection esConnection; + private ESConnection esConnection; - private ESBulkRequest esBulkRequest; + private ESBulkRequest esBulkRequest; // es 字段类型本地缓存 - private static ConcurrentMap> esFieldTypes = new ConcurrentHashMap<>(); + private static ConcurrentMap> esFieldTypes = new ConcurrentHashMap<>(); - public ES8xTemplate(ESConnection esConnection) { + public ES8xTemplate(ESConnection esConnection){ this.esConnection = esConnection; this.esBulkRequest = this.esConnection.new ES8xBulkRequest(); } @@ -64,14 +66,14 @@ public class ES8xTemplate implements ESTemplate { String parentVal = (String) esFieldData.remove("$parent_routing"); if (mapping.isUpsert()) { ESUpdateRequest updateRequest = esConnection.new ES8xUpdateRequest(mapping.get_index(), - pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); + pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); if (StringUtils.isNotEmpty(parentVal)) { updateRequest.setRouting(parentVal); } getBulk().add(updateRequest); } else { ESIndexRequest indexRequest = esConnection.new ES8xIndexRequest(mapping.get_index(), pkVal.toString()) - .setSource(esFieldData); + .setSource(esFieldData); if (StringUtils.isNotEmpty(parentVal)) { indexRequest.setRouting(parentVal); } @@ -80,13 +82,13 @@ public class ES8xTemplate implements ESTemplate { commitBulk(); } else { ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .size(10000); + .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + .size(10000); SearchResponse response = esSearchRequest.getResponse(); for (SearchHit hit : response.getHits()) { ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), - hit.getId()).setDoc(esFieldData); + hit.getId()).setDoc(esFieldData); getBulk().add(esUpdateRequest); commitBulk(); } @@ -144,17 +146,17 @@ public class ES8xTemplate implements ESTemplate { public void delete(ESMapping mapping, Object pkVal, Map esFieldData) { if (mapping.get_id() != null) { ESDeleteRequest esDeleteRequest = this.esConnection.new ES8xDeleteRequest(mapping.get_index(), - pkVal.toString()); + pkVal.toString()); getBulk().add(esDeleteRequest); commitBulk(); } else { ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .size(10000); + .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + .size(10000); SearchResponse response = esSearchRequest.getResponse(); for (SearchHit hit : response.getHits()) { ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), - hit.getId()).setDoc(esFieldData); + hit.getId()).setDoc(esFieldData); getBulk().add(esUpdateRequest); commitBulk(); } @@ -208,7 +210,7 @@ public class ES8xTemplate implements ESTemplate { } if (!fieldItem.getFieldName().equals(mapping.get_id()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { + && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } } @@ -248,9 +250,9 @@ public class ES8xTemplate implements ESTemplate { for (ColumnItem columnItem : fieldItem.getColumnItems()) { if (dmlOld.containsKey(columnItem.getColumnName()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { + && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), - getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); + getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); break; } } @@ -281,7 +283,8 @@ public class ES8xTemplate implements ESTemplate { } @Override - public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, Map esFieldData) { + public Object getESDataFromDmlData(ESMapping mapping, Map dmlData, + Map esFieldData) { SchemaItem schemaItem = mapping.getSchemaItem(); String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); Object resultIdVal = null; @@ -294,7 +297,7 @@ public class ES8xTemplate implements ESTemplate { } if (!fieldItem.getFieldName().equals(mapping.get_id()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { + && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } } @@ -305,8 +308,8 @@ public class ES8xTemplate implements ESTemplate { } @Override - public Object getESDataFromDmlData(ESMapping mapping, String owner, Map dmlData, Map dmlOld, - Map esFieldData) { + public Object getESDataFromDmlData(ESMapping mapping, String owner, Map dmlData, + Map dmlOld, Map esFieldData) { SchemaItem schemaItem = mapping.getSchemaItem(); String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); Object resultIdVal = null; @@ -323,7 +326,7 @@ public class ES8xTemplate implements ESTemplate { if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), - getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); + getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); } } @@ -346,14 +349,14 @@ public class ES8xTemplate implements ESTemplate { String parentVal = (String) esFieldData.remove("$parent_routing"); if (mapping.isUpsert()) { ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), - pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); + pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); if (StringUtils.isNotEmpty(parentVal)) { esUpdateRequest.setRouting(parentVal); } getBulk().add(esUpdateRequest); } else { ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), - pkVal.toString()).setDoc(esFieldData); + pkVal.toString()).setDoc(esFieldData); if (StringUtils.isNotEmpty(parentVal)) { esUpdateRequest.setRouting(parentVal); } @@ -361,12 +364,12 @@ public class ES8xTemplate implements ESTemplate { } } else { ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .size(10000); + .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) + .size(10000); SearchResponse response = esSearchRequest.getResponse(); for (SearchHit hit : response.getHits()) { ESUpdateRequest esUpdateRequest = this.esConnection.new ES8xUpdateRequest(mapping.get_index(), - hit.getId()).setDoc(esFieldData); + hit.getId()).setDoc(esFieldData); getBulk().add(esUpdateRequest); } } @@ -375,7 +378,7 @@ public class ES8xTemplate implements ESTemplate { /** * 获取es mapping中的属性类型 * - * @param mapping mapping配置 + * @param mapping mapping配置 * @param fieldName 属性名 * @return 类型 */ @@ -422,9 +425,9 @@ public class ES8xTemplate implements ESTemplate { Object parentVal; try { parentVal = getValFromRS(mapping, - resultSet, - parentFieldItem.getFieldName(), - parentFieldItem.getFieldName()); + resultSet, + parentFieldItem.getFieldName(), + parentFieldItem.getFieldName()); } catch (SQLException e) { throw new RuntimeException(e); } diff --git a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ESConnection.java b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ESConnection.java index f4ef033f..429542d7 100644 --- a/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ESConnection.java +++ b/client-adapter/es8x/src/main/java/com/alibaba/otter/canal/client/adapter/es8x/support/ESConnection.java @@ -1,6 +1,12 @@ package com.alibaba.otter.canal.client.adapter.es8x.support; -import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest; +import java.io.IOException; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.UnknownHostException; +import java.util.Arrays; +import java.util.Map; + import org.apache.commons.lang.StringUtils; import org.apache.http.HttpHost; import org.apache.http.auth.AuthScope; @@ -30,12 +36,7 @@ import org.elasticsearch.search.builder.SearchSourceBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.net.URI; -import java.net.URISyntaxException; -import java.net.UnknownHostException; -import java.util.Arrays; -import java.util.Map; +import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest; /** * ES 连接器, 只支持 Rest 方式 @@ -47,20 +48,22 @@ public class ESConnection { private static final Logger logger = LoggerFactory.getLogger(ESConnection.class); - private RestHighLevelClient restHighLevelClient; - public ESConnection(String[] hosts, Map properties) throws UnknownHostException { + public ESConnection(String[] hosts, Map properties) throws UnknownHostException{ HttpHost[] httpHosts = Arrays.stream(hosts).map(this::createHttpHost).toArray(HttpHost[]::new); RestClientBuilder restClientBuilder = RestClient.builder(httpHosts); String nameAndPwd = properties.get("security.auth"); if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) { String[] nameAndPwdArr = nameAndPwd.split(":"); final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); - credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1])); - restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); + credentialsProvider.setCredentials(AuthScope.ANY, + new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1])); + restClientBuilder.setHttpClientConfigCallback( + httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); } - restHighLevelClient = new RestHighLevelClientBuilder(restClientBuilder.build()).setApiCompatibilityMode(true).build(); + restHighLevelClient = new RestHighLevelClientBuilder(restClientBuilder.build()).setApiCompatibilityMode(true) + .build(); } public void close() { @@ -96,9 +99,9 @@ public class ESConnection { private IndexRequestBuilder indexRequestBuilder; - private IndexRequest indexRequest; + private IndexRequest indexRequest; - public ES8xIndexRequest(String index, String id) { + public ES8xIndexRequest(String index, String id){ indexRequest = new IndexRequest(index); indexRequest.id(id); @@ -139,9 +142,9 @@ public class ESConnection { private UpdateRequestBuilder updateRequestBuilder; - private UpdateRequest updateRequest; + private UpdateRequest updateRequest; - public ES8xUpdateRequest(String index, String id) { + public ES8xUpdateRequest(String index, String id){ updateRequest = new UpdateRequest(index, id); } @@ -188,9 +191,9 @@ public class ESConnection { private DeleteRequestBuilder deleteRequestBuilder; - private DeleteRequest deleteRequest; + private DeleteRequest deleteRequest; - public ES8xDeleteRequest(String index, String id) { + public ES8xDeleteRequest(String index, String id){ deleteRequest = new DeleteRequest(index, id); @@ -217,11 +220,11 @@ public class ESConnection { private SearchRequestBuilder searchRequestBuilder; - private SearchRequest searchRequest; + private SearchRequest searchRequest; - private SearchSourceBuilder sourceBuilder; + private SearchSourceBuilder sourceBuilder; - public ESSearchRequest(String index) { + public ESSearchRequest(String index){ searchRequest = new SearchRequest(index); sourceBuilder = new SearchSourceBuilder(); @@ -274,9 +277,9 @@ public class ESConnection { private BulkRequestBuilder bulkRequestBuilder; - private BulkRequest bulkRequest; + private BulkRequest bulkRequest; - public ES8xBulkRequest() { + public ES8xBulkRequest(){ bulkRequest = new BulkRequest(); @@ -347,7 +350,7 @@ public class ESConnection { private BulkResponse bulkResponse; - public ES8xBulkResponse(BulkResponse bulkResponse) { + public ES8xBulkResponse(BulkResponse bulkResponse){ this.bulkResponse = bulkResponse; } @@ -372,7 +375,6 @@ public class ESConnection { } } - public RestHighLevelClient getRestHighLevelClient() { return restHighLevelClient; } @@ -387,7 +389,9 @@ public class ESConnection { return HttpHost.create(uri.toString()); } try { - return HttpHost.create(new URI(uri.getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(), uri.getQuery(), uri.getFragment()).toString()); + return HttpHost.create(new URI(uri + .getScheme(), null, uri.getHost(), uri.getPort(), uri.getPath(), uri.getQuery(), uri.getFragment()) + .toString()); } catch (URISyntaxException ex) { throw new IllegalStateException(ex); }