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

【fix】 优化数据质量校验

Browse Source
This commit is contained in:
Kris 2025-10-23 10:56:14 +08:00
parent 36398c1522
commit acc8e2ad46
1 changed files with 313 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

327
src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
View File

@ -581,6 +581,12 @@ public class DataVerifyServiceImpl implements DataVerifyService {
EmailUtil.sendMineWithFile(head, text, false,arrayList); EmailUtil.sendMineWithFile(head, text, false,arrayList);
} }
private void sendObjectQualityCheckEmailList( String objectApi ,List<String> filePaths) {
String head = objectApi + "_校验报告";
String text = objectApi + "生成时间：" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss");
EmailUtil.sendMineWithFile(head, text, false,filePaths);
}
/** /**
* 处理单个数据对象 * 处理单个数据对象
*/ */
@ -603,7 +609,6 @@ public class DataVerifyServiceImpl implements DataVerifyService {
// 2. 处理字段相关信息 // 2. 处理字段相关信息
Map<String, Map<String, Field>> fieldMap = handlerFieldInfo(fields, fieldTypeSheet, targetConn, sourceConn, dataObject, excelWriter); Map<String, Map<String, Field>> fieldMap = handlerFieldInfo(fields, fieldTypeSheet, targetConn, sourceConn, dataObject, excelWriter);
String qualityCheckFilePath = generateObjectQualityCheckFilePath(objectApi);
// 3. 处理每个批次 // 3. 处理每个批次
List<DataBatch> batches = getDataBatches(objectApi); List<DataBatch> batches = getDataBatches(objectApi);
@ -634,24 +639,318 @@ public class DataVerifyServiceImpl implements DataVerifyService {
} }
}else { }else {
boolean hasError = false; boolean hasError = false;
try (ExcelWriter objectWriter = EasyExcel.write(qualityCheckFilePath).build()) { ExcelWriter objectWriter = null;
WriteSheet detailSheet = null;
int fileIndex = 1;
int writtenRows = 0;
final int MAX_ROWS_PER_FILE = 900000; // 90万行限制
ArrayList<String> arrayList = new ArrayList<>();
String currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex);
arrayList.add(currentFilePath);
try {
objectWriter = EasyExcel.write(currentFilePath).build();
for (DataBatch batch : batches) { for (DataBatch batch : batches) {
hasError = processDataBatch(batch, objectApi, fields, excelWriter, objectWriter, summarySheet,dataObject, sourceConn, targetConn, fieldMap, ignoreBeginDate) || hasError; // 为每个批次创建新的detail sheet
detailSheet = createDetailSheet(objectWriter, dataObject);
String startDate = DateFormatUtils.format(batch.getSyncStartDate(), "yyyy-MM-dd HH:mm:ss");
String endDate = DateFormatUtils.format(batch.getSyncEndDate(), "yyyy-MM-dd HH:mm:ss");
log.info("开始处理数据批次: 批次ID={} 批次总量(大于十万只校验十万)={} 批次开始时间={} 批次结束时间={} 对象API={} 忽略开始时间={}", batch.getId(),batch.getFirstSfNum(),startDate,endDate, objectApi, ignoreBeginDate);
// 分页处理记录
int page = 0;
int totalRecordNum = 0;
int targetRecordNum = 0;
int sourceRecordNum = 0;
int sorceOnlyNum = 0;
int targetOnlyNum = 0;
Map<String, Field> sourceFieldMap = fieldMap.get("source");
Map<String, Field> targetFieldMap = fieldMap.get("target");
while (true) {
List<Map<String, Object>> recordIds = queryRecordIds(objectApi, startDate, endDate, page);
if (recordIds.isEmpty()) break;
int i = 0;
while (true){
// 检查是否需要创建新文件
if (writtenRows >= MAX_ROWS_PER_FILE) {
log.info("创建新文件，当前已写入行数: {}", writtenRows);
if (objectWriter != null) {
objectWriter.finish();
}
fileIndex++;
currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex);
arrayList.add(currentFilePath);
objectWriter = EasyExcel.write(currentFilePath).build();
detailSheet = createDetailSheet(objectWriter, dataObject);
writtenRows = 0;
}
int startIndex = MAX_BATCH_RECORDS * i;
int endIndex = Math.min(startIndex + MAX_BATCH_RECORDS, recordIds.size());
if (startIndex >= endIndex) {
break;
}
List<Map<String, Object>> idList = recordIds.subList(startIndex, endIndex);
totalRecordNum += idList.size();
List<String> ids = Lists.newArrayList();
List<String> newIds = Lists.newArrayList();
HashMap<String, String> idLinks = new HashMap<>();
for (Map<String, Object> idMap : idList) {
Object idObj = idMap.get("Id");
Object newIdObj = idMap.get("new_id");
if (idObj != null) {
ids.add("'" + idObj.toString() + "'");
}
if (newIdObj != null) {
newIds.add("'" + newIdObj.toString() + "'");
}
if (idObj != null && newIdObj != null) {
idLinks.put(idObj.toString(), newIdObj.toString());
}
}
// 批量查询源和目标数据
Map<String, SObject> sourceRecords = queryRecords(sourceConn, objectApi, ids);
sourceRecordNum += sourceRecords.size();
Map<String, SObject> targetRecords = queryRecords(targetConn, objectApi, newIds);
targetRecordNum += targetRecords.size();
// 比对记录
for (String id : idLinks.keySet()) {
// 检查是否需要创建新文件
if (writtenRows >= MAX_ROWS_PER_FILE) {
log.info("创建新文件，当前已写入行数: {}", writtenRows);
if (objectWriter != null) {
objectWriter.finish();
}
fileIndex++;
currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex);
arrayList.add(currentFilePath);
objectWriter = EasyExcel.write(currentFilePath).build();
detailSheet = createDetailSheet(objectWriter, dataObject);
writtenRows = 0;
}
String newId = idLinks.get(id);
SObject sourceRecord = sourceRecords.get(id);
SObject targetRecord = targetRecords.get(newId);
if (sourceRecord == null && targetRecord != null) {
targetOnlyNum ++;
List<String> row = Lists.newArrayList();
row.add(String.valueOf(batch.getId()));
row.add(id);
row.add(newId);
row.add(objectApi);
row.add(" ");
row.add(" ");
row.add(" ");
row.add(" ");
row.add("目标系统仅有");
row.add(" ");
row.add(" ");
row.add(" ");
row.add(" ");
row.add(" ");
row.add(" ");
row.add(String.valueOf(targetRecord.getField("LastModifiedDate")));
// 写入到Excel的detailSheet，而不是添加到某个detailData列表
objectWriter.write(Collections.singletonList(row), detailSheet);
writtenRows++;
hasError = true;
continue;
}else if (sourceRecord != null && targetRecord == null) {
sorceOnlyNum ++;
List<String> row = Lists.newArrayList();
row.add(String.valueOf(batch.getId()));
row.add(id);
row.add(newId);
row.add(objectApi);
row.add(" ");
row.add(" ");
row.add(" ");
row.add(" ");
row.add("源ORG仅有");
row.add(" ");
row.add(" ");
row.add(" ");
row.add(" ");
row.add(String.valueOf(sourceRecord.getField("CreatedDate")));
row.add(String.valueOf(sourceRecord.getField("LastModifiedDate")));
row.add(" ");
// 写入到Excel的detailSheet，而不是添加到某个detailData列表
objectWriter.write(Collections.singletonList(row), detailSheet);
writtenRows++;
hasError = true;
continue;
}
boolean isSame = true;
if (ignoreBeginDate != null) {
Object lastModifiedObj = sourceRecord.getField("LastModifiedDate");
if (lastModifiedObj != null) {
try {
// 解析Salesforce返回的日期格式 2025-10-12T03:27:54.000Z
String lastModifiedStr = lastModifiedObj.toString();
Date lastModifiedDate = org.apache.commons.lang3.time.DateUtils.parseDate(lastModifiedStr, "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
if (lastModifiedDate.before(ignoreBeginDate)) {
isSame = false;
}
} catch (Exception e) {
log.warn("解析LastModifiedDate失败: {}", lastModifiedObj, e);
}
}
}
if (isSame){
// 比对字段值
for (DataField field : fields) {
// 检查是否需要创建新文件
if (writtenRows >= MAX_ROWS_PER_FILE) {
log.info("创建新文件，当前已写入行数: {}", writtenRows);
if (objectWriter != null) {
objectWriter.finish();
}
fileIndex++;
currentFilePath = generateObjectQualityCheckFilePath(objectApi + "_part" + fileIndex);
arrayList.add(currentFilePath);
objectWriter = EasyExcel.write(currentFilePath).build();
detailSheet = createDetailSheet(objectWriter, dataObject);
writtenRows = 0;
}
String fieldName = field.getField();
if (Const.EXCLUDE_FIELDS.contains(fieldName))
continue; // 跳过字段比较
Object sourceValue = sourceRecord.getField(fieldName);
Object targetValue = targetRecord.getField(fieldName);
Field sorceField = sourceFieldMap.get(fieldName);
Field targetField = targetFieldMap.get(fieldName);
if (sourceValue == null && targetValue == null) {
continue;
}
//判断reference_to内是否包含User字符串
String referenceTo = field.getReferenceTo();
if (referenceTo !=null && (referenceTo.contains(",User") || referenceTo.contains("User,"))) {
referenceTo = "User";
}
if (referenceTo != null &&field.getSfType().equals("reference") && !referenceTo.equals("data_picklist")) {
String[] split = referenceTo.split(",");
if (split.length > 1 && !"User".equals(referenceTo)){
continue;
}
if (StringUtils.isBlank(referenceTo)){
continue;
}
Map<String, Object> objectMap = customMapper.getById("new_id", referenceTo, String.valueOf(sourceValue));
if (objectMap == null || !String.valueOf(objectMap.get("new_id")).equals(String.valueOf(targetValue))) {
List<String> row = Lists.newArrayList();
row.add(String.valueOf(batch.getId()));
row.add(id);
row.add(newId);
row.add(objectApi);
row.add(field.getField());
row.add(field.getName());
row.add(String.valueOf(sourceValue));
row.add(String.valueOf(targetValue));
row.add("字段值不一致");
row.add(String.valueOf(sorceField.getType()));
row.add(String.valueOf(targetField.getType()));
row.add(String.valueOf(sorceField.isCreateable()));
row.add(String.valueOf(targetField.isCreateable()));
row.add(String.valueOf(sourceRecord.getField("CreatedDate")));
row.add(String.valueOf(sourceRecord.getField("LastModifiedDate")));
row.add(String.valueOf(targetRecord.getField("LastModifiedDate")));
// 写入到Excel的detailSheet，而不是添加到某个detailData列表
objectWriter.write(Collections.singletonList(row), detailSheet);
writtenRows++;
hasError = true;
}
} else if (!Objects.equals(sourceValue, targetValue)){
List<String> row = Lists.newArrayList();
row.add(String.valueOf(batch.getId()));
row.add(id);
row.add(newId);
row.add(objectApi);
row.add(field.getField());
row.add(field.getName());
row.add(String.valueOf(sourceValue));
row.add(String.valueOf(targetValue));
row.add("字段值不一致");
row.add(String.valueOf(sorceField.getType()));
row.add(String.valueOf(targetField.getType()));
row.add(String.valueOf(sorceField.isCreateable()));
row.add(String.valueOf(targetField.isCreateable()));
row.add(String.valueOf(sourceRecord.getField("CreatedDate")));
row.add(String.valueOf(sourceRecord.getField("LastModifiedDate")));
row.add(String.valueOf(targetRecord.getField("LastModifiedDate")));
// 写入到Excel的detailSheet，而不是添加到某个detailData列表
objectWriter.write(Collections.singletonList(row), detailSheet);
writtenRows++;
hasError = true;
}
}
}
}
i++;
}
page++;
if (page * MAX_PAGE_RECORDS >= MAX_TOTAL_RECORDS) {
log.info("达到最大处理记录数限制，停止处理");
break; // 防止无限循环
}
}
// 写入批次汇总结果
List<String> row = Lists.newArrayList();
row.add(String.valueOf(batch.getId()));
row.add(objectApi);
row.add(startDate);
row.add(endDate);
row.add(String.valueOf(sourceRecordNum));
row.add(String.valueOf(targetRecordNum));
row.add(String.valueOf(totalRecordNum));
row.add(String.valueOf(sorceOnlyNum));
row.add(String.valueOf(targetOnlyNum));
row.add(hasError?"不通过":"通过");
// 写入Excel的summarySheet
synchronized (this) {
excelWriter.write(Collections.singletonList(row), summarySheet);
}
} }
}catch (Exception e){ } catch (Exception e){
log.error("API={},处理异常：{}",objectApi, e.getMessage(),e); log.error("API={},处理异常：{}",objectApi, e.getMessage(),e);
}finally {
if (objectWriter != null) {
objectWriter.finish();
}
} }
if (sendEmail && hasError){ if (sendEmail && hasError){
sendObjectQualityCheckEmail(objectApi, qualityCheckFilePath); sendObjectQualityCheckEmailList(objectApi, arrayList);
}else { }else {
log.info("不发送对象质量校验详情邮件：sendEmail={}, batchHasError={}", sendEmail, hasError); log.info("不发送对象质量校验详情邮件：sendEmail={}, batchHasError={}", sendEmail, hasError);
// 删除临时文件以节省空间
File file = new File(qualityCheckFilePath);
if (file.exists()) {
file.delete();
}
} }
} }
@ -781,7 +1080,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
if (startIndex >= endIndex) { if (startIndex >= endIndex) {
break; break;
} }
List<Map<String, Object>> idList = recordIds.subList(startIndex, endIndex); List<Map<String, Object>> idList = recordIds.subList(startIndex, endIndex);
totalRecordNum += idList.size(); totalRecordNum += idList.size();
@ -791,7 +1090,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
for (Map<String, Object> idMap : idList) { for (Map<String, Object> idMap : idList) {
Object idObj = idMap.get("Id"); Object idObj = idMap.get("Id");
Object newIdObj = idMap.get("new_id"); Object newIdObj = idMap.get("new_id");
if (idObj != null) { if (idObj != null) {
ids.add("'" + idObj.toString() + "'"); ids.add("'" + idObj.toString() + "'");
} }
@ -872,7 +1171,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
// 解析Salesforce返回的日期格式 2025-10-12T03:27:54.000Z // 解析Salesforce返回的日期格式 2025-10-12T03:27:54.000Z
String lastModifiedStr = lastModifiedObj.toString(); String lastModifiedStr = lastModifiedObj.toString();
Date lastModifiedDate = org.apache.commons.lang3.time.DateUtils.parseDate(lastModifiedStr, "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"); Date lastModifiedDate = org.apache.commons.lang3.time.DateUtils.parseDate(lastModifiedStr, "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
if (lastModifiedDate.before(ignoreBeginDate)) { if (lastModifiedDate.before(ignoreBeginDate)) {
isSame = false; isSame = false;
} }