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

【feat】 提交bulk v2有关

Browse Source 
(cherry picked from commit 713e545420)
This commit is contained in:
Kris 2025-10-24 11:56:39 +08:00
parent f653813c5c
commit f62d1f65be
7 changed files with 286 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/com/celnet/datadump/config/SalesforceConnect.java
View File

@ -54,12 +54,16 @@ public class SalesforceConnect {
if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", String.valueOf(map1.get("value")));
}
if ("SOURCE_ORG_SEESION".equals(map1.get("code"))) {
map.put("seesionId", (String) map1.get("value"));
}
}
String username = map.get("username");
ConnectorConfig config = new ConnectorConfig();
config.setUsername(username);
config.setPassword(map.get("password"));
config.setSessionId(map.get("seesionId"));
String url = map.get("url");
config.setAuthEndpoint(url);
config.setServiceEndpoint(url);
@ -106,6 +110,9 @@ public class SalesforceConnect {
if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", String.valueOf(map1.get("value")));
}
if ("SOURCE_ORG_SEESION".equals(map1.get("code"))) {
map.put("seesionId", (String) map1.get("value"));
}
}
return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url"));
} catch (Exception e) {

7
src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java
View File

@ -49,12 +49,16 @@ public class SalesforceTargetConnect {
if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", (String) map1.get("value"));
}
if ("TARGET_ORG_SEESION".equals(map1.get("code"))) {
map.put("seesionId", (String) map1.get("value"));
}
}
String username = map.get("username");
ConnectorConfig config = new ConnectorConfig();
config.setUsername(username);
config.setPassword(map.get("password"));
config.setSessionId(map.get("seesionId"));
String url = map.get("url");
config.setAuthEndpoint(url);
config.setServiceEndpoint(url);
@ -105,6 +109,9 @@ public class SalesforceTargetConnect {
if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", (String) map1.get("value"));
}
if ("TARGET_ORG_SEESION".equals(map1.get("code"))) {
map.put("seesionId", (String) map1.get("value"));
}
}
return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url"));
} catch (Exception e) {

17
src/main/java/com/celnet/datadump/controller/JobController.java
View File

@ -337,6 +337,23 @@ public class JobController {
return commonBatchService.dumpBigObject(param);
}
@PostMapping("/dataDumpBigObjectV2Job")
@ApiOperation("存量任务（BigObject_v2）")
@LogServiceAnnotation(operateType = OperateTypeConstant.TYPE_INSERT, remark = "存量任务（BigObject_v2）")
public ReturnT<String> dataDumpBigObjectV2Job(String paramStr) throws Exception {
log.info("dataDumpBigObjectJob 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.dumpBigObjectV2(param);
}
/**
* bulk批量更新大数据量数据
* @param paramStr

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

@ -138,6 +138,29 @@ public class DataDumpNewJob {
return commonBatchService.dumpBigObject(param);
}
/**
* 存量任务（BigObject_V2）
*
* @param paramStr 参数json
* @return result
*/
@XxlJob("dataDumpBigObjectV2Job")
public ReturnT<String> dataDumpBigObjectV2Job(String paramStr) throws Exception {
log.info("dataDumpBigObjectV2Job execute start ..................");
SalesforceParam param = new SalesforceParam();
try {
if (StringUtils.isNotBlank(paramStr)) {
param = JSON.parseObject(paramStr, SalesforceParam.class);
}
} catch (Throwable throwable) {
return new ReturnT<>(500, "参数解析失败！");
}
// 参数转换
param.setBeginCreateDate(param.getBeginDate());
param.setEndCreateDate(param.getEndDate());
return commonBatchService.dumpBigObjectV2(param);
}
/**
* 导入BigObject数据到目标ORG
*

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

@ -8,7 +8,10 @@ public interface CommonBatchService {
ReturnT<String> dumpBatch(SalesforceParam param) throws Exception;
ReturnT<String> dumpBigObject(SalesforceParam param) throws Exception;
ReturnT<String> dumpBigObjectV2(SalesforceParam param) throws Exception;
ReturnT<String> importBigObject(SalesforceParam param) throws Exception;
}

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

@ -155,7 +155,23 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> dumpBigObjectV2(SalesforceParam param) throws Exception {
try {
// 手动任务
ReturnT<String> result = manualBigObjectDumpV2(param);
if (result != null) {
return result;
}
} catch (Throwable throwable) {
log.error("dump error", throwable);
throw throwable;
}
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> importBigObject(SalesforceParam param) throws Exception {
try {
@ -310,6 +326,157 @@ public class CommonBatchServiceImpl implements CommonBatchService {
return ReturnT.SUCCESS;
}
private ReturnT<String> manualBigObjectDumpV2(SalesforceParam param) throws Exception {
String api = param.getApi();
BulkConnection bulkConnect = salesforceConnect.createBulkConnect();
String jobId = null;
try {
log.info("开始使用Bulk API v2导出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();
log.info("构建查询SQL: {}", sql);
// 创建Bulk V2 API作业
String restEndpoint = bulkConnect.getConfig().getRestEndpoint();
String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
log.debug("Salesforce实例URL: {}", url);
jobId = BulkUtil.createBulkV2QueryJob(url, sql, bulkConnect);
log.info("创建Bulk V2查询作业成功, 作业ID: {}", jobId);
// 等待作业完成
BulkUtil.waitForBulkV2QueryJobCompletion(url, jobId, bulkConnect);
log.info("Bulk V2查询作业已完成, 作业ID: {}", jobId);
if (Const.FILE_TYPE == FileType.SERVER) {
// 检测路径是否存在 不存在则创建
File excel = new File(Const.SERVER_FILE_PATH + "/" + api);
if (!excel.exists()) {
log.info("创建API目录: {}", excel.getAbsolutePath());
boolean mkdir = excel.mkdir();
if (!mkdir) {
log.info("创建文件存储目录失败！");
}
}
}
// 使用确定性命名策略，避免使用随机时间戳
String fileName = Const.SERVER_FILE_PATH + "/" + api + "/" + api + "_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss");
final long maxFileSize = 50 * 1024 * 1024; // 50MB
int fileIndex = 0;
Path currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv");
log.info("开始写入数据到文件: {}", fileName);
BufferedWriter bufferedWriter = null;
FileWriter fileWriter = null;
try {
fileWriter = new FileWriter(currentFilePath.toFile(), true);
bufferedWriter = new BufferedWriter(fileWriter);
long currentFileSize = 0;
String headerLine = null;
String locator = null;
boolean hasMoreData = true;
int totalRecords = 0;
while (hasMoreData) {
// 获取查询结果
BulkV2QueryResult result = BulkUtil.getBulkV2QueryJobResults(url, jobId, bulkConnect, locator);
String csvContent = result.getCsvContent();
locator = result.getLocator();
int recordCount = result.getRecordCount();
totalRecords += recordCount;
log.info("获取到{}条记录，总计{}条记录", recordCount, totalRecords);
if (recordCount > 0) {
// 处理CSV内容
try (BufferedReader bufferedReader = new BufferedReader(new StringReader(csvContent))) {
String line;
int lineCount = 0;
while ((line = bufferedReader.readLine()) != null) {
if (headerLine == 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");
headerLine = line;
}
lineCount++;
byte[] lineBytes = (line + "\n").getBytes(StandardCharsets.UTF_8);
// 如果添加这行会超出文件大小限制，则创建新文件
if (currentFileSize + lineBytes.length > maxFileSize && currentFileSize > 0) {
// 关闭当前文件
if (bufferedWriter != null) {
bufferedWriter.close();
}
if (fileWriter != null) {
fileWriter.close();
}
log.info("当前文件已达到大小限制, 创建新文件. 文件路径: {}, 数据行数: {}", currentFilePath.toString(), lineCount);
// 创建新文件
fileIndex++;
currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv");
fileWriter = new FileWriter(currentFilePath.toFile(), true);
bufferedWriter = new BufferedWriter(fileWriter);
currentFileSize = 0;
lineCount = 1; // 重置行计数，新文件将写入标题行
// 在新文件中写入标题行
if (headerLine != null) {
bufferedWriter.write(headerLine + "\n");
currentFileSize += (headerLine + "\n").getBytes(StandardCharsets.UTF_8).length;
}
}
// 写入数据
bufferedWriter.write(line + "\n");
currentFileSize += lineBytes.length;
}
}
}
// 检查是否还有更多数据
hasMoreData = locator != null && !locator.isEmpty() && !"null".equals(locator) && StringUtils.isNotEmpty(locator);
if (hasMoreData) {
log.info("还有更多数据需要获取，locator: {}", locator);
}
}
log.info("所有数据写入完成, 总文件数: {}, 总记录数: {}", fileIndex + 1, totalRecords);
} finally {
if (bufferedWriter != null) {
bufferedWriter.close();
}
if (fileWriter != null) {
fileWriter.close();
}
}
} catch (Exception e) {
log.error("使用Bulk API v2导出BigObject数据失败, API: {}", api, e);
throw e;
}
log.info("BigObject数据导出任务完成, API: {}", api);
return ReturnT.SUCCESS;
}
private ReturnT<String> manualBigObjectImport(SalesforceParam param) throws Exception {
String api = param.getApi();

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

@ -205,6 +205,47 @@ public class BulkUtil {
}
}
/**
* 通过轮询作业状态等待 Bulk API 2.0 作业完成。
*
* @param jobId 要检查的作业ID。
* @param connection 用于获取会话信息的BulkConnection。
* @throws Exception 轮询过程中发生错误时抛出异常。
*/
public static void waitForBulkV2QueryJobCompletion(String url, String jobId, BulkConnection connection) throws Exception {
CloseableHttpClient httpClient = HttpClients.createDefault();
String jobStatusUrl = url + "services/data/v56.0/jobs/query/" + jobId;
boolean jobCompleted = false;
while (!jobCompleted) {
HttpGet httpGet = new HttpGet(jobStatusUrl);
httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
CloseableHttpResponse response = httpClient.execute(httpGet);
try {
int statusCode = response.getStatusLine().getStatusCode();
if (statusCode == 200) {
String responseBody = EntityUtils.toString(response.getEntity());
// 简单解析响应体中的状态字段，实际项目中建议使用JSON库如Jackson
if (responseBody.contains("\"state\":\"JobComplete\"")) {
System.out.println("作业成功完成。");
jobCompleted = true;
} else if (responseBody.contains("\"state\":\"Failed\"") || responseBody.contains("\"state\":\"Aborted\"")) {
throw new RuntimeException("作业失败或已被中止。响应内容: " + responseBody);
} else {
System.out.println("作业仍在运行中。等待中...");
Thread.sleep(10000); // 等待10秒后再次检查
}
} else {
String errorBody = EntityUtils.toString(response.getEntity());
throw new RuntimeException("获取作业状态失败，状态码: " + statusCode + "，错误: " + errorBody);
}
} finally {
response.close();
}
}
httpClient.close();
}
/**
* 通过轮询作业状态等待 Bulk API 2.0 作业完成。
*
@ -257,8 +298,26 @@ public class BulkUtil {
* @throws Exception 如果在获取结果过程中发生错误
*/
public static BulkV2QueryResult getBulkV2QueryJobResults(String url, String jobId, BulkConnection connection) throws Exception {
return getBulkV2QueryJobResults(url, jobId, connection, null);
}
/**
* 获取 Bulk API 2.0 查询作业的结果
*
* @param url Salesforce实例URL
* @param jobId 作业ID
* @param connection BulkConnection used to get session info.
* @param locator 分页定位器，用于获取下一批结果
* @return 查询结果的字符串内容
* @throws Exception 如果在获取结果过程中发生错误
*/
public static BulkV2QueryResult getBulkV2QueryJobResults(String url, String jobId, BulkConnection connection, String locator) throws Exception {
CloseableHttpClient httpClient = HttpClients.createDefault();
String resultsUrl = url + "/services/data/v56.0/jobs/query/" + jobId + "/results";
// 只有当locator不为null且不为空时才添加到URL中
if (locator != null && !locator.isEmpty() && !"null".equals(locator)) {
resultsUrl += "?locator=" + java.net.URLEncoder.encode(locator, "UTF-8");
}
HttpGet httpGet = new HttpGet(resultsUrl);
httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
httpGet.setHeader("Content-Type", "application/json");
@ -274,6 +333,7 @@ public class BulkUtil {
String newLocator = (locatorHeader != null) ? locatorHeader.getValue() : null;
int numberOfRecords = (numRecordsHeader != null) ? Integer.parseInt(numRecordsHeader.getValue()) : 0;
String csvContent = EntityUtils.toString(response.getEntity());
System.out.println("获取作业结果成功。" + response);
return new BulkV2QueryResult(csvContent, newLocator, numberOfRecords);
} else {
String errorBody = EntityUtils.toString(response.getEntity());