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

format es8

Browse Source
This commit is contained in:
jianghang.loujh
2023-09-01 19:58:23 +08:00
parent 63dfbf65df
commit 2786e06ccf
3 changed files with 119 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,19 @@
package com.alibaba.otter.canal.client.adapter.es8x.etl; 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;
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem; 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.AdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.Util; 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 * ES ETL Service
@@ -37,10 +39,10 @@ import java.util.concurrent.atomic.AtomicLong;
public class ESEtlService extends AbstractEtlService { public class ESEtlService extends AbstractEtlService {
private ESConnection esConnection; private ESConnection esConnection;
private ESTemplate esTemplate; private ESTemplate esTemplate;
private ESSyncConfig config; private ESSyncConfig config;
public ESEtlService(ESConnection esConnection, ESSyncConfig config) { public ESEtlService(ESConnection esConnection, ESSyncConfig config){
super("ES", config); super("ES", config);
this.esConnection = esConnection; this.esConnection = esConnection;
this.esTemplate = new ES8xTemplate(esConnection); this.esTemplate = new ES8xTemplate(esConnection);
@@ -91,14 +93,14 @@ public class ESEtlService extends AbstractEtlService {
relations.put("name", relationMapping.getName()); relations.put("name", relationMapping.getName());
if (StringUtils.isNotEmpty(relationMapping.getParent())) { if (StringUtils.isNotEmpty(relationMapping.getParent())) {
FieldItem parentFieldItem = mapping.getSchemaItem() FieldItem parentFieldItem = mapping.getSchemaItem()
.getSelectFields() .getSelectFields()
.get(relationMapping.getParent()); .get(relationMapping.getParent());
Object parentVal; Object parentVal;
try { try {
parentVal = esTemplate.getValFromRS(mapping, parentVal = esTemplate.getValFromRS(mapping,
rs, rs,
parentFieldItem.getFieldName(), parentFieldItem.getFieldName(),
parentFieldItem.getFieldName()); parentFieldItem.getFieldName());
} catch (SQLException e) { } catch (SQLException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
@@ -116,8 +118,8 @@ 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.get_index(),
idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
@@ -126,8 +128,8 @@ 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.get_index(),
idVal.toString()).setSource(esFieldData); idVal.toString()).setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esIndexRequest.setRouting(parentVal); esIndexRequest.setRouting(parentVal);
} }
@@ -136,19 +138,19 @@ 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.get_index())
.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.get_index(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
esBulkRequest.add(esUpdateRequest); esBulkRequest.add(esUpdateRequest);
} }
} }
if (esBulkRequest.numberOfActions() % mapping.getCommitBatch() == 0 if (esBulkRequest.numberOfActions() % mapping.getCommitBatch() == 0
&& esBulkRequest.numberOfActions() > 0) { && esBulkRequest.numberOfActions() > 0) {
long esBatchBegin = System.currentTimeMillis(); long esBatchBegin = System.currentTimeMillis();
ESBulkResponse rp = esBulkRequest.bulk(); ESBulkResponse rp = esBulkRequest.bulk();
if (rp.hasFailures()) { if (rp.hasFailures()) {
@@ -157,10 +159,10 @@ public class ESEtlService extends AbstractEtlService {
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.get_index());
} }
batchBegin = System.currentTimeMillis(); batchBegin = System.currentTimeMillis();
esBulkRequest.resetBulk(); esBulkRequest.resetBulk();
@@ -177,10 +179,10 @@ public class ESEtlService extends AbstractEtlService {
} }
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(), esBulkRequest.numberOfActions(),
mapping.get_index()); mapping.get_index());
} }
} }
} catch (Exception e) { } catch (Exception e) {

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

@@ -1,5 +1,22 @@
package com.alibaba.otter.canal.client.adapter.es8x.support; 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;
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig.ESMapping;
import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem; 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.es8x.support.ESConnection.ESSearchRequest;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.Util; 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 { public class ES8xTemplate implements ESTemplate {
private static final Logger logger = LoggerFactory private static final Logger logger = LoggerFactory
.getLogger(ESTemplate.class); .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 字段类型本地缓存 // es 字段类型本地缓存
private static ConcurrentMap<String, Map<String, String>> esFieldTypes = new ConcurrentHashMap<>(); private static ConcurrentMap<String, Map<String, String>> esFieldTypes = new ConcurrentHashMap<>();
public ES8xTemplate(ESConnection esConnection) { public ES8xTemplate(ESConnection esConnection){
this.esConnection = esConnection; this.esConnection = esConnection;
this.esBulkRequest = this.esConnection.new ES8xBulkRequest(); this.esBulkRequest = this.esConnection.new ES8xBulkRequest();
} }
@@ -64,14 +66,14 @@ public class ES8xTemplate implements ESTemplate {
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.get_index(),
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.get_index(), pkVal.toString())
.setSource(esFieldData); .setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
indexRequest.setRouting(parentVal); indexRequest.setRouting(parentVal);
} }
@@ -80,13 +82,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.get_index())
.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.get_index(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
} }
@@ -144,17 +146,17 @@ public class ES8xTemplate implements ESTemplate {
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.get_id() != null) {
ESDeleteRequest esDeleteRequest = this.esConnection.new ES8xDeleteRequest(mapping.get_index(), ESDeleteRequest esDeleteRequest = this.esConnection.new ES8xDeleteRequest(mapping.get_index(),
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.get_index())
.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.get_index(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
commitBulk(); commitBulk();
} }
@@ -208,7 +210,7 @@ public class ES8xTemplate implements ESTemplate {
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
} }
@@ -248,9 +250,9 @@ public class ES8xTemplate implements ESTemplate {
for (ColumnItem columnItem : fieldItem.getColumnItems()) { for (ColumnItem columnItem : fieldItem.getColumnItems()) {
if (dmlOld.containsKey(columnItem.getColumnName()) if (dmlOld.containsKey(columnItem.getColumnName())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()));
break; break;
} }
} }
@@ -281,7 +283,8 @@ public class ES8xTemplate 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.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null; Object resultIdVal = null;
@@ -294,7 +297,7 @@ public class ES8xTemplate implements ESTemplate {
} }
if (!fieldItem.getFieldName().equals(mapping.get_id()) if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
} }
} }
@@ -305,8 +308,8 @@ public class ES8xTemplate implements ESTemplate {
} }
@Override @Override
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> 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.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null; Object resultIdVal = null;
@@ -323,7 +326,7 @@ public class ES8xTemplate implements ESTemplate {
if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) { if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(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"); 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.get_index(),
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.get_index(),
pkVal.toString()).setDoc(esFieldData); pkVal.toString()).setDoc(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) { if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal); esUpdateRequest.setRouting(parentVal);
} }
@@ -361,12 +364,12 @@ public class ES8xTemplate implements ESTemplate {
} }
} else { } else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index()) ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
.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.get_index(),
hit.getId()).setDoc(esFieldData); hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest); getBulk().add(esUpdateRequest);
} }
} }
@@ -375,7 +378,7 @@ public class ES8xTemplate implements ESTemplate {
/** /**
* 获取es mapping中的属性类型 * 获取es mapping中的属性类型
* *
* @param mapping mapping配置 * @param mapping mapping配置
* @param fieldName 属性名 * @param fieldName 属性名
* @return 类型 * @return 类型
*/ */
@@ -422,9 +425,9 @@ public class ES8xTemplate implements ESTemplate {
Object parentVal; Object parentVal;
try { try {
parentVal = getValFromRS(mapping, parentVal = getValFromRS(mapping,
resultSet, resultSet,
parentFieldItem.getFieldName(), parentFieldItem.getFieldName(),
parentFieldItem.getFieldName()); parentFieldItem.getFieldName());
} catch (SQLException e) { } catch (SQLException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }

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

@@ -1,6 +1,12 @@
package com.alibaba.otter.canal.client.adapter.es8x.support; 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.commons.lang.StringUtils;
import org.apache.http.HttpHost; import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope; import org.apache.http.auth.AuthScope;
@@ -30,12 +36,7 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.io.IOException; import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Map;
/** /**
* ES 连接器, 只支持 Rest 方式 * ES 连接器, 只支持 Rest 方式
@@ -47,20 +48,22 @@ public class ESConnection {
private static final Logger logger = LoggerFactory.getLogger(ESConnection.class); private static final Logger logger = LoggerFactory.getLogger(ESConnection.class);
private RestHighLevelClient restHighLevelClient; private RestHighLevelClient restHighLevelClient;
public ESConnection(String[] hosts, Map<String, String> properties) throws UnknownHostException { public ESConnection(String[] hosts, Map<String, String> properties) throws UnknownHostException{
HttpHost[] httpHosts = Arrays.stream(hosts).map(this::createHttpHost).toArray(HttpHost[]::new); HttpHost[] httpHosts = Arrays.stream(hosts).map(this::createHttpHost).toArray(HttpHost[]::new);
RestClientBuilder restClientBuilder = RestClient.builder(httpHosts); RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);
String nameAndPwd = properties.get("security.auth"); String nameAndPwd = properties.get("security.auth");
if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) { if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) {
String[] nameAndPwdArr = nameAndPwd.split(":"); String[] nameAndPwdArr = nameAndPwd.split(":");
final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1])); credentialsProvider.setCredentials(AuthScope.ANY,
restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); 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() { public void close() {
@@ -96,9 +99,9 @@ public class ESConnection {
private IndexRequestBuilder indexRequestBuilder; 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 = new IndexRequest(index);
indexRequest.id(id); indexRequest.id(id);
@@ -139,9 +142,9 @@ public class ESConnection {
private UpdateRequestBuilder updateRequestBuilder; 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); updateRequest = new UpdateRequest(index, id);
} }
@@ -188,9 +191,9 @@ public class ESConnection {
private DeleteRequestBuilder deleteRequestBuilder; 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); deleteRequest = new DeleteRequest(index, id);
@@ -217,11 +220,11 @@ public class ESConnection {
private SearchRequestBuilder searchRequestBuilder; 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); searchRequest = new SearchRequest(index);
sourceBuilder = new SearchSourceBuilder(); sourceBuilder = new SearchSourceBuilder();
@@ -274,9 +277,9 @@ public class ESConnection {
private BulkRequestBuilder bulkRequestBuilder; private BulkRequestBuilder bulkRequestBuilder;
private BulkRequest bulkRequest; private BulkRequest bulkRequest;
public ES8xBulkRequest() { public ES8xBulkRequest(){
bulkRequest = new BulkRequest(); bulkRequest = new BulkRequest();
@@ -347,7 +350,7 @@ public class ESConnection {
private BulkResponse bulkResponse; private BulkResponse bulkResponse;
public ES8xBulkResponse(BulkResponse bulkResponse) { public ES8xBulkResponse(BulkResponse bulkResponse){
this.bulkResponse = bulkResponse; this.bulkResponse = bulkResponse;
} }
@@ -372,7 +375,6 @@ public class ESConnection {
} }
} }
public RestHighLevelClient getRestHighLevelClient() { public RestHighLevelClient getRestHighLevelClient() {
return restHighLevelClient; return restHighLevelClient;
} }
@@ -387,7 +389,9 @@ public class ESConnection {
return HttpHost.create(uri.toString()); return HttpHost.create(uri.toString());
} }
try { 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) { } catch (URISyntaxException ex) {
throw new IllegalStateException(ex); throw new IllegalStateException(ex);
} }