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

【feat】 bulk增加日志

Browse Source
This commit is contained in:
Kris 2025-11-17 16:39:12 +08:00
parent 7125baf202
commit 93d56c627d
4 changed files with 140 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1119,6 +1119,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
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;
@ -1155,7 +1156,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
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());
@ -1165,6 +1168,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
log.info("批次处理完成, 处理记录数: {}", completionOne);
dataLogService.save(dataLog);
if (completionOne == 0){
log.info("无更多数据, 结束处理");
BulkUtil.closeJob(bulkConnect, job.getId());
@ -1173,6 +1178,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
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);
@ -1194,6 +1201,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
}
dataLogService.save(dataLog1);
if (!batchRecords.isEmpty()){
log.info("开始保存或更新数据, 记录数: {}", batchRecords.size());
sfData = saveOrUpdate(api, batchRecords, true);
@ -1206,6 +1215,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
BulkUtil.closeJob(bulkConnect, job.getId());
batch ++;
log.info("关闭作业成功, 作业ID: {}", job.getId());
}catch (InterruptedException interruptedException){
return 0;

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

@ -21,9 +21,6 @@ import com.celnet.datadump.util.DataUtil;
import com.celnet.datadump.util.EmailUtil;
import com.google.common.collect.Lists;
import com.sforce.async.*;
import com.sforce.soap.partner.PartnerConnection;
import com.sforce.soap.partner.SaveResult;
import com.sforce.soap.partner.sobject.SObject;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.DateUtil;
@ -81,6 +78,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
@Autowired
private LinkConfigService linkConfigService;
@Autowired
private DataLogService dataLogService;
/**
* Insert入口
*/
@ -326,6 +326,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
for (int i = 0; i < page; i++) {
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB");
List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000");
int size = data.size();
@ -423,17 +425,30 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
break;
}
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
JobInfo salesforceInsertJob = null;
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob);
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");
sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
} catch (InterruptedException interruptedException){
@ -511,41 +526,58 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
* 读写Insert结果
*/
public int checkInsertResultsOne(BulkConnection connection, JobInfo job,
BatchInfo b,String api,String[] ids)
throws AsyncApiException, IOException {
BatchInfo b,String api,String[] ids)
throws Exception {
int index = 0;
CSVReader rdr =
new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
List<String> resultHeader = rdr.nextRecord();
int resultCols = resultHeader.size();
CSVReader rdr = null;
int retryCount = 0;
final int maxRetries = 3;
while (retryCount <= maxRetries) {
try {
rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
List<String> resultHeader = rdr.nextRecord();
int resultCols = resultHeader.size();
List<String> row;
while ((row = rdr.nextRecord()) != null) {
Map<String, String> resultInfo = new HashMap<String, String>();
for (int i = 0; i < resultCols; i++) {
resultInfo.put(resultHeader.get(i), row.get(i));
List<String> row;
while ((row = rdr.nextRecord()) != null) {
Map<String, String> resultInfo = new HashMap<String, String>();
for (int i = 0; i < resultCols; i++) {
resultInfo.put(resultHeader.get(i), row.get(i));
}
boolean insertStatus = Boolean.valueOf(resultInfo.get("Success"));
boolean created = Boolean.valueOf(resultInfo.get("Created"));
String id = resultInfo.get("Id");
String error = resultInfo.get("Error");
if (insertStatus && created) {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m = new HashMap<>();
m.put("key", "new_id");
m.put("value", id);
maps.add(m);
customMapper.updateById(api, maps, ids[index]);
} else{
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", JSON.toJSONString(error));
maps.add(linkMap1);
customMapper.updateById(api, maps, ids[index]);
log.info("Id:{},saveResults: {}",ids[index], error);
}
index ++;
}
break; // 成功执行，跳出重试循环
}catch (Exception e) {
retryCount++;
log.info("checkInsertResultsOne failed, retrying {}/{} for api: {}", retryCount, maxRetries, api);
try {
Thread.sleep(1000 * retryCount); // 简单的退避策略
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while retrying", ie);
}
}
boolean insertStatus = Boolean.valueOf(resultInfo.get("Success"));
boolean created = Boolean.valueOf(resultInfo.get("Created"));
String id = resultInfo.get("Id");
String error = resultInfo.get("Error");
if (insertStatus && created) {
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m = new HashMap<>();
m.put("key", "new_id");
m.put("value", id);
maps.add(m);
customMapper.updateById(api, maps, ids[index]);
} else{
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> linkMap1 = new HashMap<>();
linkMap1.put("key", "error_message");
linkMap1.put("value", JSON.toJSONString(error));
maps.add(linkMap1);
customMapper.updateById(api, maps, ids[index]);
log.error("Id:{},saveResults: {}",ids[index], error);
}
index ++;
}
return index;
}
@ -826,6 +858,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
for (int i = 0; i < page; i++) {
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB");
List<Map<String, Object>> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 8000");
log.error("总Update数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@ -911,10 +945,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
}
updateList.add(account);
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
JobInfo salesforceUpdateJob = null;
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
salesforceUpdateJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update);
@ -922,10 +959,18 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
BulkUtil.awaitOneCompletion(bulkConnection, salesforceUpdateJob, batchInfo);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
int updatedCount = checkUpdateResultsOne(bulkConnection, salesforceUpdateJob, batchInfo,api);
sfNum = sfNum + updatedCount;
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
} catch (InterruptedException interruptedException){
@ -1100,7 +1145,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
update.setDataLock(1);
dataObjectService.updateById(update);
if (api.endsWith("Share") || "GroupMember".equals(api)){
if (dataFieldService.hasCreatedDate(api)){
insertSingleShareData(api,bulkConnection);
continue;
}
@ -1248,8 +1293,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
String sql = "";
String sql2 = "";
if (api.endsWith("Share")){
sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null ";
sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null limit 8000";
}else {
sql = "where new_id is null ";
sql2 = "new_id is null limit 8000";
}
//表内数据总量
Integer count = customMapper.countBySQL(api, "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
Integer count = customMapper.countBySQL(api, sql);
if (count == 0) {
return;
}
@ -1277,7 +1333,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
for (int i = 0; i < page; i++) {
List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000");
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD");
List<JSONObject> data = customMapper.listJsonObject("*", api, sql2);
int size = data.size();
log.error("总Insert数据 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@ -1374,15 +1432,29 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
JobInfo salesforceInsertJob = null;
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob);
BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceInsertJob);
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");
sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
} catch (Exception e) {
log.error("createdNewIdBatch error api:{}", api, e);
throw e;
@ -1448,6 +1520,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
for (int i = 0; i < page; i++) {
DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD");
List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null and RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') order by Id asc limit " + i * 10000 + ",10000");
int size = data.size();
@ -1523,9 +1597,14 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
}
}
dataLog.setEndTime(new Date());
dataLogService.save(dataLog);
JobInfo salesforceInsertJob = null;
try {
DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
@ -1533,8 +1612,17 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos);
dataLog1.setEndTime(new Date());
dataLogService.save(dataLog1);
DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB");
checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
dataLog2.setEndTime(new Date());
dataLogService.save(dataLog2);
} catch (Exception e) {
log.error("insertSingleShareData error api:{}", api, e);
throw e;

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

@ -3503,7 +3503,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
String value = String.valueOf(data.get(j - 1).get(field));
//根据旧sfid查找引用对象新sfid
if (dataField.getIsCreateable() != null && !dataField.getIsCreateable()) {
if (dataField.getIsCreateable() == null || !dataField.getIsCreateable()) {
continue;
} else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {

1
src/main/java/com/celnet/datadump/util/DataUtil.java
View File

@ -511,6 +511,7 @@ public class DataUtil {
public static Object localDataToSfData(String fieldType, String data) throws ParseException {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
Date date;
//date转Calendar类型