kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】 增加BULK同步增量数据

Browse Source 
(cherry picked from commit 643d830939)
This commit is contained in:
Kris 2025-11-19 14:13:14 +08:00
parent c12fbf263a
commit 8544ba452a
6 changed files with 404 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
View File

@ -114,6 +114,27 @@ public class DataDumpNewJob {
return commonBatchService.dumpBatch(param);
}
/**
* 增量任务（大数据量）
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataDumpIncrementBatchJob")
public ReturnT<String> dataDumpIncrementBatchJob(String paramStr) throws Exception {
log.info("dataDumpIncrementBatchJob execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
return commonBatchService.incrementBatch(param);
}
/**
* 存量任务（BigObject）
*

2
src/main/java/com/celnet/datadump/service/CommonBatchService.java
View File

@ -17,4 +17,6 @@ public interface CommonBatchService {
ReturnT<String> dumpBigObjectTargetV1(SalesforceParam param) throws Exception;
ReturnT<String> incrementBatch(SalesforceParam param) throws Exception;
}

1
src/main/java/com/celnet/datadump/service/DataDumpSpecialService.java
View File

@ -2,6 +2,7 @@ package com.celnet.datadump.service;
import com.celnet.datadump.entity.DataObject;
import com.celnet.datadump.param.DataDumpSpecialParam;
import com.sforce.async.BulkConnection;
import com.sforce.soap.partner.PartnerConnection;
import com.xxl.job.core.biz.model.ReturnT;

352
src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
View File

@ -13,6 +13,7 @@ import com.celnet.datadump.entity.*;
import com.celnet.datadump.enums.FileType;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.global.SystemConfigCode;
import com.celnet.datadump.global.TypeCode;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.param.DataDumpParam;
import com.celnet.datadump.param.DataDumpSpecialParam;
@ -95,7 +96,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
private DataDumpSpecialService dataDumpSpecialService;
@Autowired
private DataBatchHistoryService dataBatchHistoryService;
@Autowired
private DataReportService dataReportService;
// 批次处理记录数，默认为10000
private int batchProcessCount = 10000;
@ -143,6 +145,67 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
}
@Override
public ReturnT<String> incrementBatch(SalesforceParam param) throws Exception {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
if (StringUtils.isNotBlank(param.getApi())) {
List<String> apis = DataUtil.toIdList(param.getApi());
qw.in("name", apis);
}
qw.eq("need_update", true)
.isNotNull("last_update_date");
List<DataObject> list = dataObjectService.list(qw);
if (CollectionUtils.isEmpty(list)) {
return new ReturnT<>(500, ("表" + param.getApi() + "不存在或未开启更新"));
}
List<Future<?>> futures = Lists.newArrayList();
try {
DataReport dataReport = new DataReport();
dataReport.setType(TypeCode.INCREMENT);
dataReport.setApis(list.stream().map(DataObject::getName).collect(Collectors.joining(",")));
dataReportService.save(dataReport);
for (DataObject dataObject : list) {
//无创建时间对象
if (!dataFieldService.hasCreatedDate(dataObject.getName())) {
DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam();
dataDumpSpecialParam.setApi(dataObject.getName());
Future<?> future = getDataBatch(dataDumpSpecialParam, salesforceConnect.createBulkConnect(),dataObject);
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(future);
continue;
}
Future<?> future = salesforceExecutor.execute(() -> {
try {
Date updateTime = new Date();
SalesforceParam salesforceParam = new SalesforceParam();
salesforceParam.setApi(dataObject.getName());
salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate());
salesforceParam.setType(2);
// 更新字段值不为空 按更新字段里的字段校验
if (StringUtils.isNotBlank(dataObject.getUpdateField())) {
salesforceParam.setUpdateField(dataObject.getUpdateField());
}
dumpIncrementDataBatch(salesforceParam);
dataObject.setLastUpdateDate(updateTime);
dataObject.setNeedUpdate( false);
dataObjectService.updateById(dataObject);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, 0, 0);
futures.add(future);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
return ReturnT.SUCCESS;
} catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
throw throwable;
}
}
@Override
public ReturnT<String> dumpBigObject(SalesforceParam param) throws Exception {
try {
@ -1077,6 +1140,167 @@ public class CommonBatchServiceImpl implements CommonBatchService {
return bulkConnect;
}
private BulkConnection dumpIncrementDataBatch(SalesforceParam param) throws Throwable {
String api = param.getApi();
BulkConnection bulkConnect = null;
try {
DataBatchHistory dataBatchHistory = new DataBatchHistory();
dataBatchHistory.setName(api);
dataBatchHistory.setStartDate(new Date());
dataBatchHistory.setSyncStartDate(param.getBeginCreateDate());
if (param.getEndCreateDate() != null) {
dataBatchHistory.setSyncEndDate(DateUtils.addSeconds(param.getEndCreateDate(), -1));
}
dataBatchHistory.setBatch(param.getBatch());
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) {
log.info("NO.{} dump {}, date: {} ~ {} start", param.getBatch(),
param.getApi(),
DateFormatUtils.format(param.getBeginCreateDate(), "yyyy-MM-dd HH:mm:ss"),
DateFormatUtils.format(param.getEndCreateDate(), "yyyy-MM-dd HH:mm:ss"));
}
bulkConnect = salesforceConnect.createBulkConnect();
// 存在isDeleted 只查询IsDeleted为false的
if (dataFieldService.hasDeleted(param.getApi())) {
param.setIsDeleted(false);
} else {
// 不存在 过滤
param.setIsDeleted(null);
}
int sfData = getAllBulkV1SfIncrementData(param, bulkConnect);
dataBatchHistory.setDbNum(sfData);
dataBatchHistory.setSfNum(sfData);
updateDataBatch(param, dataBatchHistory);
}catch (InterruptedException e){
return bulkConnect;
} catch (Throwable throwable) {
log.error("dataDumpJob error api:{}", api, throwable);
String type = param.getType() == 1 ? "存量" : "增量";
String format = String.format("%s数据迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", type, api, JSON.toJSONString(param, DataDumpParam.getFilter()), throwable);
EmailUtil.send("DataDump ERROR", format);
throw throwable;
}
return bulkConnect;
}
public Future<?> getDataBatch(DataDumpSpecialParam param, BulkConnection bulkConnect,DataObject dataObject) {
String api = param.getApi();
return salesforceExecutor.execute(() -> {
try {
// 检测表 不存在就生成 但不生成批次
commonService.checkApi(api, true);
int count = 0;
boolean hasMore = true;
QueryWrapper<DataField> wrapper = new QueryWrapper<>();
wrapper.eq("api", api);
List<DataField> list = dataFieldService.list(wrapper);
Map<String, Object> map = Maps.newHashMap();
SalesforceParam salesforceParam = new SalesforceParam();
salesforceParam.setApi(api);
salesforceParam.setIdField(param.getField());
String maxId = null;
List<String> fields = Lists.newArrayList();
for (DataField field : list) {
if (!"base64".equalsIgnoreCase(field.getSfType()) && field.getSfType() != null){
fields.add(field.getField());
}
}
salesforceParam.setSelect(StringUtils.join(fields, ","));
while (hasMore) {
int sfData = 0;
salesforceParam.setMaxId(maxId);
salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate());
salesforceParam.setUpdateField(dataFieldService.returnUpdateDateField(dataObject.getName()));
salesforceParam.setLimit(batchProcessCount);
map.put("param", salesforceParam);
JobInfo job = null;
try {
String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderByIdShare", map);
log.info("query sql: {}", sql);
XxlJobLogger.log("query sql: {}", sql);
job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll);
log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8)));
log.info("创建批次成功, 批次ID: {}", batchInfo.getId());
int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
log.info("批次处理完成, 处理记录数: {}", completionOne);
if (completionOne == 0){
log.info("无更多数据, 结束处理");
BulkUtil.closeJob(bulkConnect, job.getId());
break;
}
List<Map<String, Object>> batchRecords = new ArrayList<>();
QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
for (final String resultId : queryResultList.getResult()) {
InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId);
CSVReader rdr = new CSVReader(resultStream);
List<String> headers = rdr.nextRecord();
log.debug("处理结果ID: {}, 表头数量: {}", resultId, headers.size());
// 批量处理记录
List<String> record;
while ((record = rdr.nextRecord()) != null) {
Map<String, Object> recordMap = new HashMap<>();
for (int i = 0; i < headers.size() && i < record.size(); i++) {
recordMap.put(headers.get(i), record.get(i));
}
batchRecords.add(recordMap);
}
}
if (!batchRecords.isEmpty()){
log.info("开始保存或更新数据, 记录数: {}", batchRecords.size());
sfData = saveOrUpdate(api, batchRecords, true);
count += sfData;
maxId = batchRecords.get(batchRecords.size() - 1).get("Id").toString();
log.info("处理结果: api:{}, sf数量:{}, 当前批次最大Id:{}", api, sfData, maxId);
} else {
log.info("批次无数据记录");
}
BulkUtil.closeJob(bulkConnect, job.getId());
log.info("关闭作业成功, 作业ID: {}", job.getId());
}catch (InterruptedException interruptedException){
return ;
} catch (Exception e) {
log.error("处理查询结果时异常: ", e);
throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown);
}finally {
if (sfData != batchProcessCount){
hasMore = false;
log.info("当前批次数据不足{}条, 结束循环", batchProcessCount);
}
}
}
UpdateWrapper<DataBatch> updateWrapper = new UpdateWrapper<>();
updateWrapper.eq("name", param.getApi());
updateWrapper.set("first_db_num", count);
updateWrapper.set("first_sf_num", count);
dataBatchService.update(updateWrapper);
} catch (Throwable throwable) {
String format = String.format("数据特殊表迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", api, JSON.toJSONString(param), throwable);
EmailUtil.send("DataDump ERROR", format);
log.error("dataDumpSpecial error ", throwable);
throw new RuntimeException(throwable);
}
}, 0, 0);
}
public Future<?> getData(DataDumpSpecialParam param, BulkConnection bulkConnect) {
String api = param.getApi();
return salesforceExecutor.execute(() -> {
@ -1191,6 +1415,132 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}, 0, 0);
}
private int getAllBulkV1SfIncrementData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable {
boolean hasMore = true;
QueryWrapper<DataField> wrapper = new QueryWrapper<>();
String api = param.getApi();
wrapper.eq("api", api);
List<DataField> list = dataFieldService.list(wrapper);
Map<String, Object> map = Maps.newHashMap();
String dateName = param.getType() == 1 ? Const.CREATED_DATE : param.getUpdateField();
int count = 0;
int batch = 0;
Date lastCreatedDate = null;
String maxId = null;
List<String> fields = Lists.newArrayList();
for (DataField field : list) {
if (!"base64".equalsIgnoreCase(field.getSfType()) && field.getSfType() != null){
fields.add(field.getField());
}
}
DataObject dataObject = dataObjectService.getById(api);
if (dataObject != null && StringUtils.isNotBlank(dataObject.getExtraField())) {
fields.addAll(Arrays.asList(StringUtils.split(dataObject.getExtraField().replaceAll(StringUtils.SPACE, StringUtils.EMPTY), ",")));
}
param.setSelect(StringUtils.join(fields, ","));
log.info("开始处理API: {}, 类型: {}", api, param.getType() == 1 ? "按创建时间" : "按修改时间");
while (hasMore) {
int sfData = 0;
param.setMaxId(maxId);
param.setLimit(batchProcessCount);
// 获取创建时间
param.setTimestamp(lastCreatedDate);
map.put("param", param);
String sql;
JobInfo job = null;
try {
if (param.getType() == 1) {
// type 1 按创建时间纬度
sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.list", map);
} else {
// type 2 按修改时间纬度
sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listByModifyTime", map);
}
log.info("query sql: {}", sql);
XxlJobLogger.log("query sql: {}", sql);
DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF");
job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll);
log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8)));
log.info("创建批次成功, 批次ID: {}", batchInfo.getId());
int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
log.info("批次处理完成, 处理记录数: {}", completionOne);
dataLogService.save(dataLog);
if (completionOne == 0){
log.info("无更多数据, 结束处理");
BulkUtil.closeJob(bulkConnect, job.getId());
break;
}
List<Map<String, Object>> batchRecords = new ArrayList<>();
DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB");
QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
for (final String resultId : queryResultList.getResult()) {
InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId);
CSVReader rdr = new CSVReader(resultStream);
List<String> headers = rdr.nextRecord();
log.debug("处理结果ID: {}, 表头数量: {}", resultId, headers.size());
// 批量处理记录
List<String> record;
while ((record = rdr.nextRecord()) != null) {
Map<String, Object> recordMap = new HashMap<>();
for (int i = 0; i < headers.size() && i < record.size(); i++) {
recordMap.put(headers.get(i), record.get(i));
}
batchRecords.add(recordMap);
}
}
dataLogService.save(dataLog1);
if (!batchRecords.isEmpty()){
log.info("开始保存或更新数据, 记录数: {}", batchRecords.size());
sfData = saveOrUpdate(api, batchRecords, true);
count += sfData;
maxId = batchRecords.get(batchRecords.size() - 1).get("Id").toString();
lastCreatedDate = DateUtils.addHours(DateUtils.parseDate((String) batchRecords.get(batchRecords.size() - 1).get("CreatedDate"), Const.SF_DATE_FORMAT), 8);
log.info("处理结果: api:{}, sf数量:{}, 当前批次最大Id:{}, 当前批次最后创建时间:{}", api, sfData, maxId, lastCreatedDate);
} else {
log.info("批次无数据记录");
}
BulkUtil.closeJob(bulkConnect, job.getId());
batch ++;
log.info("关闭作业成功, 作业ID: {}", job.getId());
}catch (InterruptedException interruptedException){
return 0;
} catch (Exception e) {
log.error("处理查询结果时异常: ", e);
throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown);
}finally {
if (sfData != batchProcessCount){
hasMore = false;
log.info("当前批次数据不足{}条, 结束循环", batchProcessCount);
}
}
}
log.info("API: {} 数据处理完成, 总处理记录数: {}", api, count);
return count;
}
private int getAllBulkV1SfData(SalesforceParam param, BulkConnection bulkConnect) throws Throwable {
boolean hasMore = true;

33
src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java
View File

@ -289,8 +289,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
dbQw.and(wrapper -> wrapper.eq("is_createable", 1)
.eq("is_nillable", 0)
.eq("is_defaulted_on_create", 0)
.or().eq("field", "CreatedDate")
.or().eq("field", "CreatedById"));
.or().eq("field", "Id")
.or().eq("field", "CreatedDate")
.or().eq("field", "CreatedById"));
List<DataField> list = dataFieldService.list(dbQw);
String beginDateStr = null;
@ -437,6 +438,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
JobInfo salesforceInsertJob = null;
if (insertList.isEmpty()){
log.info("当前批次无可获取SFID的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
}
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
@ -447,6 +452,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
@ -540,19 +546,23 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
int retryCount = 0;
final int maxRetries = 3;
while (retryCount <= maxRetries) {
while (retryCount < maxRetries) {
try {
rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
InputStream batchResultStream = connection.getBatchResultStream(job.getId(), b.getId());
rdr = new CSVReader(batchResultStream);
break; // 成功执行，跳出重试循环
}catch (Exception e) {
retryCount++;
log.info("checkInsertResultsOne failed, retrying {}/{} for api: {}", retryCount, maxRetries, api);
try {
Thread.sleep(1000 * retryCount); // 简单的退避策略
Thread.sleep(5000 * retryCount); // 简单的退避策略
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while retrying", ie);
}
if (maxRetries <= retryCount){
return 0;
}
}
}
@ -964,6 +974,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
JobInfo salesforceUpdateJob = null;
if (updateList.isEmpty()){
log.info("当前批次无可更新的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
}
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
@ -1448,8 +1462,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
JobInfo salesforceInsertJob = null;
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
if (insertList.isEmpty()){
log.info("当前批次无可一次性插入的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
}
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
@ -1636,6 +1655,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
JobInfo salesforceInsertJob = null;
if (insertList.isEmpty()){
log.info("当前批次无可一次性插入的数据:{}；-开始时间:{}；-结束时间:{}；-api:{}；",i, beginDateStr, endDateStr, api);
}
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");

3
src/main/java/com/celnet/datadump/util/CsvConverterUtil.java
View File

@ -259,8 +259,7 @@ public class CsvConverterUtil {
if (isEditable){
header = Stream.concat(
fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
.map(DataField::getField),
fields.stream().map(DataField::getField),
Stream.of("old_owner_id__c", "old_sfdc_id__c")
).toArray(String[]::new);
}else {