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

[feat] 提交初版BigObject数据同步功能

Browse Source
This commit is contained in:
Kris 2025-09-28 14:49:57 +08:00
parent 89dcbb48bd
commit 6a2492cea7
3 changed files with 161 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -158,18 +158,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
@Override
public ReturnT<String> importBigObject(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
// 手动任务
ReturnT<String> result = manualBigObjectImport(param, futures);
if (result != null) {
return result;
}
}
return ReturnT.SUCCESS;
return manualBigObjectImport(param);
} catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("import error", throwable);
throw throwable;
}
@ -186,9 +177,11 @@ public class CommonBatchServiceImpl implements CommonBatchService {
log.info("开始导出BigObject数据, API: {}", api);
List<DataField> fieldList = dataFieldService.list(new QueryWrapper<DataField>().eq("api", param.getApi()));
String fieldStr = fieldList.stream()
.map(DataField::getField)
.filter(field -> !"SystemModstamp".equals(field) && !"LastModifiedDate".equals(field))
.collect(Collectors.joining(", "));
String sql = "select " + fieldStr + " from " + param.getApi() ;
@ -242,6 +235,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
long currentFileSize = 0;
int resultIndex = 0;
String headerLine = null;
for (final String resultId : queryResultList.getResult()) {
log.info("处理结果ID: {}, 进度: {}/{}", resultId, ++resultIndex, queryResultList.getResult().length);
@ -251,6 +246,13 @@ public class CommonBatchServiceImpl implements CommonBatchService {
String line;
int lineCount = 0;
while ((line = bufferedReader.readLine()) != null) {
// 保存第一行作为标题行，并替换CreatedDate和CreatedById为Original_Created_Date__c和Original_Created_By__c
// 但要确保不替换CreatedDate__c和CreatedById__c这样的自定义字段
line = line.replaceAll("(?<!__)CreatedDate(?!__)", "Original_Created_Date__c")
.replaceAll("(?<!__)CreatedById(?!__)", "Original_Created_By__c");
if (headerLine == null) {
headerLine = line;
}
lineCount++;
byte[] lineBytes = (line + "\n").getBytes(StandardCharsets.UTF_8);
@ -272,7 +274,13 @@ public class CommonBatchServiceImpl implements CommonBatchService {
fileWriter = new FileWriter(currentFilePath.toFile(), true);
bufferedWriter = new BufferedWriter(fileWriter);
currentFileSize = 0;
lineCount = 0;
lineCount = 1; // 重置行计数，新文件将写入标题行
// 在新文件中写入标题行
if (headerLine != null) {
bufferedWriter.write(headerLine + "\n");
currentFileSize += (headerLine + "\n").getBytes(StandardCharsets.UTF_8).length;
}
}
// 写入数据
@ -304,13 +312,14 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future<?>> futures) throws Exception {
private ReturnT<String> manualBigObjectImport(SalesforceParam param) throws Exception {
String api = param.getApi();
// 创建目标ORG连接
BulkConnection targetBulkConnect = salesforceTargetConnect.createBulkConnect();
File directory = new File(api);
// 修改目录路径，使用与导出相同的路径规范
File directory = new File(Const.SERVER_FILE_PATH + "/" + api);
if (!directory.exists() || !directory.isDirectory()) {
throw new FileNotFoundException(api + "目录不存在: " + directory.getAbsolutePath());
}
@ -327,26 +336,61 @@ private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future
// 使用Bulk V2 API导入所有文件
for (File csvFile : files) {
String filePath = csvFile.getAbsolutePath();
log.info("处理CSV文件: {}", filePath);
// 创建Bulk V2 API作业
String restEndpoint = targetBulkConnect.getConfig().getRestEndpoint();
String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
// 创建Bulk V2作业
String jobId = BulkUtil.createBulkV2InsertJob(url, api, targetBulkConnect);
log.info("创建Bulk V2作业成功, 作业ID: {}", jobId);
// 上传CSV文件数据
uploadCsvDataToJob(url, jobId, targetBulkConnect, filePath);
// 关闭作业
closeBulkV2Job(url, jobId, targetBulkConnect);
// 等待作业完成
BulkUtil.waitForBulkV2JobCompletion(url, jobId, targetBulkConnect);
log.info("文件 {} 数据导入完成", csvFile.getName());
try {
String filePath = csvFile.getAbsolutePath();
log.info("开始处理CSV文件: {}", filePath);
// 检查CSV文件是否为空或只有一行标题
try (BufferedReader reader = new BufferedReader(new FileReader(csvFile))) {
String firstLine = reader.readLine();
if (firstLine == null || firstLine.trim().isEmpty()) {
log.warn("CSV文件为空，跳过处理: {}", filePath);
continue;
}
String secondLine = reader.readLine();
if (secondLine == null || secondLine.trim().isEmpty()) {
log.warn("CSV文件只包含标题行，跳过处理: {}", filePath);
continue;
}
}
// 创建Bulk V2 API作业
String restEndpoint = targetBulkConnect.getConfig().getRestEndpoint();
String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
log.debug("Salesforce实例URL: {}", url);
// 创建Bulk V2作业
String jobId = BulkUtil.createBulkV2InsertJob(url, api, targetBulkConnect);
log.info("创建Bulk V2作业成功, 作业ID: {}", jobId);
// 上传CSV文件数据
log.info("开始上传CSV文件数据，文件路径: {}, 作业ID: {}", filePath, jobId);
uploadCsvDataToJob(url, jobId, targetBulkConnect, filePath);
log.info("CSV文件数据上传完成，文件路径: {}, 作业ID: {}", filePath, jobId);
// 上传文件完成
log.info("开始关闭文件上传作业，作业ID: {}", jobId);
closeBulkV2Job(url, jobId, targetBulkConnect);
log.info("作业关闭请求已发送，作业ID: {}", jobId);
// 等待作业完成
log.info("等待作业完成，作业ID: {}", jobId);
BulkUtil.waitForBulkV2JobCompletion(url, jobId, targetBulkConnect);
log.info("文件 {} 数据导入完成，作业ID: {}", csvFile.getName(), jobId);
// 下载作业结果文件（成功、失败、未处理记录）
log.info("开始下载作业结果文件，作业ID: {}", jobId);
downloadJobResultFiles(url, jobId, targetBulkConnect, api, csvFile.getName());
log.info("作业结果文件下载完成，作业ID: {}", jobId);
TimeUnit.SECONDS.sleep(30);
} catch (InterruptedException interruptedExc ){
return ReturnT.FAIL;
}catch (Exception e) {
log.error("处理文件 {} 时发生错误: {}", csvFile.getName(), e.getMessage(), e);
}
}
log.info("所有BigObject数据导入完成: {}", api);
@ -365,16 +409,25 @@ private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future
private void uploadCsvDataToJob(String url, String jobId, BulkConnection connection, String csvFilePath) throws Exception {
CloseableHttpClient httpClient = HttpClients.createDefault();
String uploadUrl = url + "services/data/v56.0/jobs/ingest/" + jobId + "/batches/";
// 创建HttpPut请求，用于上传CSV文件数据到Bulk API 2.0作业
HttpPut httpPut = new HttpPut(uploadUrl);
// 设置请求内容类型为CSV格式
httpPut.setHeader("Content-Type", "text/csv");
// 设置认证信息，使用Bearer Token方式认证
httpPut.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
// 设置接受的响应内容类型为JSON格式
httpPut.setHeader("Accept", "application/json");
// 设置美化打印输出，便于调试和日志查看
httpPut.setHeader("X-PrettyPrint", "1");
FileEntity fileEntity = new FileEntity(new File(csvFilePath));
httpPut.setEntity(fileEntity);
CloseableHttpResponse response = httpClient.execute(httpPut);
try {
int statusCode = response.getStatusLine().getStatusCode();
if (statusCode != 201) { // 201 Created
String errorBody = EntityUtils.toString(response.getEntity());
throw new RuntimeException("Failed to upload data to Bulk API 2.0 job, status code: " + statusCode + ", error: " + errorBody);
@ -420,6 +473,78 @@ private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future
log.info("作业关闭成功，作业ID: {}", jobId);
}
/**
* 下载Bulk API V2作业的结果文件（成功、失败、未处理记录）
*
* @param url Salesforce实例URL
* @param jobId 作业ID
* @param connection BulkConnection连接对象
* @param api API名称
* @param sourceFileName 源文件名
* @throws Exception 下载过程中发生错误时抛出异常
*/
private void downloadJobResultFiles(String url, String jobId, BulkConnection connection, String api, String sourceFileName) throws Exception {
// 确保结果目录存在
String resultDirPath = Const.SERVER_FILE_PATH + "/" + api + "/results";
Path resultDir = Paths.get(resultDirPath);
if (!Files.exists(resultDir)) {
Files.createDirectories(resultDir);
log.info("创建结果目录: {}", resultDir.toAbsolutePath());
}
// 构造基础文件名（不含扩展名）
String baseFileName = sourceFileName;
if (sourceFileName.endsWith(".csv")) {
baseFileName = sourceFileName.substring(0, sourceFileName.length() - 4);
}
// 下载成功记录
downloadJobResultFile(url, jobId, connection, "successfulResults", resultDirPath, baseFileName + "_successful.csv");
// 下载失败记录
downloadJobResultFile(url, jobId, connection, "failedResults", resultDirPath, baseFileName + "_failed.csv");
// 下载未处理记录
downloadJobResultFile(url, jobId, connection, "unprocessedrecords", resultDirPath, baseFileName + "_unprocessed.csv");
}
/**
* 下载单个作业结果文件
*
* @param url Salesforce实例URL
* @param jobId 作业ID
* @param connection BulkConnection连接对象
* @param resultType 结果类型 (successfulResults, failedResults, unprocessedrecords)
* @param resultDirPath 结果目录路径
* @param fileName 文件名
* @throws Exception 下载过程中发生错误时抛出异常
*/
private void downloadJobResultFile(String url, String jobId, BulkConnection connection,
String resultType, String resultDirPath, String fileName) throws Exception {
CloseableHttpClient httpClient = HttpClients.createDefault();
String resultUrl = url + "services/data/v56.0/jobs/ingest/" + jobId + "/" + resultType;
HttpGet httpGet = new HttpGet(resultUrl);
httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
httpGet.setHeader("Content-Type", "application/json; charset=UTF-8");
CloseableHttpResponse response = httpClient.execute(httpGet);
try {
int statusCode = response.getStatusLine().getStatusCode();
if (statusCode == 200) {
String responseBody = EntityUtils.toString(response.getEntity());
Path resultFilePath = Paths.get(resultDirPath, fileName);
Files.write(resultFilePath, responseBody.getBytes(StandardCharsets.UTF_8));
log.info("下载{}文件成功: {}", resultType, resultFilePath.toAbsolutePath());
} else {
String errorBody = EntityUtils.toString(response.getEntity());
log.warn("下载{}文件失败，状态码: {}，错误: {}", resultType, statusCode, errorBody);
}
} finally {
response.close();
httpClient.close();
}
}
private ReturnT<String> manualBatchDump(SalesforceParam param, List<Future<?>> futures) throws InterruptedException {
List<String> apis;
apis = DataUtil.toIdList(param.getApi());

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

@ -652,7 +652,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
targetRow.add(targetField.isNillable());
targetRow.add(targetField.isUnique());
targetRow.add(targetField.isPolymorphicForeignKey() ? "是" : "否");
targetRow.add((targetField.getReferenceTo()!=null && targetField.getReferenceTo().length> 2 )? "多态字段" : String.join(",", sourceField.getReferenceTo()));
targetRow.add((targetField.getReferenceTo()!=null && targetField.getReferenceTo().length> 2 )? "多态字段" : String.join(",", targetField.getReferenceTo()));
// 写入Excel
excelWriter.write(Collections.singletonList(targetRow), fieldTypeSheet);
@ -866,7 +866,6 @@ public class DataVerifyServiceImpl implements DataVerifyService {
row.add(String.valueOf(totalRecordNum));
row.add(String.valueOf(sorceOnlyNum));
row.add(String.valueOf(targetOnlyNum));
row.add(String.valueOf(totalRecordNum * fields.size()));
row.add(isError?"不通过":"通过");
// 写入Excel的summarySheet

2
src/main/java/com/celnet/datadump/util/BulkUtil.java
View File

@ -214,7 +214,7 @@ public class BulkUtil {
*/
public static void waitForBulkV2JobCompletion(String url, String jobId, BulkConnection connection) throws Exception {
CloseableHttpClient httpClient = HttpClients.createDefault();
String jobStatusUrl = url + "services/data/v56.0/jobs/query/" + jobId;
String jobStatusUrl = url + "services/data/v56.0/jobs/ingest/" + jobId;
boolean jobCompleted = false;
while (!jobCompleted) {