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

Fix es bulk (#1743)

Browse Source 
thanks
This commit is contained in:
Jet
2019-04-25 09:36:09 +08:00
committed by rewerma
parent e94ee64316
commit 8f088cddc0
2 changed files with 12 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -459,7 +459,7 @@ public class ESSyncService {
sql = ESSyncUtil.appendCondition(sql, condition); sql = ESSyncUtil.appendCondition(sql, condition);
DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Main table insert ot es index by query sql, destination:{}, table: {}, index: {}, sql: {}", logger.trace("Main table insert to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.get_index(),
@@ -473,7 +473,7 @@ public class ESSyncService {
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace( logger.trace(
"Main table insert ot es index by query sql, destination:{}, table: {}, index: {}, id: {}", "Main table insert to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.get_index(),
@@ -517,7 +517,7 @@ public class ESSyncService {
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace( logger.trace(
"Main table delete ot es index by query sql, destination:{}, table: {}, index: {}, id: {}", "Main table delete to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.get_index(),
@@ -784,7 +784,7 @@ public class ESSyncService {
Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData); Object idVal = esTemplate.getESDataFromDmlData(mapping, data, old, esFieldData);
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Main table update ot es index, destination:{}, table: {}, index: {}, id: {}", logger.trace("Main table update to es index, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.get_index(),
@@ -807,7 +807,7 @@ public class ESSyncService {
sql = ESSyncUtil.appendCondition(sql, condition); sql = ESSyncUtil.appendCondition(sql, condition);
DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); DataSource ds = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace("Main table update ot es index by query sql, destination:{}, table: {}, index: {}, sql: {}", logger.trace("Main table update to es index by query sql, destination:{}, table: {}, index: {}, sql: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.get_index(),
@@ -821,7 +821,7 @@ public class ESSyncService {
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace( logger.trace(
"Main table update ot es index by query sql, destination:{}, table: {}, index: {}, id: {}", "Main table update to es index by query sql, destination:{}, table: {}, index: {}, id: {}",
config.getDestination(), config.getDestination(),
dml.getTable(), dml.getTable(),
mapping.get_index(), mapping.get_index(),

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

@@ -58,6 +58,10 @@ public class ESTemplate {
return bulkRequestBuilder; return bulkRequestBuilder;
} }
public void resetBulkRequestBuilder() {
this.bulkRequestBuilder = this.transportClient.prepareBulk();
}
/** /**
* 插入数据 * 插入数据
* *
@@ -210,11 +214,12 @@ public class ESTemplate {
} }
/** /**
* 如果大于批量数则提交批次 * 如果大于批量数则提交批次, 调用后es bulk请求后，numberOfActions不会清理，需要主动调用函数清0，否则不能起到批量请求的效果
*/ */
private void commitBulk() { private void commitBulk() {
if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) { if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {
commit(); commit();
resetBulkRequestBuilder();
} }
} }