chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

ES adapter代码重新整理

Browse Source 
新增upsert属性
This commit is contained in:
mcy
2019-02-13 15:19:09 +08:00
parent aba8df699c
commit 83b760494d
7 changed files with 545 additions and 674 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
View File

@@ -146,9 +146,10 @@ public class ESAdapter implements OuterAdapter {
for (Dml dml : dmls) { for (Dml dml : dmls) {
sync(dml); sync(dml);
} }
esSyncService.commit(); // 批次统一提交
} }
public void sync(Dml dml) { private void sync(Dml dml) {
String database = dml.getDatabase(); String database = dml.getDatabase();
String table = dml.getTable(); String table = dml.getTable();
Map<String, ESSyncConfig> configMap; Map<String, ESSyncConfig> configMap;
@@ -174,7 +175,7 @@ public class ESAdapter implements OuterAdapter {
DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
ESEtlService esEtlService = new ESEtlService(transportClient, config); ESEtlService esEtlService = new ESEtlService(transportClient, config);
if (dataSource != null) { if (dataSource != null) {
return esEtlService.importData(params, false); return esEtlService.importData(params);
} else { } else {
etlResult.setSucceeded(false); etlResult.setSucceeded(false);
etlResult.setErrorMessage("DataSource not found"); etlResult.setErrorMessage("DataSource not found");
@@ -188,7 +189,7 @@ public class ESAdapter implements OuterAdapter {
// 取所有的destination为task的配置 // 取所有的destination为task的配置
if (configTmp.getDestination().equals(task)) { if (configTmp.getDestination().equals(task)) {
ESEtlService esEtlService = new ESEtlService(transportClient, configTmp); ESEtlService esEtlService = new ESEtlService(transportClient, configTmp);
EtlResult etlRes = esEtlService.importData(params, false); EtlResult etlRes = esEtlService.importData(params);
if (!etlRes.getSucceeded()) { if (!etlRes.getSucceeded()) {
resSuccess = false; resSuccess = false;
resultMsg.append(etlRes.getErrorMessage()).append("\n"); resultMsg.append(etlRes.getErrorMessage()).append("\n");

23
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java
View File

@@ -30,8 +30,8 @@ public class ESSyncConfig {
if (esMapping._type == null) { if (esMapping._type == null) {
throw new NullPointerException("esMapping._type"); throw new NullPointerException("esMapping._type");
} }
if (esMapping._id == null && esMapping.pk == null) { if (esMapping._id == null) {
throw new NullPointerException("esMapping._id and esMapping.pk"); throw new NullPointerException("esMapping._id");
} }
if (esMapping.sql == null) { if (esMapping.sql == null) {
throw new NullPointerException("esMapping.sql"); throw new NullPointerException("esMapping.sql");
@@ -83,8 +83,9 @@ public class ESSyncConfig {
private String _index; private String _index;
private String _type; private String _type;
private String _id; private String _id;
private boolean upsert = false;
private String pk; private String pk;
private String parent; // private String parent;
private String sql; private String sql;
// 对象字段, 例: objFields: // 对象字段, 例: objFields:
// - _labels: array:; // - _labels: array:;
@@ -121,6 +122,14 @@ public class ESSyncConfig {
this._id = _id; this._id = _id;
} }
public boolean isUpsert() {
return upsert;
}
public void setUpsert(boolean upsert) {
this.upsert = upsert;
}
public String getPk() { public String getPk() {
return pk; return pk;
} }
@@ -129,14 +138,6 @@ public class ESSyncConfig {
this.pk = pk; this.pk = pk;
} }
public String getParent() {
return parent;
}
public void setParent(String parent) {
this.parent = parent;
}
public Map<String, String> getObjFields() { public Map<String, String> getObjFields() {
return objFields; return objFields;
} }

10
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/SchemaItem.java
View File

@@ -134,11 +134,11 @@ public class SchemaItem {
} }
public FieldItem getIdFieldItem(ESMapping mapping) { public FieldItem getIdFieldItem(ESMapping mapping) {
if (mapping.get_id() != null) { //TODO if (mapping.get_id() != null) {
return getSelectFields().get(mapping.get_id()); return getSelectFields().get(mapping.get_id());
} else { // } else {
return getSelectFields().get(mapping.getPk()); // return getSelectFields().get(mapping.getPk());
} // }
} }
public static class TableItem { public static class TableItem {

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

@@ -1,6 +1,9 @@
package com.alibaba.otter.canal.client.adapter.es.service; package com.alibaba.otter.canal.client.adapter.es.service;
import java.util.*; import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.Future;
@@ -13,11 +16,8 @@ import javax.sql.DataSource;
import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -51,7 +51,7 @@ public class ESEtlService {
this.config = config; this.config = config;
} }
public EtlResult importData(List<String> params, boolean bulk) { public EtlResult importData(List<String> params) {
EtlResult etlResult = new EtlResult(); EtlResult etlResult = new EtlResult();
AtomicLong impCount = new AtomicLong(); AtomicLong impCount = new AtomicLong();
List<String> errMsg = new ArrayList<>(); List<String> errMsg = new ArrayList<>();
@@ -93,54 +93,49 @@ public class ESEtlService {
logger.debug("etl sql : {}", mapping.getSql()); logger.debug("etl sql : {}", mapping.getSql());
} }
if (bulk) { // 获取总数
// 获取总数 String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT ";
String countSql = "SELECT COUNT(1) FROM ( " + sql + ") _CNT "; long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> {
long cnt = (Long) ESSyncUtil.sqlRS(dataSource, countSql, rs -> { Long count = null;
Long count = null; try {
try { if (rs.next()) {
if (rs.next()) { count = ((Number) rs.getObject(1)).longValue();
count = ((Number) rs.getObject(1)).longValue();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
} }
return count == null ? 0L : count; } catch (Exception e) {
}); logger.error(e.getMessage(), e);
// 当大于1万条记录时开启多线程
if (cnt >= 10000) {
int threadCount = 3; // TODO 从配置读取默认为3
long perThreadCnt = cnt / threadCount;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
List<Future<Boolean>> futures = new ArrayList<>(threadCount);
for (int i = 0; i < threadCount; i++) {
long offset = i * perThreadCnt;
Long size = null;
if (i != threadCount - 1) {
size = perThreadCnt;
}
String sqlFinal;
if (size != null) {
sqlFinal = sql + " LIMIT " + offset + "," + size;
} else {
sqlFinal = sql + " LIMIT " + offset + "," + cnt;
}
Future<Boolean> future = executor
.submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
futures.add(future);
}
for (Future<Boolean> future : futures) {
future.get();
}
executor.shutdown();
} else {
executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
} }
return count == null ? 0L : count;
});
// 当大于1万条记录时开启多线程
if (cnt >= 10000) {
int threadCount = 3; // 从配置读取默认为3
long perThreadCnt = cnt / threadCount;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
List<Future<Boolean>> futures = new ArrayList<>(threadCount);
for (int i = 0; i < threadCount; i++) {
long offset = i * perThreadCnt;
Long size = null;
if (i != threadCount - 1) {
size = perThreadCnt;
}
String sqlFinal;
if (size != null) {
sqlFinal = sql + " LIMIT " + offset + "," + size;
} else {
sqlFinal = sql + " LIMIT " + offset + "," + cnt;
}
Future<Boolean> future = executor
.submit(() -> executeSqlImport(dataSource, sqlFinal, mapping, impCount, errMsg));
futures.add(future);
}
for (Future<Boolean> future : futures) {
future.get();
}
executor.shutdown();
} else { } else {
logger.info("自动ETL，无需统计记录总条数，直接进行ETL, index: {}", esIndex);
executeSqlImport(dataSource, sql, mapping, impCount, errMsg); executeSqlImport(dataSource, sql, mapping, impCount, errMsg);
} }
@@ -205,32 +200,35 @@ public class ESEtlService {
} }
if (idVal != null) { if (idVal != null) {
if (mapping.getParent() == null) { if (mapping.isUpsert()) {
bulkRequestBuilder.add(transportClient
.prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString())
.setDoc(esFieldData)
.setDocAsUpsert(true));
} else {
bulkRequestBuilder.add(transportClient bulkRequestBuilder.add(transportClient
.prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString()) .prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString())
.setSource(esFieldData)); .setSource(esFieldData));
} else {
// ignore
} }
} else { } else {
idVal = rs.getObject(mapping.getPk()); // TODO idVal = rs.getObject(mapping.getPk());
if (mapping.getParent() == null) { // if (mapping.getParent() == null) {
// 删除pk对应的数据 // // 删除pk对应的数据
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) // SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type()) // .setTypes(mapping.get_type())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) // .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal))
.get(); // .get();
for (SearchHit hit : response.getHits()) { // for (SearchHit hit : response.getHits()) {
bulkRequestBuilder.add(transportClient // bulkRequestBuilder.add(transportClient
.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); // .prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId()));
} // }
//
bulkRequestBuilder // bulkRequestBuilder
.add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()) // .add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type())
.setSource(esFieldData)); // .setSource(esFieldData));
} else { // } else {
// ignore // // ignore
} // }
} }
if (bulkRequestBuilder.numberOfActions() % mapping.getCommitBatch() == 0 if (bulkRequestBuilder.numberOfActions() % mapping.getCommitBatch() == 0
@@ -238,11 +236,11 @@ public class ESEtlService {
long esBatchBegin = System.currentTimeMillis(); long esBatchBegin = System.currentTimeMillis();
BulkResponse rp = bulkRequestBuilder.execute().actionGet(); BulkResponse rp = bulkRequestBuilder.execute().actionGet();
if (rp.hasFailures()) { if (rp.hasFailures()) {
this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent())); this.processFailBulkResponse(rp, /* TODO Objects.nonNull(mapping.getParent()) */ false);
} }
if (logger.isDebugEnabled()) { if (logger.isTraceEnabled()) {
logger.debug("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
bulkRequestBuilder.numberOfActions(), bulkRequestBuilder.numberOfActions(),
@@ -259,10 +257,10 @@ public class ESEtlService {
long esBatchBegin = System.currentTimeMillis(); long esBatchBegin = System.currentTimeMillis();
BulkResponse rp = bulkRequestBuilder.execute().actionGet(); BulkResponse rp = bulkRequestBuilder.execute().actionGet();
if (rp.hasFailures()) { if (rp.hasFailures()) {
this.processFailBulkResponse(rp, Objects.nonNull(mapping.getParent())); this.processFailBulkResponse(rp, /* TODO Objects.nonNull(mapping.getParent()) */false);
} }
if (logger.isDebugEnabled()) { if (logger.isTraceEnabled()) {
logger.debug("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}",
(System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - batchBegin),
(System.currentTimeMillis() - esBatchBegin), (System.currentTimeMillis() - esBatchBegin),
bulkRequestBuilder.numberOfActions(), bulkRequestBuilder.numberOfActions(),

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

@@ -203,7 +203,7 @@ public class ESSyncService {
// ------主表 查询sql来更新------ // ------主表 查询sql来更新------
if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) { if (schemaItem.getMainTable().getTableName().equalsIgnoreCase(dml.getTable())) {
ESMapping mapping = config.getEsMapping(); ESMapping mapping = config.getEsMapping();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id();
FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName); FieldItem idFieldItem = schemaItem.getSelectFields().get(idFieldName);
boolean idFieldSimple = true; boolean idFieldSimple = true;
@@ -344,14 +344,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.delete(mapping, idVal); esTemplate.delete(mapping, idVal);
if (!result) {
logger.error("Main table delete es index error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} else { } else {
// ------主键带函数, 查询sql获取主键删除------ // ------主键带函数, 查询sql获取主键删除------
mainTableDelete(config, dml, data); mainTableDelete(config, dml, data);
@@ -417,14 +410,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.insert(mapping, idVal, esFieldData); esTemplate.insert(mapping, idVal, esFieldData);
if (!result) {
logger.error("Single table insert to es index error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
/** /**
@@ -461,15 +447,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.insert(mapping, idVal, esFieldData); esTemplate.insert(mapping, idVal, esFieldData);
if (!result) {
logger.error(
"Main table insert to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -504,15 +482,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.delete(mapping, idVal); esTemplate.delete(mapping, idVal);
if (!result) {
logger.error(
"Main table delete to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -799,14 +769,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.update(mapping, idVal, esFieldData); esTemplate.update(mapping, idVal, esFieldData);
if (!result) {
logger.error("Main table update to es index error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
/** /**
@@ -843,15 +806,7 @@ public class ESSyncService {
mapping.get_index(), mapping.get_index(),
idVal); idVal);
} }
boolean result = esTemplate.update(mapping, idVal, esFieldData); esTemplate.update(mapping, idVal, esFieldData);
if (!result) {
logger.error(
"Main table update to es index by query sql error, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(),
dml.getTable(),
mapping.get_index(),
idVal);
}
} }
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
@@ -859,4 +814,11 @@ public class ESSyncService {
return 0; return 0;
}); });
} }
/**
* 提交批次
*/
public void commit() {
esTemplate.commit();
}
} }

960
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
View File

@@ -1,526 +1,434 @@
package com.alibaba.otter.canal.client.adapter.es.support; package com.alibaba.otter.canal.client.adapter.es.support;
import java.sql.ResultSet; import java.sql.ResultSet;
import java.sql.SQLException; import java.sql.SQLException;
import java.util.*; import java.util.LinkedHashMap;
import java.util.concurrent.ConcurrentHashMap; import java.util.Map;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit; import java.util.concurrent.ConcurrentMap;
import javax.sql.DataSource; import javax.sql.DataSource;
import com.alibaba.fastjson.JSON; import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkItemResponse; import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.cluster.metadata.MappingMetaData; import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.index.query.QueryBuilder; import org.slf4j.Logger;
import org.elasticsearch.index.query.QueryBuilders; import org.slf4j.LoggerFactory;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryAction; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig;
import org.elasticsearch.index.reindex.UpdateByQueryRequestBuilder; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping;
import org.elasticsearch.rest.RestStatus; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem;
import org.elasticsearch.script.Script; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem;
import org.elasticsearch.script.ScriptType; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
import org.elasticsearch.search.SearchHit; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; /**
import org.springframework.util.CollectionUtils; * ES 操作模板
*
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; * @author rewerma 2018-11-01
import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; * @version 1.0.0
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; */
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem; public class ESTemplate {
import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class);
/** private static final int MAX_BATCH_SIZE = 1000;
* ES 操作模板
* private TransportClient transportClient;
* @author rewerma 2018-11-01
* @version 1.0.0 private volatile BulkRequestBuilder bulkRequestBuilder;
*/
public class ESTemplate { public ESTemplate(TransportClient transportClient){
this.transportClient = transportClient;
private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class); this.bulkRequestBuilder = transportClient.prepareBulk();
}
private static final int MAX_BATCH_SIZE = 1000;
public BulkRequestBuilder getBulk() {
private TransportClient transportClient; return bulkRequestBuilder;
}
public ESTemplate(TransportClient transportClient){
this.transportClient = transportClient; /**
} * 插入数据
*
/** * @param mapping
* 插入数据 * @param pkVal
* * @param esFieldData
* @param mapping * @return
* @param pkVal */
* @param esFieldData public void insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
* @return if (mapping.get_id() != null) {
*/ if (mapping.isUpsert()) {
public boolean insert(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); .setDoc(esFieldData)
if (mapping.get_id() != null) { .setDocAsUpsert(true));
bulkRequestBuilder } else {
.add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setSource(esFieldData)); .setSource(esFieldData));
} else { }
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) commitBatch();
.setTypes(mapping.get_type()) } else {
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) // TODO SearchResponse response =
.setSize(MAX_BATCH_SIZE) // transportClient.prepareSearch(mapping.get_index())
.get(); // .setTypes(mapping.get_type())
for (SearchHit hit : response.getHits()) { // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
bulkRequestBuilder // .setSize(MAX_BATCH_SIZE)
.add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); // .get();
} // for (SearchHit hit : response.getHits()) {
bulkRequestBuilder // bulkRequestBuilder
.add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type()).setSource(esFieldData)); // .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(),
} // hit.getId()));
return commitBulkRequest(bulkRequestBuilder); // }
} // bulkRequestBuilder
// .add(transportClient.prepareIndex(mapping.get_index(),
/** // mapping.get_type()).setSource(esFieldData));
* 根据主键更新数据 }
* // return commitBulkRequest(bulkRequestBuilder);
* @param mapping }
* @param pkVal
* @param esFieldData /**
* @return * 根据主键更新数据
*/ *
public boolean update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { * @param mapping
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); * @param pkVal
append4Update(bulkRequestBuilder, mapping, pkVal, esFieldData); * @param esFieldData
return commitBulkRequest(bulkRequestBuilder); * @return
} */
public void update(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
public void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal, append4Update(getBulk(), mapping, pkVal, esFieldData);
Map<String, Object> esFieldData) { commitBatch();
if (mapping.get_id() != null) { }
bulkRequestBuilder
.add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) private void append4Update(BulkRequestBuilder bulkRequestBuilder, ESMapping mapping, Object pkVal,
.setDoc(esFieldData)); Map<String, Object> esFieldData) {
} else { if (mapping.get_id() != null) {
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) bulkRequestBuilder
.setTypes(mapping.get_type()) .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) .setDoc(esFieldData));
.setSize(MAX_BATCH_SIZE) } else {
.get(); // TODO SearchResponse response =
for (SearchHit hit : response.getHits()) { // transportClient.prepareSearch(mapping.get_index())
bulkRequestBuilder // .setTypes(mapping.get_type())
.add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.setDoc(esFieldData)); // .setSize(MAX_BATCH_SIZE)
} // .get();
} // for (SearchHit hit : response.getHits()) {
} // bulkRequestBuilder
// .add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(),
/** // hit.getId())
* update by query // .setDoc(esFieldData));
* // }
* @param config }
* @param paramsTmp }
* @param esFieldData
* @return /**
*/ * TODO XXX update by query
public boolean updateByQuery(ESSyncConfig config, Map<String, Object> paramsTmp, Map<String, Object> esFieldData) { *
if (paramsTmp.isEmpty()) { * @param config
return false; * @param paramsTmp
} * @param esFieldData
ESMapping mapping = config.getEsMapping(); * @return
BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery(); */
paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value))); public boolean updateByQuery(ESSyncConfig config, Map<String, Object> paramsTmp, Map<String, Object> esFieldData) {
if (paramsTmp.isEmpty()) {
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) return false;
.setTypes(mapping.get_type()) }
.setSize(0) ESMapping mapping = config.getEsMapping();
.setQuery(queryBuilder) BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
.get(); paramsTmp.forEach((fieldName, value) -> queryBuilder.must(QueryBuilders.termsQuery(fieldName, value)));
long count = response.getHits().getTotalHits();
// 如果更新量大于Max, 查询sql批量更新 // 查询sql批量更新
if (count > MAX_BATCH_SIZE) { DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE ");
paramsTmp.forEach(
DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); (fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND "));
// 查询sql更新 int len = sql.length();
StringBuilder sql = new StringBuilder("SELECT * FROM (" + mapping.getSql() + ") _v WHERE "); sql.delete(len - 4, len);
paramsTmp.forEach( Integer syncCount = (Integer) ESSyncUtil.sqlRS(ds, sql.toString(), rs -> {
(fieldName, value) -> sql.append("_v.").append(fieldName).append("=").append(value).append(" AND ")); int count = 0;
int len = sql.length(); try {
sql.delete(len - 4, len); while (rs.next()) {
ESSyncUtil.sqlRS(ds, sql.toString(), rs -> { Object idVal = getIdValFromRS(mapping, rs);
int exeCount = 1; append4Update(getBulk(), mapping, idVal, esFieldData);
try { commitBatch();
BulkRequestBuilder bulkRequestBuilderTmp = bulkRequestBuilder; count++;
while (rs.next()) { }
Object idVal = getIdValFromRS(mapping, rs); } catch (Exception e) {
append4Update(bulkRequestBuilderTmp, mapping, idVal, esFieldData); throw new RuntimeException(e);
}
if (exeCount % mapping.getCommitBatch() == 0 && bulkRequestBuilderTmp.numberOfActions() > 0) { return count;
commitBulkRequest(bulkRequestBuilderTmp); });
bulkRequestBuilderTmp = transportClient.prepareBulk(); if (logger.isTraceEnabled()) {
} logger.trace("Update ES by query effect {} records", syncCount);
exeCount++; }
} return true;
}
if (bulkRequestBuilder.numberOfActions() > 0) {
commitBulkRequest(bulkRequestBuilderTmp); /**
} * 通过主键删除数据
} catch (Exception e) { *
throw new RuntimeException(e); * @param mapping
} * @param pkVal
return 0; * @return
}); */
return true; public void delete(ESMapping mapping, Object pkVal) {
} else { if (mapping.get_id() != null) {
return updateByQuery(mapping, queryBuilder, esFieldData, 1); getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString()));
} commitBatch();
} } else {
// TODO SearchResponse response =
private boolean updateByQuery(ESMapping mapping, QueryBuilder queryBuilder, Map<String, Object> esFieldData, // transportClient.prepareSearch(mapping.get_index())
int counter) { // .setTypes(mapping.get_type())
if (CollectionUtils.isEmpty(esFieldData)) { // .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
return true; // .setSize(MAX_BATCH_SIZE)
} // .get();
// for (SearchHit hit : response.getHits()) {
StringBuilder sb = new StringBuilder(); // bulkRequestBuilder
esFieldData.forEach((key, value) -> { // .add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(),
if (value instanceof Map) { // hit.getId()));
Map<?, ?> mapValue = (Map<?, ?>) value; // }
if (mapValue.containsKey("lon") && mapValue.containsKey("lat") && mapValue.size() == 2) { }
sb.append("ctx._source") // return commitBulkRequest(bulkRequestBuilder);
.append("['") }
.append(key)
.append("']") /**
.append(" = [") * 提交批次
.append(mapValue.get("lon")) */
.append(", ") public void commit() {
.append(mapValue.get("lat")) if (getBulk().numberOfActions() >= 0) {
.append("];"); BulkResponse response = getBulk().execute().actionGet();
} else { if (response.hasFailures()) {
sb.append("ctx._source").append("[\"").append(key).append("\"]").append(" = "); for (BulkItemResponse itemResponse : response.getItems()) {
sb.append(JSON.toJSONString(value)); if (!itemResponse.isFailed()) {
sb.append(";"); continue;
} }
} else if (value instanceof List) {
sb.append("ctx._source").append("[\"").append(key).append("\"]").append(" = "); if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) {
sb.append(JSON.toJSONString(value)); logger.error(itemResponse.getFailureMessage());
sb.append(";"); } else {
} else if (value instanceof String) { throw new RuntimeException("ES sync commit error" + itemResponse.getFailureMessage());
sb.append("ctx._source") }
.append("['") }
.append(key) }
.append("']") }
.append(" = '") }
.append(value)
.append("';"); /**
} else { * 如果大于批量则提交批次
sb.append("ctx._source").append("['").append(key).append("']").append(" = ").append(value).append(";"); */
} private void commitBatch() {
}); if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {
String scriptLine = sb.toString(); commit();
if (logger.isTraceEnabled()) { }
logger.trace(scriptLine); }
}
public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(transportClient); String columnName) throws SQLException {
updateByQuery.source(mapping.get_index()) String esType = getEsType(mapping, fieldName);
.abortOnVersionConflict(false)
.filter(queryBuilder) Object value = resultSet.getObject(columnName);
.script(new Script(ScriptType.INLINE, "painless", scriptLine, Collections.emptyMap())); if (value instanceof Boolean) {
if (!"boolean".equals(esType)) {
BulkByScrollResponse response = updateByQuery.get(); value = resultSet.getByte(columnName);
if (logger.isTraceEnabled()) { }
logger.trace("updateByQuery response: {}", response.getStatus()); }
}
if (!CollectionUtils.isEmpty(response.getSearchFailures())) { // 如果是对象类型
logger.error("script update_for_search has search error: " + response.getBulkFailures()); if (mapping.getObjFields().containsKey(fieldName)) {
return false; return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName));
} } else {
return ESSyncUtil.typeConvert(value, esType);
if (!CollectionUtils.isEmpty(response.getBulkFailures())) { }
logger.error("script update_for_search has update error: " + response.getBulkFailures()); }
return false;
} public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet,
Map<String, Object> esFieldData) throws SQLException {
if (response.getStatus().getVersionConflicts() > 0) { SchemaItem schemaItem = mapping.getSchemaItem();
if (counter >= 3) { String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id();
logger.error("第 {} 次执行updateByQuery, 依旧存在分片版本冲突，不再继续重试。", counter); Object resultIdVal = null;
return false; for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
} Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
logger.warn("本次updateByQuery存在分片版本冲突，准备重新执行...");
try { if (fieldItem.getFieldName().equals(idFieldName)) {
TimeUnit.SECONDS.sleep(1); resultIdVal = value;
} catch (InterruptedException e) { }
// ignore
} if (!fieldItem.getFieldName().equals(mapping.get_id())
return updateByQuery(mapping, queryBuilder, esFieldData, ++counter); && !mapping.getSkips().contains(fieldItem.getFieldName())) {
} esFieldData.put(fieldItem.getFieldName(), value);
}
return true; }
} return resultIdVal;
}
/**
* 通过主键删除数据 public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException {
* SchemaItem schemaItem = mapping.getSchemaItem();
* @param mapping String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id();
* @param pkVal Object resultIdVal = null;
* @return for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
*/ Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
public boolean delete(ESMapping mapping, Object pkVal) {
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); if (fieldItem.getFieldName().equals(idFieldName)) {
if (mapping.get_id() != null) { resultIdVal = value;
bulkRequestBuilder break;
.add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); }
} else { }
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) return resultIdVal;
.setTypes(mapping.get_type()) }
.setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal))
.setSize(MAX_BATCH_SIZE) public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld,
.get(); Map<String, Object> esFieldData) throws SQLException {
for (SearchHit hit : response.getHits()) { SchemaItem schemaItem = mapping.getSchemaItem();
bulkRequestBuilder String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id();
.add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), hit.getId())); Object resultIdVal = null;
} for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
} if (fieldItem.getFieldName().equals(idFieldName)) {
return commitBulkRequest(bulkRequestBuilder); resultIdVal = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName());
} }
/** for (ColumnItem columnItem : fieldItem.getColumnItems()) {
* 批量提交 if (dmlOld.containsKey(columnItem.getColumnName())
* && !mapping.getSkips().contains(fieldItem.getFieldName())) {
* @param bulkRequestBuilder esFieldData.put(fieldItem.getFieldName(),
* @return getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()));
*/ break;
private static boolean commitBulkRequest(BulkRequestBuilder bulkRequestBuilder) { }
if (bulkRequestBuilder.numberOfActions() > 0) { }
BulkResponse response = bulkRequestBuilder.execute().actionGet(); }
if (response.hasFailures()) { return resultIdVal;
for (BulkItemResponse itemResponse : response.getItems()) { }
if (!itemResponse.isFailed()) {
continue; public Object getValFromData(ESMapping mapping, Map<String, Object> dmlData, String fieldName, String columnName) {
} String esType = getEsType(mapping, fieldName);
Object value = dmlData.get(columnName);
if (itemResponse.getFailure().getStatus() == RestStatus.NOT_FOUND) { if (value instanceof Byte) {
logger.warn(itemResponse.getFailureMessage()); if ("boolean".equals(esType)) {
} else { value = ((Byte) value).intValue() != 0;
logger.error("ES sync commit error: {}", itemResponse.getFailureMessage()); }
} }
}
} // 如果是对象类型
if (mapping.getObjFields().containsKey(fieldName)) {
return !response.hasFailures(); return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName));
} } else {
return true; return ESSyncUtil.typeConvert(value, esType);
} }
}
public Object getValFromRS(ESMapping mapping, ResultSet resultSet, String fieldName,
String columnName) throws SQLException { /**
String esType = getEsType(mapping, fieldName); * 将dml的data转换为es的data
*
Object value = resultSet.getObject(columnName); * @param mapping 配置mapping
if (value instanceof Boolean) { * @param dmlData dml data
if (!"boolean".equals(esType)) { * @param esFieldData es data
value = resultSet.getByte(columnName); * @return 返回 id 值
} */
} public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData,
Map<String, Object> esFieldData) {
// 如果是对象类型 SchemaItem schemaItem = mapping.getSchemaItem();
if (mapping.getObjFields().containsKey(fieldName)) { String idFieldName = mapping.get_id(); // TODO == null ? mapping.getPk() : mapping.get_id();
return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName)); Object resultIdVal = null;
} else { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
return ESSyncUtil.typeConvert(value, esType); String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
} Object value = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
}
if (fieldItem.getFieldName().equals(idFieldName)) {
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, resultIdVal = value;
Map<String, Object> esFieldData) throws SQLException { }
SchemaItem schemaItem = mapping.getSchemaItem();
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); if (!fieldItem.getFieldName().equals(mapping.get_id())
Object resultIdVal = null; && !mapping.getSkips().contains(fieldItem.getFieldName())) {
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { esFieldData.put(fieldItem.getFieldName(), value);
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); }
}
if (fieldItem.getFieldName().equals(idFieldName)) { return resultIdVal;
resultIdVal = value; }
}
/**
if (!fieldItem.getFieldName().equals(mapping.get_id()) * 将dml的data, old转换为es的data
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { *
esFieldData.put(fieldItem.getFieldName(), value); * @param mapping 配置mapping
} * @param dmlData dml data
} * @param esFieldData es data
return resultIdVal; * @return 返回 id 值
} */
public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, Map<String, Object> dmlOld,
public Object getIdValFromRS(ESMapping mapping, ResultSet resultSet) throws SQLException { 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(); // TODO == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null; Object resultIdVal = null;
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { for (FieldItem fieldItem : schemaItem.getSelectFields().values()) {
Object value = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); String columnName = fieldItem.getColumnItems().iterator().next().getColumnName();
if (fieldItem.getFieldName().equals(idFieldName)) { if (fieldItem.getFieldName().equals(idFieldName)) {
resultIdVal = value; resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
break; }
}
} if (dmlOld.get(columnName) != null && !mapping.getSkips().contains(fieldItem.getFieldName())) {
return resultIdVal; esFieldData.put(fieldItem.getFieldName(),
} getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
}
public Object getESDataFromRS(ESMapping mapping, ResultSet resultSet, Map<String, Object> dmlOld, }
Map<String, Object> esFieldData) throws SQLException { return resultIdVal;
SchemaItem schemaItem = mapping.getSchemaItem(); }
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id();
Object resultIdVal = null; /**
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { * es 字段类型本地缓存
if (fieldItem.getFieldName().equals(idFieldName)) { */
resultIdVal = getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName()); private static ConcurrentMap<String, Map<String, String>> esFieldTypes = new ConcurrentHashMap<>();
}
/**
for (ColumnItem columnItem : fieldItem.getColumnItems()) { * 获取es mapping中的属性类型
if (dmlOld.containsKey(columnItem.getColumnName()) *
&& !mapping.getSkips().contains(fieldItem.getFieldName())) { * @param mapping mapping配置
esFieldData.put(fieldItem.getFieldName(), * @param fieldName 属性名
getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); * @return 类型
break; */
} @SuppressWarnings("unchecked")
} private String getEsType(ESMapping mapping, String fieldName) {
} String key = mapping.get_index() + "-" + mapping.get_type();
return resultIdVal; Map<String, String> fieldType = esFieldTypes.get(key);
} if (fieldType == null) {
ImmutableOpenMap<String, MappingMetaData> mappings;
public Object getValFromData(ESMapping mapping, Map<String, Object> dmlData, String fieldName, String columnName) { try {
String esType = getEsType(mapping, fieldName); mappings = transportClient.admin()
Object value = dmlData.get(columnName); .cluster()
if (value instanceof Byte) { .prepareState()
if ("boolean".equals(esType)) { .execute()
value = ((Byte) value).intValue() != 0; .actionGet()
} .getState()
} .getMetaData()
.getIndices()
// 如果是对象类型 .get(mapping.get_index())
if (mapping.getObjFields().containsKey(fieldName)) { .getMappings();
return ESSyncUtil.convertToEsObj(value, mapping.getObjFields().get(fieldName)); } catch (NullPointerException e) {
} else { throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
return ESSyncUtil.typeConvert(value, esType); }
} MappingMetaData mappingMetaData = mappings.get(mapping.get_type());
} if (mappingMetaData == null) {
throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
/** }
* 将dml的data转换为es的data
* fieldType = new LinkedHashMap<>();
* @param mapping 配置mapping
* @param dmlData dml data Map<String, Object> sourceMap = mappingMetaData.getSourceAsMap();
* @param esFieldData es data Map<String, Object> esMapping = (Map<String, Object>) sourceMap.get("properties");
* @return 返回 id 值 for (Map.Entry<String, Object> entry : esMapping.entrySet()) {
*/ Map<String, Object> value = (Map<String, Object>) entry.getValue();
public Object getESDataFromDmlData(ESMapping mapping, Map<String, Object> dmlData, if (value.containsKey("properties")) {
Map<String, Object> esFieldData) { fieldType.put(entry.getKey(), "object");
SchemaItem schemaItem = mapping.getSchemaItem(); } else {
String idFieldName = mapping.get_id() == null ? mapping.getPk() : mapping.get_id(); fieldType.put(entry.getKey(), (String) value.get("type"));
Object resultIdVal = null; }
for (FieldItem fieldItem : schemaItem.getSelectFields().values()) { }
String columnName = fieldItem.getColumnItems().iterator().next().getColumnName(); esFieldTypes.put(key, fieldType);
Object value = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName); }
if (fieldItem.getFieldName().equals(idFieldName)) { return fieldType.get(fieldName);
resultIdVal = value; }
} }
if (!fieldItem.getFieldName().equals(mapping.get_id())
&& !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(), value);
}
}
return resultIdVal;
}
/**
* 将dml的data, old转换为es的data
*
* @param mapping 配置mapping
* @param dmlData dml data
* @param esFieldData es data
* @return 返回 id 值
*/
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.get(columnName) != null && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(),
getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
}
}
return resultIdVal;
}
/**
* es 字段类型本地缓存
*/
private static ConcurrentMap<String, Map<String, String>> esFieldTypes = new ConcurrentHashMap<>();
/**
* 获取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) {
ImmutableOpenMap<String, MappingMetaData> mappings;
try {
mappings = transportClient.admin()
.cluster()
.prepareState()
.execute()
.actionGet()
.getState()
.getMetaData()
.getIndices()
.get(mapping.get_index())
.getMappings();
} catch (NullPointerException e) {
throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index());
}
MappingMetaData mappingMetaData = mappings.get(mapping.get_type());
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);
}
}

1
client-adapter/elasticsearch/src/main/resources/es/mytest_user.yml
View File

@@ -5,6 +5,7 @@ esMapping:
_index: mytest_user _index: mytest_user
_type: _doc _type: _doc
_id: _id _id: _id
upsert: true
# pk: id # pk: id
sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name, sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,
a.c_time as _c_time from user a a.c_time as _c_time from user a