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

Add elasticsearch 7.x version adapter support (#2246)

Browse Source 
* init es data sync upgrade

* Add elasticsearch 7.x version adapter support
This commit is contained in:
rewerma
2019-09-25 18:51:34 +08:00
committed by agapple
parent 61726c7e05
commit f45c0206da
54 changed files with 7172 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
client-adapter/es7x/pom.xml Normal file
View File

@@ -0,0 +1,100 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId>
<version>1.1.5-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.es7x</artifactId>
<packaging>jar</packaging>
<name>canal client adapter es v7x module for otter ${project.version}</name>
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>client-adapter.es-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>7.3.0</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>7.3.0</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
<version>7.3.0</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.3.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>run</goal>
</goals>
<configuration>
<tasks>
<copy todir="${project.basedir}/../launcher/target/classes/es7" overwrite="true">
<fileset dir="${project.basedir}/target/classes/es7" erroronmissingdir="true">
<include name="*.yml" />
</fileset>
</copy>
</tasks>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

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

@@ -0,0 +1,123 @@
package com.alibaba.otter.canal.client.adapter.es7x;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import com.alibaba.otter.canal.client.adapter.es7x.etl.ESEtlService;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import org.elasticsearch.action.search.SearchResponse;
import com.alibaba.otter.canal.client.adapter.es.core.ESAdapter;
import com.alibaba.otter.canal.client.adapter.es.core.config.ESSyncConfig;
import com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate;
import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection;
import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.SPI;
import javax.sql.DataSource;
/**
* ES 7.x 外部适配器
*
* @author rewerma 2019-09-23
* @version 1.0.0
*/
@SPI("es7")
public class ES7xAdapter extends ESAdapter {
private ESConnection esConnection;
public ESConnection getEsConnection() {
return esConnection;
}
@Override
public void init(OuterAdapterConfig configuration, Properties envProperties) {
try {
Map<String, String> properties = configuration.getProperties();
String[] hostArray = configuration.getHosts().split(",");
String mode = properties.get("mode");
if ("rest".equalsIgnoreCase(mode) || "http".equalsIgnoreCase(mode)) {
esConnection = new ESConnection(hostArray, properties, ESConnection.ESClientMode.REST);
} else {
esConnection = new ESConnection(hostArray, properties, ESConnection.ESClientMode.TRANSPORT);
}
this.esTemplate = new ES7xTemplate(esConnection);
envProperties.put("es.version", "es7");
super.init(configuration, envProperties);
} catch (Throwable e) {
throw new RuntimeException(e);
}
}
@Override
public Map<String, Object> count(String task) {
ESSyncConfig config = esSyncConfig.get(task);
ESSyncConfig.ESMapping mapping = config.getEsMapping();
SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index()).size(0).getResponse();
long rowCount = response.getHits().getTotalHits().value;
Map<String, Object> res = new LinkedHashMap<>();
res.put("esIndex", mapping.get_index());
res.put("count", rowCount);
return res;
}
@Override
public EtlResult etl(String task, List<String> params) {
EtlResult etlResult = new EtlResult();
ESSyncConfig config = esSyncConfig.get(task);
if (config != null) {
DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
ESEtlService esEtlService = new ESEtlService(esConnection, config);
if (dataSource != null) {
return esEtlService.importData(params);
} else {
etlResult.setSucceeded(false);
etlResult.setErrorMessage("DataSource not found");
return etlResult;
}
} else {
StringBuilder resultMsg = new StringBuilder();
boolean resSuccess = true;
for (ESSyncConfig configTmp : esSyncConfig.values()) {
// 取所有的destination为task的配置
if (configTmp.getDestination().equals(task)) {
ESEtlService esEtlService = new ESEtlService(esConnection, configTmp);
EtlResult etlRes = esEtlService.importData(params);
if (!etlRes.getSucceeded()) {
resSuccess = false;
resultMsg.append(etlRes.getErrorMessage()).append("\n");
} else {
resultMsg.append(etlRes.getResultMessage()).append("\n");
}
}
}
if (resultMsg.length() > 0) {
etlResult.setSucceeded(resSuccess);
if (resSuccess) {
etlResult.setResultMessage(resultMsg.toString());
} else {
etlResult.setErrorMessage(resultMsg.toString());
}
return etlResult;
}
}
etlResult.setSucceeded(false);
etlResult.setErrorMessage("Task not found");
return etlResult;
}
@Override
public void destroy() {
super.destroy();
if (esConnection != null) {
esConnection.close();
}
}
}

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

@@ -0,0 +1,202 @@
package com.alibaba.otter.canal.client.adapter.es7x.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;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESBulkResponse;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESIndexRequest;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESUpdateRequest;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate;
import com.alibaba.otter.canal.client.adapter.es7x.support.ES7xTemplate;
import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection;
import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection.ESSearchRequest;
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;
/**
* ES ETL Service
*
* @author rewerma 2018-11-01
* @version 1.0.0
*/
public class ESEtlService extends AbstractEtlService {
private ESConnection esConnection;
private ESTemplate esTemplate;
private ESSyncConfig config;
public ESEtlService(ESConnection esConnection, ESSyncConfig config){
super("ES", config);
this.esConnection = esConnection;
this.esTemplate = new ES7xTemplate(esConnection);
this.config = config;
}
public EtlResult importData(List<String> params) {
ESMapping mapping = config.getEsMapping();
logger.info("start etl to import data to index: {}", mapping.get_index());
String sql = mapping.getSql();
return importData(sql, params);
}
protected boolean executeSqlImport(DataSource ds, String sql, List<Object> values,
AdapterConfig.AdapterMapping adapterMapping, AtomicLong impCount,
List<String> errMsg) {
try {
ESMapping mapping = (ESMapping) adapterMapping;
Util.sqlRS(ds, sql, values, rs -> {
int count = 0;
try {
ESBulkRequest esBulkRequest = this.esConnection.new ES7xBulkRequest();
long batchBegin = System.currentTimeMillis();
while (rs.next()) {
Map<String, Object> esFieldData = new LinkedHashMap<>();
Object idVal = null;
for (FieldItem fieldItem : mapping.getSchemaItem().getSelectFields().values()) {
String fieldName = fieldItem.getFieldName();
if (mapping.getSkips().contains(fieldName)) {
continue;
}
// 如果是主键字段则不插入
if (fieldItem.getFieldName().equals(mapping.get_id())) {
idVal = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
} else {
Object val = esTemplate.getValFromRS(mapping, rs, fieldName, fieldName);
esFieldData.put(Util.cleanColumn(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(Util.cleanColumn(relationField), relations);
});
}
if (idVal != null) {
String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(
mapping.get_index(),
idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal);
}
esBulkRequest.add(esUpdateRequest);
} else {
ESIndexRequest esIndexRequest = this.esConnection.new ES7xIndexRequest(
mapping.get_index(),
idVal.toString()).setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) {
esIndexRequest.setRouting(parentVal);
}
esBulkRequest.add(esIndexRequest);
}
} else {
idVal = esFieldData.get(mapping.getPk());
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
.size(10000);
SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(
mapping.get_index(),
hit.getId()).setDoc(esFieldData);
esBulkRequest.add(esUpdateRequest);
}
}
if (esBulkRequest.numberOfActions() % mapping.getCommitBatch() == 0
&& esBulkRequest.numberOfActions() > 0) {
long esBatchBegin = System.currentTimeMillis();
ESBulkResponse rp = esBulkRequest.bulk();
if (rp.hasFailures()) {
rp.processFailBulkResponse("全量数据 etl 异常 ");
}
if (logger.isTraceEnabled()) {
logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(),
mapping.get_index());
}
batchBegin = System.currentTimeMillis();
esBulkRequest.resetBulk();
}
count++;
impCount.incrementAndGet();
}
if (esBulkRequest.numberOfActions() > 0) {
long esBatchBegin = System.currentTimeMillis();
ESBulkResponse rp = esBulkRequest.bulk();
if (rp.hasFailures()) {
rp.processFailBulkResponse("全量数据 etl 异常 ");
}
if (logger.isTraceEnabled()) {
logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin),
esBulkRequest.numberOfActions(),
mapping.get_index());
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
errMsg.add(mapping.get_index() + " etl failed! ==>" + e.getMessage());
throw new RuntimeException(e);
}
return count;
});
return true;
} catch (Exception e) {
logger.error(e.getMessage(), e);
return false;
}
}
}

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

@@ -0,0 +1,462 @@
package com.alibaba.otter.canal.client.adapter.es7x.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;
import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.ColumnItem;
import com.alibaba.otter.canal.client.adapter.es.core.config.SchemaItem.FieldItem;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESBulkResponse;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESDeleteRequest;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESIndexRequest;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest.ESUpdateRequest;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESSyncUtil;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESTemplate;
import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection.ESSearchRequest;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
public class ES7xTemplate implements ESTemplate {
private static final Logger logger = LoggerFactory
.getLogger(ESTemplate.class);
private static final int MAX_BATCH_SIZE = 1000;
private ESConnection esConnection;
private ESBulkRequest esBulkRequest;
// es 字段类型本地缓存
private static ConcurrentMap<String, Map<String, String>> esFieldTypes = new ConcurrentHashMap<>();
public ES7xTemplate(ESConnection esConnection){
this.esConnection = esConnection;
this.esBulkRequest = this.esConnection.new ES7xBulkRequest();
}
public ESBulkRequest getBulk() {
return esBulkRequest;
}
public void resetBulkRequestBuilder() {
this.esBulkRequest.resetBulk();
}
@Override
public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) {
ESUpdateRequest updateRequest = esConnection.new ES7xUpdateRequest(mapping.get_index(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) {
updateRequest.setRouting(parentVal);
}
getBulk().add(updateRequest);
} else {
ESIndexRequest indexRequest = esConnection.new ES7xIndexRequest(mapping.get_index(), pkVal.toString())
.setSource(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) {
indexRequest.setRouting(parentVal);
}
getBulk().add(indexRequest);
}
commitBulk();
} else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000);
SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest);
commitBulk();
}
}
}
@Override
public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
Map<String, Object> esFieldDataTmp = new LinkedHashMap<>(esFieldData.size());
esFieldData.forEach((k, v) -> esFieldDataTmp.put(Util.cleanColumn(k), v));
append4Update(mapping, pkVal, esFieldDataTmp);
commitBulk();
}
@Override
public void updateByQuery(ESSyncConfig config, Map<String, Object> paramsTmp, Map<String, Object> esFieldData) {
if (paramsTmp.isEmpty()) {
return;
}
ESMapping mapping = config.getEsMapping();
BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value)));
// 查询sql批量更新
DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE ");
List<Object> values = new ArrayList<>();
paramsTmp.forEach((fieldName, value) -> {
sql.append("_v.").append(fieldName).append("=? AND ");
values.add(value);
});
// TODO 直接外部包裹sql会导致全表扫描性能低, 待优化拼接内部where条件
int len = sql.length();
sql.delete(len - 4, len);
Integer syncCount = (Integer) Util.sqlRS(ds, sql.toString(), values, rs -> {
int count = 0;
try {
while (rs.next()) {
Object idVal = getIdValFromRS(mapping, rs);
append4Update(mapping, idVal, esFieldData);
commitBulk();
count++;
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return count;
});
if (logger.isTraceEnabled()) {
logger.trace("Update ES by query affected {} records", syncCount);
}
}
@Override
public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) {
ESDeleteRequest esDeleteRequest = this.esConnection.new ES7xDeleteRequest(mapping.get_index(),
pkVal.toString());
getBulk().add(esDeleteRequest);
commitBulk();
} else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000);
SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest);
commitBulk();
}
}
}
@Override
public void commit() {
if (getBulk().numberOfActions() > 0) {
ESBulkResponse response = getBulk().bulk();
if (response.hasFailures()) {
response.processFailBulkResponse("ES sync commit error ");
}
resetBulkRequestBuilder();
}
}
@Override
public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
String columnName) throws SQLException {
fieldName = Util.cleanColumn(fieldName);
columnName = Util.cleanColumn(columnName);
String esType = getEsType(mapping, fieldName);
Object value = resultSet.getObject(columnName);
if (value instanceof Boolean) {
if (!"boolean".equals(esType)) {
value = resultSet.getByte(columnName);
}
}
// 如果是对象类型
if (mapping.getObjFields().containsKey(fieldName)) {
return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName));
} else {
return ESSyncUtil.typeConvert(value, esType);
}
}
@Override
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet,
Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
if (fieldItem.getFieldName().equals(idFieldName)) {
resultIdVal = value;
}
if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
}
}
// 添加父子文档关联信息
putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
return resultIdVal;
}
@Override
public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
if (fieldItem.getFieldName().equals(idFieldName)) {
resultIdVal = value;
break;
}
}
return resultIdVal;
}
@Override
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
Map<String, Object> esFieldData) throws SQLException {
SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
if (fieldItem.getFieldName().equals(idFieldName)) {
resultIdVal = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
}
for (ColumnItem columnItem : fieldItem.getColumnItems()) {
if (dmlOld.containsKey(columnItem.getColumnName())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()));
break;
}
}
}
// 添加父子文档关联信息
putRelationDataFromRS(mapping, schemaItem, resultSet, esFieldData);
return resultIdVal;
}
@Override
public Object getValFromData(ESMapping mapping, Map<String, Object> dmlData, String fieldName, String columnName) {
String esType = getEsType(mapping, fieldName);
Object value = dmlData.get(columnName);
if (value instanceof Byte) {
if ("boolean".equals(esType)) {
value = ((Byte) value).intValue() != 0;
}
}
// 如果是对象类型
if (mapping.getObjFields().containsKey(fieldName)) {
return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName));
} else {
return ESSyncUtil.typeConvert(value, esType);
}
}
@Override
public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData,
Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
Object value = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
if (fieldItem.getFieldName().equals(idFieldName)) {
resultIdVal = value;
}
if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value);
}
}
// 添加父子文档关联信息
putRelationData(mapping, schemaItem, dmlData, esFieldData);
return resultIdVal;
}
@Override
public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> dmlOld,
Map<String, Object> esFieldData) {
SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
if (fieldItem.getFieldName().equals(idFieldName)) {
resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
}
if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()),
getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
}
}
// 添加父子文档关联信息
putRelationData(mapping, schemaItem, dmlOld, esFieldData);
return resultIdVal;
}
/**
* 如果大于批量数则提交批次
*/
private void commitBulk() {
if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {
commit();
}
}
private void append4Update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) {
String parentVal = (String) esFieldData.remove("$parent_routing");
if (mapping.isUpsert()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true);
if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal);
}
getBulk().add(esUpdateRequest);
} else {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
pkVal.toString()).setDoc(esFieldData);
if (StringUtils.isNotEmpty(parentVal)) {
esUpdateRequest.setRouting(parentVal);
}
getBulk().add(esUpdateRequest);
}
} else {
ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.size(10000);
SearchResponse response = esSearchRequest.getResponse();
for (SearchHit hit : response.getHits()) {
ESUpdateRequest esUpdateRequest = this.esConnection.new ES7xUpdateRequest(mapping.get_index(),
hit.getId()).setDoc(esFieldData);
getBulk().add(esUpdateRequest);
}
}
}
/**
* 获取es mapping中的属性类型
*
* @param mapping mapping配置
* @param fieldName 属性名
* @return 类型
*/
@SuppressWarnings("unchecked")
private String getEsType(ESMapping mapping, String fieldName) {
String key = mapping.get_index() + "-" + mapping.get_type();
Map<String, String> fieldType = esFieldTypes.get(key);
if (fieldType != null) {
return fieldType.get(fieldName);
} else {
MappingMetaData mappingMetaData = esConnection.getMapping(mapping.get_index());
if (mappingMetaData == null) {
throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
}
fieldType = new LinkedHashMap<>();
Map<String, Object> sourceMap = mappingMetaData.getSourceAsMap();
Map<String, Object> esMapping = (Map<String, Object>) sourceMap.get("properties");
for (Map.Entry<String, Object> entry : esMapping.entrySet()) {
Map<String, Object> value = (Map<String, Object>) entry.getValue();
if (value.containsKey("properties")) {
fieldType.put(entry.getKey(), "object");
} else {
fieldType.put(entry.getKey(), (String) value.get("type"));
}
}
esFieldTypes.put(key, fieldType);
return fieldType.get(fieldName);
}
}
private void putRelationDataFromRS(ESMapping mapping, SchemaItem schemaItem, ResultSet resultSet,
Map<String, Object> esFieldData) {
// 添加父子文档关联信息
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 = schemaItem.getSelectFields().get(relationMapping.getParent());
Object parentVal;
try {
parentVal = getValFromRS(mapping,
resultSet,
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);
});
}
}
private void putRelationData(ESMapping mapping, SchemaItem schemaItem, Map<String, Object> dmlData,
Map<String, Object> esFieldData) {
// 添加父子文档关联信息
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 = schemaItem.getSelectFields().get(relationMapping.getParent());
String columnName = parentFieldItem.getColumnItems().iterator().next().getColumnName();
Object parentVal = getValFromData(mapping, dmlData, parentFieldItem.getFieldName(), columnName);
if (parentVal != null) {
relations.put("parent", parentVal.toString());
esFieldData.put("$parent_routing", parentVal.toString());
}
}
esFieldData.put(relationField, relations);
});
}
}
}

504
client-adapter/es7x/src/main/java/com/alibaba/otter/canal/client/adapter/es7x/support/ESConnection.java Normal file
View File

@@ -0,0 +1,504 @@
package com.alibaba.otter.canal.client.adapter.es7x.support;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteRequestBuilder;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetMappingsRequest;
import org.elasticsearch.client.indices.GetMappingsResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.es.core.support.ESBulkRequest;
/**
* ES 连接器, Transport Rest 两种方式
*
* @author rewerma 2019-08-01
* @version 1.0.0
*/
public class ESConnection {
private static final Logger logger = LoggerFactory.getLogger(ESConnection.class);
public enum ESClientMode {
TRANSPORT, REST
}
private ESClientMode mode;
private TransportClient transportClient;
private RestHighLevelClient restHighLevelClient;
public ESConnection(String[] hosts, Map<String, String> properties, ESClientMode mode) throws UnknownHostException{
this.mode = mode;
if (mode == ESClientMode.TRANSPORT) {
Settings.Builder settingBuilder = Settings.builder();
settingBuilder.put("cluster.name", properties.get("cluster.name"));
Settings settings = settingBuilder.build();
transportClient = new PreBuiltTransportClient(settings);
for (String host : hosts) {
int i = host.indexOf(":");
transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)),
Integer.parseInt(host.substring(i + 1))));
}
} else {
HttpHost[] httpHosts = new HttpHost[hosts.length];
for (int i = 0; i < hosts.length; i++) {
String host = hosts[i];
int j = host.indexOf(":");
HttpHost httpHost = new HttpHost(InetAddress.getByName(host.substring(0, j)),
Integer.parseInt(host.substring(j + 1)));
httpHosts[i] = httpHost;
}
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));
}
restHighLevelClient = new RestHighLevelClient(restClientBuilder);
}
}
public void close() {
if (mode == ESClientMode.TRANSPORT) {
transportClient.close();
} else {
try {
restHighLevelClient.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
public MappingMetaData getMapping(String index) {
MappingMetaData mappingMetaData = null;
if (mode == ESClientMode.TRANSPORT) {
try {
mappingMetaData = transportClient.admin()
.cluster()
.prepareState()
.execute()
.actionGet()
.getState()
.getMetaData()
.getIndices()
.get(index)
.mapping();
} catch (NullPointerException e) {
throw new IllegalArgumentException("Not found the mapping info of index: " + index);
}
} else {
Map<String, MappingMetaData> mappings;
try {
GetMappingsRequest request = new GetMappingsRequest();
request.indices(index);
GetMappingsResponse response = restHighLevelClient.indices()
.getMapping(request, RequestOptions.DEFAULT);
mappings = response.mappings();
} catch (NullPointerException e) {
throw new IllegalArgumentException("Not found the mapping info of index: " + index);
} catch (IOException e) {
logger.error(e.getMessage(), e);
return null;
}
mappingMetaData = mappings.get(index);
}
return mappingMetaData;
}
public class ES7xIndexRequest implements ESBulkRequest.ESIndexRequest {
private IndexRequestBuilder indexRequestBuilder;
private IndexRequest indexRequest;
public ES7xIndexRequest(String index, String id){
if (mode == ESClientMode.TRANSPORT) {
indexRequestBuilder = transportClient.prepareIndex();
indexRequestBuilder.setIndex(index);
indexRequestBuilder.setId(id);
} else {
indexRequest = new IndexRequest(index);
indexRequest.id(id);
}
}
public ES7xIndexRequest setSource(Map<String, ?> source) {
if (mode == ESClientMode.TRANSPORT) {
indexRequestBuilder.setSource(source);
} else {
indexRequest.source(source);
}
return this;
}
public ES7xIndexRequest setRouting(String routing) {
if (mode == ESClientMode.TRANSPORT) {
indexRequestBuilder.setRouting(routing);
} else {
indexRequest.routing(routing);
}
return this;
}
public IndexRequestBuilder getIndexRequestBuilder() {
return indexRequestBuilder;
}
public void setIndexRequestBuilder(IndexRequestBuilder indexRequestBuilder) {
this.indexRequestBuilder = indexRequestBuilder;
}
public IndexRequest getIndexRequest() {
return indexRequest;
}
public void setIndexRequest(IndexRequest indexRequest) {
this.indexRequest = indexRequest;
}
}
public class ES7xUpdateRequest implements ESBulkRequest.ESUpdateRequest {
private UpdateRequestBuilder updateRequestBuilder;
private UpdateRequest updateRequest;
public ES7xUpdateRequest(String index, String id){
if (mode == ESClientMode.TRANSPORT) {
updateRequestBuilder = transportClient.prepareUpdate();
updateRequestBuilder.setIndex(index);
updateRequestBuilder.setId(id);
} else {
updateRequest = new UpdateRequest(index, id);
}
}
public ES7xUpdateRequest setDoc(Map source) {
if (mode == ESClientMode.TRANSPORT) {
updateRequestBuilder.setDoc(source);
} else {
updateRequest.doc(source);
}
return this;
}
public ES7xUpdateRequest setDocAsUpsert(boolean shouldUpsertDoc) {
if (mode == ESClientMode.TRANSPORT) {
updateRequestBuilder.setDocAsUpsert(shouldUpsertDoc);
} else {
updateRequest.docAsUpsert(shouldUpsertDoc);
}
return this;
}
public ES7xUpdateRequest setRouting(String routing) {
if (mode == ESClientMode.TRANSPORT) {
updateRequestBuilder.setRouting(routing);
} else {
updateRequest.routing(routing);
}
return this;
}
public UpdateRequestBuilder getUpdateRequestBuilder() {
return updateRequestBuilder;
}
public void setUpdateRequestBuilder(UpdateRequestBuilder updateRequestBuilder) {
this.updateRequestBuilder = updateRequestBuilder;
}
public UpdateRequest getUpdateRequest() {
return updateRequest;
}
public void setUpdateRequest(UpdateRequest updateRequest) {
this.updateRequest = updateRequest;
}
}
public class ES7xDeleteRequest implements ESBulkRequest.ESDeleteRequest {
private DeleteRequestBuilder deleteRequestBuilder;
private DeleteRequest deleteRequest;
public ES7xDeleteRequest(String index, String id){
if (mode == ESClientMode.TRANSPORT) {
deleteRequestBuilder = transportClient.prepareDelete();
deleteRequestBuilder.setIndex(index);
deleteRequestBuilder.setId(id);
} else {
deleteRequest = new DeleteRequest(index, id);
}
}
public DeleteRequestBuilder getDeleteRequestBuilder() {
return deleteRequestBuilder;
}
public void setDeleteRequestBuilder(DeleteRequestBuilder deleteRequestBuilder) {
this.deleteRequestBuilder = deleteRequestBuilder;
}
public DeleteRequest getDeleteRequest() {
return deleteRequest;
}
public void setDeleteRequest(DeleteRequest deleteRequest) {
this.deleteRequest = deleteRequest;
}
}
public class ESSearchRequest {
private SearchRequestBuilder searchRequestBuilder;
private SearchRequest searchRequest;
private SearchSourceBuilder sourceBuilder;
public ESSearchRequest(String index){
if (mode == ESClientMode.TRANSPORT) {
searchRequestBuilder = transportClient.prepareSearch(index);
} else {
searchRequest = new SearchRequest(index);
sourceBuilder = new SearchSourceBuilder();
}
}
public ESSearchRequest setQuery(QueryBuilder queryBuilder) {
if (mode == ESClientMode.TRANSPORT) {
searchRequestBuilder.setQuery(queryBuilder);
} else {
sourceBuilder.query(queryBuilder);
}
return this;
}
public ESSearchRequest size(int size) {
if (mode == ESClientMode.TRANSPORT) {
searchRequestBuilder.setSize(size);
} else {
sourceBuilder.size(size);
}
return this;
}
public SearchResponse getResponse() {
if (mode == ESClientMode.TRANSPORT) {
return searchRequestBuilder.get();
} else {
searchRequest.source(sourceBuilder);
try {
return restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
public SearchRequestBuilder getSearchRequestBuilder() {
return searchRequestBuilder;
}
public void setSearchRequestBuilder(SearchRequestBuilder searchRequestBuilder) {
this.searchRequestBuilder = searchRequestBuilder;
}
public SearchRequest getSearchRequest() {
return searchRequest;
}
public void setSearchRequest(SearchRequest searchRequest) {
this.searchRequest = searchRequest;
}
}
public class ES7xBulkRequest implements ESBulkRequest {
private BulkRequestBuilder bulkRequestBuilder;
private BulkRequest bulkRequest;
public ES7xBulkRequest(){
if (mode == ESClientMode.TRANSPORT) {
bulkRequestBuilder = transportClient.prepareBulk();
} else {
bulkRequest = new BulkRequest();
}
}
public void resetBulk() {
if (mode == ESClientMode.TRANSPORT) {
bulkRequestBuilder = transportClient.prepareBulk();
} else {
bulkRequest = new BulkRequest();
}
}
public ES7xBulkRequest add(ESIndexRequest esIndexRequest) {
ES7xIndexRequest eir = (ES7xIndexRequest) esIndexRequest;
if (mode == ESClientMode.TRANSPORT) {
bulkRequestBuilder.add(eir.indexRequestBuilder);
} else {
bulkRequest.add(eir.indexRequest);
}
return this;
}
public ES7xBulkRequest add(ESUpdateRequest esUpdateRequest) {
ES7xUpdateRequest eur = (ES7xUpdateRequest) esUpdateRequest;
if (mode == ESClientMode.TRANSPORT) {
bulkRequestBuilder.add(eur.updateRequestBuilder);
} else {
bulkRequest.add(eur.updateRequest);
}
return this;
}
public ES7xBulkRequest add(ESDeleteRequest esDeleteRequest) {
ES7xDeleteRequest edr = (ES7xDeleteRequest) esDeleteRequest;
if (mode == ESClientMode.TRANSPORT) {
bulkRequestBuilder.add(edr.deleteRequestBuilder);
} else {
bulkRequest.add(edr.deleteRequest);
}
return this;
}
public int numberOfActions() {
if (mode == ESClientMode.TRANSPORT) {
return bulkRequestBuilder.numberOfActions();
} else {
return bulkRequest.numberOfActions();
}
}
public ESBulkResponse bulk() {
if (mode == ESClientMode.TRANSPORT) {
BulkResponse responses = bulkRequestBuilder.execute().actionGet();
return new ES7xBulkResponse(responses);
} else {
try {
BulkResponse responses = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
return new ES7xBulkResponse(responses);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
public BulkRequestBuilder getBulkRequestBuilder() {
return bulkRequestBuilder;
}
public void setBulkRequestBuilder(BulkRequestBuilder bulkRequestBuilder) {
this.bulkRequestBuilder = bulkRequestBuilder;
}
public BulkRequest getBulkRequest() {
return bulkRequest;
}
public void setBulkRequest(BulkRequest bulkRequest) {
this.bulkRequest = bulkRequest;
}
}
public static class ES7xBulkResponse implements ESBulkRequest.ESBulkResponse {
private BulkResponse bulkResponse;
public ES7xBulkResponse(BulkResponse bulkResponse){
this.bulkResponse = bulkResponse;
}
@Override
public boolean hasFailures() {
return bulkResponse.hasFailures();
}
@Override
public void processFailBulkResponse(String errorMsg) {
for (BulkItemResponse itemResponse : bulkResponse.getItems()) {
if (!itemResponse.isFailed()) {
continue;
}
if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
logger.error(itemResponse.getFailureMessage());
} else {
throw new RuntimeException(errorMsg + itemResponse.getFailureMessage());
}
}
}
}
// ------ get/set ------
public ESClientMode getMode() {
return mode;
}
public void setMode(ESClientMode mode) {
this.mode = mode;
}
public TransportClient getTransportClient() {
return transportClient;
}
public void setTransportClient(TransportClient transportClient) {
this.transportClient = transportClient;
}
public RestHighLevelClient getRestHighLevelClient() {
return restHighLevelClient;
}
public void setRestHighLevelClient(RestHighLevelClient restHighLevelClient) {
this.restHighLevelClient = restHighLevelClient;
}
}

1
client-adapter/es7x/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter Normal file
View File

@@ -0,0 +1 @@
es7=com.alibaba.otter.canal.client.adapter.es7x.ES7xAdapter

20
client-adapter/es7x/src/main/resources/es7/biz_order.yml Normal file
View File

@@ -0,0 +1,20 @@
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: customer
_id: _id
relations:
customer_order:
name: order
parent: customer_id
sql: "select concat('oid_', t.id) as _id,
t.customer_id,
t.id as order_id,
t.serial_code as order_serial,
t.c_time as order_time
from biz_order t"
skips:
- customer_id
etlCondition: "where t.c_time>={}"
commitBatch: 3000

46
client-adapter/es7x/src/main/resources/es7/customer.yml Normal file
View File

@@ -0,0 +1,46 @@
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: customer
_id: id
relations:
customer_order:
name: customer
sql: "select t.id, t.name, t.email from customer t"
etlCondition: "where t.c_time>={}"
commitBatch: 3000
#{
# "mappings":{
# "_doc":{
# "properties":{
# "id": {
# "type": "long"
# },
# "name": {
# "type": "text"
# },
# "email": {
# "type": "text"
# },
# "order_id": {
# "type": "long"
# },
# "order_serial": {
# "type": "text"
# },
# "order_time": {
# "type": "date"
# },
# "customer_order":{
# "type":"join",
# "relations":{
# "customer":"order"
# }
# }
# }
# }
# }
#}

15
client-adapter/es7x/src/main/resources/es7/mytest_user.yml Normal file
View File

@@ -0,0 +1,15 @@
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: mytest_user
_id: _id
# upsert: true
# pk: id
sql: "select a.id as _id, a.name, a.role_id, b.role_name,
a.c_time from user a
left join role b on b.id=a.role_id"
# objFields:
# _labels: array:;
etlCondition: "where a.c_time>={}"
commitBatch: 3000

119
client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ES7xTest.java Normal file
View File

@@ -0,0 +1,119 @@
package com.alibaba.otter.canal.client.adapter.es7x.test;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.LinkedHashMap;
import java.util.Map;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
@Ignore
public class ES7xTest {
private TransportClient transportClient;
@Before
public void init() throws UnknownHostException {
Settings.Builder settingBuilder = Settings.builder();
settingBuilder.put("cluster.name", TestConstant.clusterName);
Settings settings = settingBuilder.build();
transportClient = new PreBuiltTransportClient(settings);
String[] hostArray = TestConstant.esHosts.split(",");
for (String host : hostArray) {
int i = host.indexOf(":");
transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)),
Integer.parseInt(host.substring(i + 1))));
}
}
@Test
public void test01() {
SearchResponse response = transportClient.prepareSearch("test")
.setQuery(QueryBuilders.termQuery("_id", "1"))
.setSize(10000)
.get();
for (SearchHit hit : response.getHits()) {
System.out.println(hit.getSourceAsMap().get("data").getClass());
}
}
@Test
public void test02() {
Map<String, Object> esFieldData = new LinkedHashMap<>();
esFieldData.put("userId", 2L);
esFieldData.put("eventId", 4L);
esFieldData.put("eventName", "网络异常");
esFieldData.put("description", "第四个事件信息");
Map<String, Object> relations = new LinkedHashMap<>();
esFieldData.put("user_event", relations);
relations.put("name", "event");
relations.put("parent", "2");
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
bulkRequestBuilder
.add(transportClient.prepareIndex("test", "osm", "2_4").setRouting("2").setSource(esFieldData));
commit(bulkRequestBuilder);
}
@Test
public void test03() {
Map<String, Object> esFieldData = new LinkedHashMap<>();
esFieldData.put("userId", 2L);
esFieldData.put("eventName", "网络异常1");
Map<String, Object> relations = new LinkedHashMap<>();
esFieldData.put("user_event", relations);
relations.put("name", "event");
relations.put("parent", "2");
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
bulkRequestBuilder.add(transportClient.prepareUpdate("test", "osm", "2_4").setRouting("2").setDoc(esFieldData));
commit(bulkRequestBuilder);
}
@Test
public void test04() {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
bulkRequestBuilder.add(transportClient.prepareDelete("test", "osm", "2_4"));
commit(bulkRequestBuilder);
}
private void commit(BulkRequestBuilder bulkRequestBuilder) {
if (bulkRequestBuilder.numberOfActions() > 0) {
BulkResponse response = bulkRequestBuilder.execute().actionGet();
if (response.hasFailures()) {
for (BulkItemResponse itemResponse : response.getItems()) {
if (!itemResponse.isFailed()) {
continue;
}
if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
System.out.println(itemResponse.getFailureMessage());
} else {
System.out.println("ES bulk commit error" + itemResponse.getFailureMessage());
}
}
}
}
}
@After
public void after() {
transportClient.close();
}
}

45
client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/ESConnectionTest.java Normal file
View File

@@ -0,0 +1,45 @@
package com.alibaba.otter.canal.client.adapter.es7x.test;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import com.alibaba.otter.canal.client.adapter.es7x.support.ESConnection;
import org.springframework.util.Assert;
@Ignore
public class ESConnectionTest {
ESConnection esConnection;
@Before
public void init() throws UnknownHostException {
String[] hosts = new String[] { "127.0.0.1:9200" };
Map<String, String> properties = new HashMap<>();
properties.put("cluster.name", "elasticsearch");
esConnection = new ESConnection(hosts, properties, ESConnection.ESClientMode.REST);
}
@Test
public void test01() {
MappingMetaData mappingMetaData = esConnection.getMapping("mytest_user");
Map<String, Object> sourceMap = mappingMetaData.getSourceAsMap();
Map<String, Object> esMapping = (Map<String, Object>) sourceMap.get("properties");
for (Map.Entry<String, Object> entry : esMapping.entrySet()) {
Map<String, Object> value = (Map<String, Object>) entry.getValue();
if (value.containsKey("properties")) {
System.out.println(entry.getKey() + " object");
} else {
System.out.println(entry.getKey() + " " + value.get("type"));
Assert.notNull(entry.getKey(), "null column name");
Assert.notNull(value.get("type"), "null column type");
}
}
}
}

40
client-adapter/es7x/src/test/java/com/alibaba/otter/canal/client/adapter/es7x/test/TestConstant.java Normal file
View File

@@ -0,0 +1,40 @@
package com.alibaba.otter.canal.client.adapter.es7x.test;
import java.sql.SQLException;
import com.alibaba.druid.pool.DruidDataSource;
public class TestConstant {
public final static String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true";
public final static String jdbcUser = "root";
public final static String jdbcPassword = "121212";
public final static String esHosts = "127.0.0.1:9300";
public final static String clusterName = "elasticsearch";
public final static DruidDataSource dataSource;
static {
dataSource = new DruidDataSource();
dataSource.setDriverClassName("com.mysql.jdbc.Driver");
dataSource.setUrl(jdbcUrl);
dataSource.setUsername(jdbcUser);
dataSource.setPassword(jdbcPassword);
dataSource.setInitialSize(1);
dataSource.setMinIdle(1);
dataSource.setMaxActive(1);
dataSource.setMaxWait(60000);
dataSource.setTimeBetweenEvictionRunsMillis(60000);
dataSource.setMinEvictableIdleTimeMillis(300000);
dataSource.setPoolPreparedStatements(false);
dataSource.setMaxPoolPreparedStatementPerConnectionSize(20);
dataSource.setValidationQuery("select 1");
try {
dataSource.init();
} catch (SQLException e) {
e.printStackTrace();
}
}
}