diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java index 1d54334..f48ae77 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -302,8 +302,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if (count == 0) { return; } - //批量插入5000一次 - int page = count%5000 == 0 ? count/5000 : (count/5000) + 1; + //批量插入8000一次 + int page = count%8000 == 0 ? count/8000 : (count/8000) + 1; log.error("总Insert数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api); @@ -326,17 +326,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (int i = 0; i < page; i++) { - List data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 5000"); + List data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000"); int size = data.size(); log.error("总Insert数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api); List insertList = new ArrayList<>(); - //判断引用对象是否存在new_id - DataObject update = new DataObject(); - update.setName(api); - //更新对象的new_id String[] ids = new String[size]; @@ -345,6 +341,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { DateTimeFormatter outputFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSxx"); for (int j = 1; j <= size; j++) { + String message = null; JSONObject account = new JSONObject(); for (DataField dataField : list) { if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField()) @@ -363,6 +360,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { Map CreatedByIdMap = customMapper.getById("new_id", "User", data.get(j-1).get("CreatedById").toString()); if(CreatedByIdMap.get("new_id") != null && StringUtils.isNotEmpty(CreatedByIdMap.get("new_id").toString())){ account.put("CreatedById", CreatedByIdMap.get("new_id")); + }else { + message = "CreatedById:" + data.get(j - 1).get("CreatedById") + "不存在"; + log.info( message); + break; } continue; } @@ -376,12 +377,14 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if (data.get(j-1).get(linkfield)!=null){ reference = data.get(j-1).get(linkfield).toString(); }else { - log.info("对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！"); + message ="对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的关联类型不存在！"; + log.info(message); } } if (reference == null){ continue; } + List> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1"); if (referenceMap.isEmpty()){ QueryWrapper maxIndex = new QueryWrapper<>(); @@ -389,12 +392,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { maxIndex.ne("name", api); Map map = dataObjectService.getMap(maxIndex); //如果必填lookup字段没有值，跳过 - update.setDataIndex(Integer.parseInt(map.get("data_index").toString()) + 1); - dataObjectService.updateById(update); - String message = "api:" + api + "的引用对象:" + reference + "不存在数据！"; - String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s", api, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message); - EmailUtil.send("DataDump ERROR", format); - log.info(message); + dataObject.setDataIndex(Integer.parseInt(map.get("data_index").toString()) + 1); + dataObjectService.updateById(dataObject); return; }else{ account.put(dataField.getField(), referenceMap.get(0).get("new_id")); @@ -408,8 +407,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } account.put(dataField.getField(), DataUtil.fieldTypeBulkToSf(dataField)); } - } + if (message != null){ + continue; + } + if (dataObject.getIsEditable()){ account.put("old_owner_id__c", data.get(j-1).get("OwnerId")); account.put("old_sfdc_id__c", data.get(j-1).get("Id")); @@ -417,7 +419,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { ids[j-1] = data.get(j-1).get("Id").toString(); insertList.add(account); - if (i*5000+j >= count){ + if (i*8000+j >= count){ break; } } @@ -426,17 +428,24 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { try { salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); - List batchInfos = CsvConverterUtil.writeToCsvNew(insertList, UUID.randomUUID().toString(),list,true,bulkConnection,salesforceInsertJob); + BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob); - BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); + BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo); - sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); + sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids); - } catch (Exception e) { - log.error("createdNewIdBatch error api:{}", api, e); + TimeUnit.MILLISECONDS.sleep(1); + + } catch (InterruptedException interruptedException){ + return; + }catch (Exception e) { throw e; }finally { - BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId()); + try { + BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId()); + }catch (Exception e){ + log.error("closeJob error", e); + } } } @@ -498,6 +507,50 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } + /** + * 读写Insert结果 + */ + public int checkInsertResultsOne(BulkConnection connection, JobInfo job, + BatchInfo b,String api,String[] ids) + throws AsyncApiException, IOException { + int index = 0; + CSVReader rdr = + new CSVReader(connection.getBatchResultStream(job.getId(), b.getId())); + List resultHeader = rdr.nextRecord(); + int resultCols = resultHeader.size(); + + List row; + while ((row = rdr.nextRecord()) != null) { + Map resultInfo = new HashMap(); + 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> maps = new ArrayList<>(); + Map m = new HashMap<>(); + m.put("key", "new_id"); + m.put("value", id); + maps.add(m); + customMapper.updateById(api, maps, ids[index]); + } else{ + List> maps = new ArrayList<>(); + Map 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; + } + + public Integer checkQueryResultsNew(BulkConnection connection, JobInfo job, BatchInfo batchInfo,String api) throws AsyncApiException, IOException { int num = 0; // 获取查询结果的 ResultId (对于Query类型，每个Batch可能包含多个ResultId) @@ -744,7 +797,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss"); } // 表内数据总量 - Integer count = customMapper.countBySQL(api, "where new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"); + Integer count = customMapper.countBySQL(api, "where new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"); if(count == 0){ return; @@ -766,14 +819,14 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { // 总更新数 int sfNum = 0; - // 批量更新5000一次 - int page = count%5000 == 0 ? count/5000 : (count/5000) + 1; + // 批量更新8000一次 + int page = count%8000 == 0 ? count/8000 : (count/8000) + 1; log.error("总Update数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api); for (int i = 0; i < page; i++) { - List> mapList = customMapper.list("*", api, "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 5000 + ",5000"); + List> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 8000 + ",8000"); log.error("总Update数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api); @@ -781,6 +834,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (Map map : mapList) { JSONObject account = new JSONObject(); + String message = null; //给对象赋值 for (DataField dataField : list) { String field = dataField.getField(); @@ -791,9 +845,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if (field.equals("Id")) { account.put("Id",String.valueOf(map.get("new_id"))); - } else if (!DataUtil.isUpdate(field) || (dataField.getIsCreateable() != null && !dataField.getIsCreateable())) { - continue; - } else if ( "WhoId".equals(field) ||"WhatId".equals(field)){ + } else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) { continue; } else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null ) { @@ -818,10 +870,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if (m != null && !m.isEmpty()) { account.put(field, m.get("new_id")); }else { - String message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！"; - EmailUtil.send("DataDump ERROR", message); + message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！"; log.info(message); - return; + break; } } } else { @@ -846,18 +897,24 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { salesforceUpdateJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update); - List batchInfos = CsvConverterUtil.writeToCsvNew(updateList, UUID.randomUUID().toString(),list,true,bulkConnection,salesforceUpdateJob); + BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(updateList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceUpdateJob); - BulkUtil.awaitCompletion(bulkConnection, salesforceUpdateJob, batchInfos); + BulkUtil.awaitOneCompletion(bulkConnection, salesforceUpdateJob, batchInfo); - sfNum = sfNum + checkUpdateResults(bulkConnection, salesforceUpdateJob, batchInfos,api); + sfNum = sfNum + checkUpdateResultsOne(bulkConnection, salesforceUpdateJob, batchInfo,api); - } catch (Throwable e) { - log.info(JSON.toJSONString(e)); + TimeUnit.MILLISECONDS.sleep(1); + + } catch (InterruptedException interruptedException){ + return; + }catch (Exception e) { throw e; }finally { - - BulkUtil.closeJob(bulkConnection, salesforceUpdateJob.getId()); + try { + BulkUtil.closeJob(bulkConnection, salesforceUpdateJob.getId()); + }catch (Exception e){ + log.error("closeJob error", e); + } } } @@ -877,6 +934,54 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } + /** + * 读写update结果 + */ + public int checkUpdateResultsOne(BulkConnection connection, JobInfo job, + BatchInfo b,String api) + throws AsyncApiException, IOException { + int index = 0; + // batchInfoList was populated when batches were created and submitted + CSVReader rdr = + new CSVReader(connection.getBatchResultStream(job.getId(), b.getId())); + List resultHeader = rdr.nextRecord(); + int resultCols = resultHeader.size(); + + List row; + while ((row = rdr.nextRecord()) != null) { + Map resultInfo = new HashMap(); + for (int i = 0; i < resultCols; i++) { + resultInfo.put(resultHeader.get(i), row.get(i)); + } + boolean updateStatus = Boolean.valueOf(resultInfo.get("Success")); + String id = resultInfo.get("Id"); + String error = resultInfo.get("Error"); + if (updateStatus) { + index ++; + List> maps = new ArrayList<>(); + Map linkMap = new HashMap<>(); + linkMap.put("key", "is_update"); + linkMap.put("value", 1); + maps.add(linkMap); + customMapper.updateByNewId(maps,api, id); + } else { + List> maps = new ArrayList<>(); + Map linkMap = new HashMap<>(); + linkMap.put("key", "is_update"); + linkMap.put("value", 2); + maps.add(linkMap); + Map linkMap1 = new HashMap<>(); + linkMap1.put("key", "error_message"); + linkMap1.put("value", JSON.toJSONString(error)); + maps.add(linkMap1); + customMapper.updateByNewId(maps,api, id); + String format = String.format("数据更新 error, api name: %s, \ncause:\n%s", api, JSON.toJSONString(error)); + log.info(format); + } + } + return index; + } + /** * 读写update结果 */ @@ -1125,8 +1230,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if (count == 0) { return; } - //批量插入10000一次 - int page = count%5000 == 0 ? count/5000 : (count/5000) + 1; + //批量插入8000一次 + int page = count%8000 == 0 ? count/8000 : (count/8000) + 1; log.error("总Insert数据 count:{}；总批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api); @@ -1149,7 +1254,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (int i = 0; i < page; i++) { - List data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 5000"); + List data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000"); int size = data.size(); log.error("总Insert数据 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api); @@ -1165,6 +1270,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (int j = 1; j <= size; j++) { JSONObject account = new JSONObject(); + String message = null; for (DataField dataField : list) { if ("Owner_Type".equals(dataField.getField()) || "Id".equals(dataField.getField())){ @@ -1210,8 +1316,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { if (m != null && !m.isEmpty()) { account.put(dataField.getField(), m.get("new_id")); }else { - String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(dataField.getField()) +"不存在！"; + message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(dataField.getField()) +"不存在！"; log.info(message); + break; } }else { if (data.get(j - 1).get(dataField.getField()) != null && StringUtils.isNotBlank(dataField.getSfType())) { @@ -1222,14 +1329,16 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } } } - + if (message!=null){ + continue; + } if (dataObject.getIsEditable()) { account.put("old_owner_id__c", data.get(j - 1).get("OwnerId")); account.put("old_sfdc_id__c", data.get(j - 1).get("Id")); } ids[j-1] = data.get(j-1).get("Id").toString(); insertList.add(account); - if (i*5000+j == count){ + if (i*8000+j >= count){ break; } } @@ -1239,11 +1348,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { try { salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); - List batchInfos = CsvConverterUtil.writeToCsvNew(insertList, UUID.randomUUID().toString(),list,true,bulkConnection,salesforceInsertJob); + BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob); - BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); + BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo); - sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); + sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids); } catch (Exception e) { log.error("createdNewIdBatch error api:{}", api, e); @@ -1289,7 +1398,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { return; } //批量插入10000一次 - int page = count%5000 == 0 ? count/5000 : (count/5000) + 1; + int page = count%8000 == 0 ? count/8000 : (count/8000) + 1; log.error("总Insert数据 count:{}；总批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api); @@ -1371,7 +1480,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { ids[j-1] = data.get(j-1).get("Id").toString(); insertList.add(account); - if (i*5000+j == count){ + if (i*8000+j == count){ break; } } diff --git a/src/main/java/com/celnet/datadump/util/BulkUtil.java b/src/main/java/com/celnet/datadump/util/BulkUtil.java index c955b78..ad6ed12 100644 --- a/src/main/java/com/celnet/datadump/util/BulkUtil.java +++ b/src/main/java/com/celnet/datadump/util/BulkUtil.java @@ -115,6 +115,40 @@ public class BulkUtil { } } + /** + * Wait for a job to complete by polling the Bulk API. + * + * @param connection + * BulkConnection used to check results. + * @param job + * The job awaiting completion. + * List of batches for this job. + * @throws AsyncApiException + */ + public static void awaitOneCompletion(BulkConnection connection, JobInfo job, + BatchInfo bi) + throws AsyncApiException { + long sleepTime = 0L; + Set incomplete = new HashSet(); + incomplete.add(bi.getId()); + while (!incomplete.isEmpty()) { + try { + Thread.sleep(sleepTime); + } catch (InterruptedException e) {} + System.out.println("Awaiting results..." + incomplete.size()); + sleepTime = 10000L; + BatchInfo[] statusList = + connection.getBatchInfoList(job.getId()).getBatchInfo(); + for (BatchInfo b : statusList) { + if (b.getState() == BatchStateEnum.Completed + || b.getState() == BatchStateEnum.Failed) { + if (incomplete.remove(b.getId())) { + System.out.println("BATCH STATUS:\n" + b); + } + } + } + } + } /** * Create a new job using the Bulk API. * diff --git a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java b/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java index 1b2760e..1496afb 100644 --- a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java +++ b/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java @@ -163,6 +163,83 @@ public class CsvConverterUtil { return batchInfos; } + /** + * 将JSON数据写入CSV(UTF-8)文件并上传 + * @param jsonList JSON数据列表 + * @param fileName 文件名(不含扩展名) + * @param fields 字段定义 + * @param isEditable 是否编辑模式 + * @return 批处理信息列表 + */ + public static BatchInfo writeToCsvNewOne(List jsonList, + String fileName, + List fields, + Boolean isEditable, + BulkConnection connection, + JobInfo jobInfo) { + + // 1. 创建目标目录（不存在则创建） + File targetDir = FileUtil.mkdir("data-dump/dataFile"); + + // 2. 构建完整文件路径 + String fullPath = targetDir.getAbsolutePath() + File.separator + fileName + ".csv"; + + CsvWriter csvWriter = CsvUtil.getWriter(fullPath, CharsetUtil.CHARSET_UTF_8, false); + + String[] header = null; + + File file = null; + + if (isEditable){ + header = Stream.concat( + fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable())) + .map(DataField::getField), + Stream.of("Id","old_owner_id__c", "old_sfdc_id__c") + ).toArray(String[]::new); + }else { + header = fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable())) + .map(DataField::getField).toArray(String[]::new); + } + + BatchInfo batch = null; + + try { + + // 2. 写入表头（必须使用 String[]） + csvWriter.writeHeaderLine(header); + + // 遍历数据列表 + for (JSONObject jsonObject : jsonList) { + // 按表头顺序获取值 + String[] row = new String[header.length]; + for (int i = 0; i < header.length; i++) { + // 将每个值转换为字符串（如果为null则转为空字符串） + Object value = jsonObject.get(header[i]); + row[i] = value == null ? "" : value.toString(); + } + csvWriter.writeLine(row); + } + + // 关闭writer（在try-with-resources中可省略，但这里我们显式关闭） + csvWriter.close(); + + file = new File(fullPath); + + InputStream inputStream = Files.newInputStream(file.toPath()); + + batch = connection.createBatchFromStream(jobInfo, inputStream); + + } catch (IOException e) { + log.error("CSV文件操作失败: {}", fullPath, e); + file.deleteOnExit(); + } catch (AsyncApiException e) { + log.error("Bulk API上传失败: {}", e.getMessage(), e); + file.deleteOnExit(); + } + + return batch; + } + public static String exportToCsv(List> data, String fileName) throws IOException { // 1. 创建目标目录（不存在则创建） File targetDir = FileUtil.mkdir("data-dump/dataFile"); diff --git a/src/main/resources/mapper/CustomMapper.xml b/src/main/resources/mapper/CustomMapper.xml index 1923816..cadc0f2 100644 --- a/src/main/resources/mapper/CustomMapper.xml +++ b/src/main/resources/mapper/CustomMapper.xml @@ -128,7 +128,7 @@ UPDATE `${tableName}` SET - + ${map.key} = #{map.value} WHERE