[feat] bigobject数据下载本地服务器csv文件
This commit is contained in:
parent
c63c1032fa
commit
6c8f75f57f
@ -144,7 +144,7 @@ public class JobController {
|
||||
@PostMapping("/createApi")
|
||||
@ApiOperation("创建表结构")
|
||||
public Result createApi(String paramStr) throws Exception {
|
||||
log.info("dataImportBatchJob execute start ..................");
|
||||
log.info("createApi execute start ..................");
|
||||
SalesforceParam param = new SalesforceParam();
|
||||
if (StringUtils.isNotBlank(paramStr)) {
|
||||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||||
@ -161,6 +161,31 @@ public class JobController {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建表结构
|
||||
*
|
||||
* @return result
|
||||
*/
|
||||
@PostMapping("/createBigObjectApi")
|
||||
@ApiOperation("创建BigOBject表结构")
|
||||
public Result createBigObjectApi(String paramStr) throws Exception {
|
||||
log.info("createBigObjectApi execute start ..................");
|
||||
SalesforceParam param = new SalesforceParam();
|
||||
if (StringUtils.isNotBlank(paramStr)) {
|
||||
param = JSON.parseObject(paramStr, SalesforceParam.class);
|
||||
}
|
||||
param.setType(1);
|
||||
// 参数转换
|
||||
param.setBeginCreateDate(param.getBeginDate());
|
||||
param.setEndCreateDate(param.getEndDate());
|
||||
ReturnT<String> returnT = commonService.createBigObjectApi(param);
|
||||
if (returnT.getCode() == ReturnT.SUCCESS_CODE) {
|
||||
return Result.success();
|
||||
} else {
|
||||
return Result.fail(returnT.getMsg());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 存量任务
|
||||
*
|
||||
@ -251,7 +276,7 @@ public class JobController {
|
||||
}
|
||||
|
||||
/**
|
||||
* bulk批量大数据生成newSFID
|
||||
* 存量任务（大数据量）
|
||||
* @param paramStr
|
||||
* @author kris
|
||||
* @return
|
||||
@ -277,6 +302,23 @@ public class JobController {
|
||||
return commonBatchService.dumpBatch(param);
|
||||
}
|
||||
|
||||
@PostMapping("/dataDumpBigObjectJob")
|
||||
@ApiOperation("存量任务（BigObject）")
|
||||
@LogServiceAnnotation(operateType = OperateTypeConstant.TYPE_INSERT, remark = "存量任务（BigObject）")
|
||||
public ReturnT<String> dataDumpBigObjectJob(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.dumpBigObject(param);
|
||||
}
|
||||
|
||||
/**
|
||||
* bulk批量更新大数据量数据
|
||||
* @param paramStr
|
||||
|
||||
@ -17,10 +17,7 @@ import com.celnet.datadump.param.DataDumpParam;
|
||||
import com.celnet.datadump.param.DataDumpSpecialParam;
|
||||
import com.celnet.datadump.param.SalesforceParam;
|
||||
import com.celnet.datadump.service.*;
|
||||
import com.celnet.datadump.util.BulkUtil;
|
||||
import com.celnet.datadump.util.DataUtil;
|
||||
import com.celnet.datadump.util.EmailUtil;
|
||||
import com.celnet.datadump.util.SqlUtil;
|
||||
import com.celnet.datadump.util.*;
|
||||
import com.google.common.collect.Lists;
|
||||
import com.google.common.collect.Maps;
|
||||
|
||||
@ -45,7 +42,12 @@ import org.apache.commons.lang3.time.DateUtils;
|
||||
import org.apache.http.Header;
|
||||
import org.apache.http.client.methods.CloseableHttpResponse;
|
||||
import org.apache.http.client.methods.HttpGet;
|
||||
import org.apache.http.client.methods.HttpPatch;
|
||||
import org.apache.http.client.methods.HttpPost;
|
||||
import org.apache.http.client.methods.HttpPut;
|
||||
import org.apache.http.client.methods.HttpPut;
|
||||
import org.apache.http.entity.FileEntity;
|
||||
import org.apache.http.entity.StringEntity;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.apache.http.impl.client.HttpClients;
|
||||
import org.apache.http.util.EntityUtils;
|
||||
@ -56,6 +58,8 @@ import javax.annotation.PostConstruct;
|
||||
import java.io.*;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
@ -94,6 +98,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
|
||||
// 批次处理记录数，默认为10000
|
||||
private int batchProcessCount = 10000;
|
||||
|
||||
private static final String BIGOBJECT_FILE_PATH = "bigObject/";
|
||||
|
||||
|
||||
@Autowired
|
||||
private SystemConfigService systemConfigService;
|
||||
|
||||
@ -176,231 +183,241 @@ public class CommonBatchServiceImpl implements CommonBatchService {
|
||||
String api = param.getApi();
|
||||
|
||||
BulkConnection bulkConnect = salesforceConnect.createBulkConnect();
|
||||
|
||||
String beginDateStr = null;
|
||||
String endDateStr = null;
|
||||
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
|
||||
Date beginDate = param.getBeginCreateDate();
|
||||
Date endDate = param.getEndCreateDate();
|
||||
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
|
||||
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
|
||||
}
|
||||
JobInfo job = null;
|
||||
|
||||
try {
|
||||
List<SalesforceParam> salesforceParams = null;
|
||||
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
|
||||
dbQw.eq("name", api);
|
||||
if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
|
||||
dbQw.eq("sync_start_date", beginDateStr); // 等于开始时间
|
||||
dbQw.eq("sync_end_date", endDateStr); // 等于结束时间
|
||||
}
|
||||
List<DataBatch> list = dataBatchService.list(dbQw);
|
||||
AtomicInteger batch = new AtomicInteger(1);
|
||||
if (CollectionUtils.isNotEmpty(list)) {
|
||||
salesforceParams = list.stream().map(t -> {
|
||||
SalesforceParam salesforceParam = param.clone();
|
||||
salesforceParam.setApi(t.getName());
|
||||
salesforceParam.setBeginCreateDate(t.getSyncStartDate());
|
||||
salesforceParam.setEndCreateDate(t.getSyncEndDate());
|
||||
salesforceParam.setBatch(batch.getAndIncrement());
|
||||
return salesforceParam;
|
||||
}).collect(Collectors.toList());
|
||||
|
||||
log.info("开始导出BigObject数据, API: {}", api);
|
||||
|
||||
// 检测路径是否存在 不存在则创建
|
||||
File excel = new File( param.getApi() + "/");
|
||||
if (!excel.exists()) {
|
||||
System.out.println("创建目录: " + param.getApi() + "/");
|
||||
boolean mkdir = excel.mkdir();
|
||||
}
|
||||
|
||||
// 手动任务优先执行
|
||||
for (SalesforceParam salesforceParam : salesforceParams) {
|
||||
if (param.getIsSingleThread()){
|
||||
manualBigObject(salesforceParam, bulkConnect);
|
||||
}else {
|
||||
Future<?> future = salesforceExecutor.execute(() -> {
|
||||
try {
|
||||
manualBigObject(salesforceParam, bulkConnect);
|
||||
} catch (Throwable throwable) {
|
||||
log.error("salesforceExecutor error", throwable);
|
||||
throw new RuntimeException(throwable);
|
||||
List<DataField> fieldList = dataFieldService.list(new QueryWrapper<DataField>().eq("api", param.getApi()));
|
||||
String fieldStr = fieldList.stream()
|
||||
.map(DataField::getField)
|
||||
.collect(Collectors.joining(", "));
|
||||
|
||||
String sql = "select " + fieldStr + " from " + param.getApi() ;
|
||||
log.info("构建查询SQL: {}", sql);
|
||||
|
||||
job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll);
|
||||
log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
|
||||
|
||||
BatchInfo batchInfo = bulkConnect.createBatchFromStream(job, new ByteArrayInputStream(sql.getBytes(StandardCharsets.UTF_8)));
|
||||
log.info("创建批次成功, 批次ID: {}", batchInfo.getId());
|
||||
|
||||
int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
|
||||
log.info("批次处理完成, 处理记录数: {}", completionOne);
|
||||
|
||||
if (completionOne == 0){
|
||||
log.info("无更多数据, 结束处理");
|
||||
BulkUtil.closeJob(bulkConnect, job.getId());
|
||||
return ReturnT.SUCCESS;
|
||||
}
|
||||
|
||||
QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
|
||||
log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
|
||||
|
||||
|
||||
|
||||
// 使用确定性命名策略，避免使用随机时间戳
|
||||
String fileName = param.getApi() + "/" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss");
|
||||
final long maxFileSize = 50 * 1024 * 1024; // 50MB
|
||||
|
||||
int fileIndex = 0;
|
||||
String currentFilePath = fileName + "_" + fileIndex + ".csv";
|
||||
|
||||
log.info("开始写入数据到文件 : {}", fileName);
|
||||
|
||||
BufferedWriter bufferedWriter = null;
|
||||
FileWriter fileWriter = null;
|
||||
|
||||
try {
|
||||
fileWriter = new FileWriter(currentFilePath, true);
|
||||
bufferedWriter = new BufferedWriter(fileWriter);
|
||||
long currentFileSize = 0;
|
||||
|
||||
int resultIndex = 0;
|
||||
for (final String resultId : queryResultList.getResult()) {
|
||||
log.info("处理结果ID: {}, 进度: {}/{}", resultId, ++resultIndex, queryResultList.getResult().length);
|
||||
|
||||
InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId);
|
||||
|
||||
try (BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(resultStream, StandardCharsets.UTF_8))) {
|
||||
String line;
|
||||
int lineCount = 0;
|
||||
while ((line = bufferedReader.readLine()) != null) {
|
||||
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, lineCount);
|
||||
|
||||
// 创建新文件
|
||||
fileIndex++;
|
||||
currentFilePath = fileName + "_" + fileIndex + ".csv";
|
||||
fileWriter = new FileWriter(currentFilePath, true);
|
||||
bufferedWriter = new BufferedWriter(fileWriter);
|
||||
currentFileSize = 0;
|
||||
lineCount = 0;
|
||||
}
|
||||
|
||||
// 写入数据
|
||||
bufferedWriter.write(line + "\n");
|
||||
currentFileSize += lineBytes.length;
|
||||
}
|
||||
}, salesforceParam.getBatch(), 1);
|
||||
futures.add(future);
|
||||
log.info("结果ID {} 处理完成", resultId);
|
||||
}
|
||||
}
|
||||
log.info("所有数据写入完成, 总文件数: {}", fileIndex + 1);
|
||||
} finally {
|
||||
if (bufferedWriter != null) {
|
||||
bufferedWriter.close();
|
||||
}
|
||||
if (fileWriter != null) {
|
||||
fileWriter.close();
|
||||
}
|
||||
}
|
||||
// 等待当前所有线程执行完成
|
||||
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
|
||||
|
||||
}catch (InterruptedException interruptedException){
|
||||
return ReturnT.FAIL;
|
||||
BulkUtil.closeJob(bulkConnect, job.getId());
|
||||
log.info("关闭作业成功, 作业ID: {}", job.getId());
|
||||
|
||||
}catch (Exception e) {
|
||||
log.error("导出BigObject数据失败, API: {}", api, e);
|
||||
throw e;
|
||||
}
|
||||
log.info("BigObject数据导出任务完成, API: {}", api);
|
||||
return ReturnT.SUCCESS;
|
||||
}
|
||||
|
||||
private void manualBigObject(SalesforceParam param, BulkConnection bulkConnect) throws Exception {
|
||||
|
||||
String sql = "select * from " + param.getApi() + " where CreatedDate >= '" + param.getBeginCreateDate() + "' and CreatedDate <= '" + param.getEndCreateDate() + "'";
|
||||
|
||||
log.info("BigObject query SQL: {}", sql);
|
||||
XxlJobLogger.log("BigObject query SQL: {}", sql);
|
||||
|
||||
// 获取Salesforce实例URL
|
||||
String restEndpoint = bulkConnect.getConfig().getRestEndpoint();
|
||||
String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
|
||||
|
||||
// 创建Bulk API 2.0查询作业
|
||||
String jobId = BulkUtil.createBulkV2Job(url, sql, bulkConnect);
|
||||
log.info("Created Bulk API 2.0 job with ID: {}", jobId);
|
||||
|
||||
// 等待作业完成
|
||||
BulkUtil.waitForBulkV2JobCompletion(url, jobId, bulkConnect);
|
||||
|
||||
// 获取查询结果
|
||||
String csvResult = BulkUtil.getBulkV2QueryJobResults(url, jobId, bulkConnect);
|
||||
|
||||
// 保存到本地文件
|
||||
String fileName = param.getApi() + "_" + System.currentTimeMillis() + ".csv";
|
||||
String filePath = "bigobject/" + fileName;
|
||||
|
||||
// 确保目录存在
|
||||
File directory = new File("bigobject");
|
||||
if (!directory.exists()) {
|
||||
directory.mkdirs();
|
||||
}
|
||||
|
||||
// 写入文件
|
||||
try (FileWriter fileWriter = new FileWriter(filePath);
|
||||
BufferedWriter bufferedWriter = new BufferedWriter(fileWriter)) {
|
||||
bufferedWriter.write(csvResult);
|
||||
}
|
||||
|
||||
log.info("BigObject data saved to file: {}", filePath);
|
||||
XxlJobLogger.log("BigObject data saved to file: {}", filePath);
|
||||
}
|
||||
|
||||
private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future<?>> futures) throws Exception {
|
||||
private ReturnT<String> manualBigObjectImport(SalesforceParam param, List<Future<?>> futures) throws Exception {
|
||||
String api = param.getApi();
|
||||
|
||||
// 创建目标ORG连接
|
||||
BulkConnection targetBulkConnect = salesforceTargetConnect.createBulkConnect();
|
||||
|
||||
String beginDateStr = null;
|
||||
String endDateStr = null;
|
||||
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
|
||||
Date beginDate = param.getBeginCreateDate();
|
||||
Date endDate = param.getEndCreateDate();
|
||||
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
|
||||
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
|
||||
}
|
||||
|
||||
try {
|
||||
List<SalesforceParam> salesforceParams = null;
|
||||
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
|
||||
dbQw.eq("name", api);
|
||||
if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
|
||||
dbQw.eq("sync_start_date", beginDateStr); // 等于开始时间
|
||||
dbQw.eq("sync_end_date", endDateStr); // 等于结束时间
|
||||
}
|
||||
List<DataBatch> list = dataBatchService.list(dbQw);
|
||||
AtomicInteger batch = new AtomicInteger(1);
|
||||
if (CollectionUtils.isNotEmpty(list)) {
|
||||
salesforceParams = list.stream().map(t -> {
|
||||
SalesforceParam salesforceParam = param.clone();
|
||||
salesforceParam.setApi(t.getName());
|
||||
salesforceParam.setBeginCreateDate(t.getSyncStartDate());
|
||||
salesforceParam.setEndCreateDate(t.getSyncEndDate());
|
||||
salesforceParam.setBatch(batch.getAndIncrement());
|
||||
return salesforceParam;
|
||||
}).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
// 手动任务优先执行
|
||||
for (SalesforceParam salesforceParam : salesforceParams) {
|
||||
if (param.getIsSingleThread()){
|
||||
manualImportBigObject(salesforceParam, targetBulkConnect);
|
||||
}else {
|
||||
Future<?> future = salesforceExecutor.execute(() -> {
|
||||
try {
|
||||
manualImportBigObject(salesforceParam, targetBulkConnect);
|
||||
} catch (Throwable throwable) {
|
||||
log.error("salesforceExecutor error", throwable);
|
||||
throw new RuntimeException(throwable);
|
||||
}
|
||||
}, salesforceParam.getBatch(), 1);
|
||||
futures.add(future);
|
||||
}
|
||||
}
|
||||
// 等待当前所有线程执行完成
|
||||
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
|
||||
|
||||
}catch (InterruptedException interruptedException){
|
||||
return ReturnT.FAIL;
|
||||
}catch (Exception e) {
|
||||
throw e;
|
||||
}
|
||||
return ReturnT.SUCCESS;
|
||||
}
|
||||
|
||||
private void manualImportBigObject(SalesforceParam param, BulkConnection targetBulkConnect) throws Exception {
|
||||
String api = param.getApi();
|
||||
|
||||
// 构造CSV文件路径 - 使用与导出时相同的文件名格式（但需要知道确切的时间戳）
|
||||
// 这里应该查找最新的相关文件，而不是随机生成一个文件名
|
||||
File directory = new File("bigobject");
|
||||
File directory = new File(api);
|
||||
if (!directory.exists() || !directory.isDirectory()) {
|
||||
log.error("BigObject目录不存在: {}", directory.getAbsolutePath());
|
||||
throw new FileNotFoundException("BigObject目录不存在: " + directory.getAbsolutePath());
|
||||
throw new FileNotFoundException(api + "目录不存在: " + directory.getAbsolutePath());
|
||||
}
|
||||
|
||||
// 查找匹配的CSV文件（最近的）
|
||||
File[] files = directory.listFiles((dir, name) ->
|
||||
name.startsWith(api + "_") && name.endsWith(".csv"));
|
||||
File[] files = directory.listFiles((dir, name) -> name.endsWith(".csv"));
|
||||
|
||||
if (files == null || files.length == 0) {
|
||||
log.error("未找到API {} 的CSV文件", api);
|
||||
throw new FileNotFoundException("未找到API " + api + " 的CSV文件");
|
||||
}
|
||||
|
||||
// 选择最新的文件
|
||||
File csvFile = Arrays.stream(files)
|
||||
.max(Comparator.comparingLong(File::lastModified))
|
||||
.orElse(null);
|
||||
|
||||
if (csvFile == null) {
|
||||
log.error("未找到API {} 的CSV文件", api);
|
||||
throw new FileNotFoundException("未找到API " + api + " 的CSV文件");
|
||||
}
|
||||
|
||||
String filePath = csvFile.getAbsolutePath();
|
||||
log.info("找到CSV文件: {}", filePath);
|
||||
|
||||
log.info("开始导入BigObject数据到目标ORG: {}", api);
|
||||
|
||||
// 创建Bulk API作业
|
||||
JobInfo job = BulkUtil.createJob(targetBulkConnect, api, OperationEnum.insert);
|
||||
log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
|
||||
// 使用Bulk V2 API导入所有文件
|
||||
for (File csvFile : files) {
|
||||
String filePath = csvFile.getAbsolutePath();
|
||||
log.info("处理CSV文件: {}", filePath);
|
||||
|
||||
// 从CSV文件创建批次
|
||||
List<BatchInfo> batchInfos = BulkUtil.createBatchesFromCSVFile(targetBulkConnect, job, filePath);
|
||||
log.info("创建批次成功, 批次数: {}", batchInfos.size());
|
||||
// 创建Bulk V2 API作业
|
||||
String restEndpoint = targetBulkConnect.getConfig().getRestEndpoint();
|
||||
String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
|
||||
|
||||
// 等待作业完成
|
||||
BulkUtil.awaitCompletion(targetBulkConnect, job, batchInfos);
|
||||
log.info("数据导入完成");
|
||||
// 创建Bulk V2作业
|
||||
String jobId = BulkUtil.createBulkV2InsertJob(url, api, targetBulkConnect);
|
||||
log.info("创建Bulk V2作业成功, 作业ID: {}", jobId);
|
||||
|
||||
// 检查作业状态
|
||||
job = targetBulkConnect.getJobStatus(job.getId());
|
||||
if (job.getState() == JobStateEnum.Failed) {
|
||||
log.error("Bulk作业失败, 作业ID: {}", job.getId());
|
||||
throw new RuntimeException("Bulk作业失败 ");
|
||||
// 上传CSV文件数据
|
||||
uploadCsvDataToJob(url, jobId, targetBulkConnect, filePath);
|
||||
|
||||
// 关闭作业
|
||||
closeBulkV2Job(url, jobId, targetBulkConnect);
|
||||
|
||||
// 等待作业完成
|
||||
BulkUtil.waitForBulkV2JobCompletion(url, jobId, targetBulkConnect);
|
||||
log.info("文件 {} 数据导入完成", csvFile.getName());
|
||||
}
|
||||
|
||||
// 关闭作业
|
||||
BulkUtil.closeJob(targetBulkConnect, job.getId());
|
||||
log.info("关闭作业成功, 作业ID: {}", job.getId());
|
||||
log.info("所有BigObject数据导入完成: {}", api);
|
||||
return ReturnT.SUCCESS;
|
||||
}
|
||||
|
||||
// 记录导入统计信息
|
||||
int totalRecords = batchInfos.stream()
|
||||
.mapToInt(BatchInfo::getNumberRecordsProcessed)
|
||||
.sum();
|
||||
log.info("成功导入 {} 条记录到 {}", totalRecords, api);
|
||||
/**
|
||||
* 上传CSV数据到Bulk API 2.0作业
|
||||
*
|
||||
* @param url Salesforce实例URL
|
||||
* @param jobId 作业ID
|
||||
* @param connection BulkConnection连接对象
|
||||
* @param csvFilePath CSV文件路径
|
||||
* @throws Exception 上传过程中发生错误时抛出异常
|
||||
*/
|
||||
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 httpPut = new HttpPut(uploadUrl);
|
||||
httpPut.setHeader("Content-Type", "text/csv");
|
||||
httpPut.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
|
||||
|
||||
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);
|
||||
}
|
||||
} finally {
|
||||
response.close();
|
||||
httpClient.close();
|
||||
}
|
||||
log.info("数据上传成功，作业ID: {}", jobId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 关闭Bulk API 2.0作业
|
||||
*
|
||||
* @param url Salesforce实例URL
|
||||
* @param jobId 作业ID
|
||||
* @param connection BulkConnection连接对象
|
||||
* @throws Exception 关闭作业时发生错误时抛出异常
|
||||
*/
|
||||
private void closeBulkV2Job(String url, String jobId, BulkConnection connection) throws Exception {
|
||||
CloseableHttpClient httpClient = HttpClients.createDefault();
|
||||
String closeUrl = url + "services/data/v56.0/jobs/ingest/" + jobId;
|
||||
HttpPatch httpPatch = new HttpPatch(closeUrl);
|
||||
httpPatch.setHeader("Content-Type", "application/json");
|
||||
httpPatch.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
|
||||
|
||||
// 构建关闭作业的 JSON 请求体
|
||||
String jsonRequest = "{\"state\":\"UploadComplete\"}";
|
||||
StringEntity stringEntity = new StringEntity(jsonRequest, "UTF-8");
|
||||
httpPatch.setEntity(stringEntity);
|
||||
|
||||
CloseableHttpResponse response = httpClient.execute(httpPatch);
|
||||
try {
|
||||
int statusCode = response.getStatusLine().getStatusCode();
|
||||
if (statusCode != 200) {
|
||||
String errorBody = EntityUtils.toString(response.getEntity());
|
||||
throw new RuntimeException("Failed to close Bulk API 2.0 job, status code: " + statusCode + ", error: " + errorBody);
|
||||
}
|
||||
} finally {
|
||||
response.close();
|
||||
httpClient.close();
|
||||
}
|
||||
log.info("作业关闭成功，作业ID: {}", jobId);
|
||||
}
|
||||
|
||||
private ReturnT<String> manualBatchDump(SalesforceParam param, List<Future<?>> futures) throws InterruptedException {
|
||||
@ -853,7 +870,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
|
||||
|
||||
String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
|
||||
|
||||
v2JobId = BulkUtil.createBulkV2Job(url,sql, bulkConnect);
|
||||
v2JobId = BulkUtil.createBulkV2QueryJob(url,sql, bulkConnect);
|
||||
|
||||
BulkUtil.waitForBulkV2JobCompletion(url,v2JobId, bulkConnect);
|
||||
|
||||
@ -934,7 +951,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
|
||||
}
|
||||
}
|
||||
|
||||
}catch (Exception e){
|
||||
}catch (Exception e){
|
||||
log.error("processQueryResults error", e);
|
||||
}
|
||||
return totalProcessedRecords;
|
||||
|
||||
@ -1607,64 +1607,51 @@ public class CommonServiceImpl implements CommonService {
|
||||
public ReturnT<String> createBigObjectApi(SalesforceParam param) throws Exception {
|
||||
|
||||
String api = param.getApi();
|
||||
log.info("开始创建BigObject API: {}", api);
|
||||
|
||||
PartnerConnection connection = salesforceConnect.createConnect();
|
||||
log.debug("Salesforce连接创建成功");
|
||||
|
||||
DescribeSObjectResult dsr = connection.describeSObject(api);
|
||||
log.info("获取对象描述信息成功, API: {}, Label: {}", api, dsr.getLabel());
|
||||
|
||||
|
||||
// 查询data_field表，如果有数据则不执行，没有数据执行
|
||||
QueryWrapper<DataField> queryWrapper = new QueryWrapper<>();
|
||||
queryWrapper.eq("api", api);
|
||||
long count = dataFieldService.count(queryWrapper);
|
||||
|
||||
if (count == 0) {
|
||||
List<DataField> fieldList = Lists.newArrayList();
|
||||
log.debug("开始处理字段信息，字段总数: {}", dsr.getFields().length);
|
||||
for (Field field : dsr.getFields()) {
|
||||
DataField dataField = new DataField();
|
||||
dataField.setApi(api);
|
||||
dataField.setSfType(field.getType().toString());
|
||||
dataField.setField(field.getName());
|
||||
dataField.setName(field.getLabel());
|
||||
dataField.setIsCreateable(field.getCreateable());
|
||||
dataField.setIsUpdateable(field.getUpdateable());
|
||||
dataField.setIsNillable(field.getNillable());
|
||||
dataField.setIsDefaultedOnCreate(field.getDefaultedOnCreate());
|
||||
fieldList.add(dataField);
|
||||
}
|
||||
log.info("字段信息处理完成，共处理{}个字段", fieldList.size());
|
||||
|
||||
dataFieldService.saveBatch(fieldList);
|
||||
log.info("字段信息批量保存完成");
|
||||
} else {
|
||||
log.info("API {} 的字段信息已存在，跳过字段处理", api);
|
||||
}
|
||||
|
||||
DataObject dataObject = dataObjectService.getById(api);
|
||||
dataObject.setLabel(dsr.getLabel());
|
||||
dataObject.setLastUpdateDate(new Date());
|
||||
dataObjectService.updateById(dataObject);
|
||||
log.info("数据对象信息更新完成");
|
||||
|
||||
Date startDate = dataObject.getLastUpdateDate() == null ? DataUtil.DEFAULT_BEGIN_DATE : dataObject.getLastUpdateDate();
|
||||
Date endCreateDate = DateUtils.parseDate(DateFormatUtils.format(new Date(), "yyyy-MM-dd"), "yyyy-MM-dd");
|
||||
|
||||
// 按年生成批次
|
||||
DataBatch one = new DataBatch();
|
||||
one.setFirstDbNum(0);
|
||||
one.setFirstSfNum(0);
|
||||
one.setDbNum(0);
|
||||
one.setSfNum(0);
|
||||
one.setName(api);
|
||||
one.setLabel(dsr.getLabel());
|
||||
|
||||
// 按年份划分开始时间和结束时间创建批次
|
||||
Calendar startCalendar = Calendar.getInstance();
|
||||
startCalendar.setTime(startDate);
|
||||
int startYear = startCalendar.get(Calendar.YEAR);
|
||||
|
||||
Calendar endCalendar = Calendar.getInstance();
|
||||
endCalendar.setTime(endCreateDate);
|
||||
int endYear = endCalendar.get(Calendar.YEAR);
|
||||
|
||||
// 优化批次创建逻辑，确保正确的日期范围
|
||||
for (int year = startYear; year <= endYear; year++) {
|
||||
Date batchStartDate;
|
||||
Date batchEndDate;
|
||||
|
||||
// 设置批次开始时间为当年1月1日 00:00:00
|
||||
batchStartDate = DateUtils.parseDate(year + "-01-01 00:00:00", "yyyy-MM-dd HH:mm:ss");
|
||||
|
||||
if (year == endYear) {
|
||||
// 结束年份使用实际结束日期
|
||||
batchEndDate = endCreateDate;
|
||||
} else {
|
||||
int nextYear = year + 1;
|
||||
// 其他年份结束时间为当年12月31日 00:00:00
|
||||
batchEndDate = DateUtils.parseDate(nextYear + "-01-01 00:00:00", "yyyy-MM-dd HH:mm:ss");
|
||||
}
|
||||
|
||||
// 确保批次开始时间不早于实际开始时间
|
||||
if (batchStartDate.before(startDate)) {
|
||||
batchStartDate = startDate;
|
||||
}
|
||||
|
||||
DataBatch dataBatch = one.clone();
|
||||
dataBatch.setSyncStartDate(batchStartDate);
|
||||
dataBatch.setSyncEndDate(batchEndDate);
|
||||
dataBatchService.save(dataBatch);
|
||||
}
|
||||
|
||||
log.info("BigObject API创建完成: {}", api);
|
||||
return ReturnT.SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
@ -100,7 +100,7 @@ public class DataFieldServiceImpl extends ServiceImpl<DataFieldMapper, DataField
|
||||
} else if (fieldNames.contains("LastModifiedDate")) {
|
||||
return "LastModifiedDate";
|
||||
} else {
|
||||
return "";
|
||||
return "CreatedDate";
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1301,6 +1301,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
|
||||
linkMap.put("key", "new_id");
|
||||
linkMap.put("value", saveResults[j].getId());
|
||||
dList.add(linkMap);
|
||||
log.info("ContentDocumentLink Id: {}，对应的new_id: {} 更新成功！ " , ids[j], saveResults[j].getId());
|
||||
customMapper.updateById("ContentDocumentLink", dList, ids[j]);
|
||||
}
|
||||
}
|
||||
|
||||
@ -379,6 +379,7 @@ public class DataImportServiceImpl implements DataImportService {
|
||||
|
||||
List<Map<String, Object>> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1");
|
||||
if (referenceMap.isEmpty()){
|
||||
log.info("引用对象：" + reference + "不存在数据！请检查这个对象详情！！！");
|
||||
QueryWrapper<DataObject> maxIndex = new QueryWrapper<>();
|
||||
maxIndex.select("IFNULL(max(data_index),0) as data_index");
|
||||
maxIndex.ne("name", api);
|
||||
@ -386,7 +387,6 @@ public class DataImportServiceImpl implements DataImportService {
|
||||
//如果必填lookup字段没有值，跳过
|
||||
dataObject.setDataIndex(Integer.parseInt(map.get("data_index").toString()) + 1);
|
||||
dataObjectService.updateById(dataObject);
|
||||
log.info("引用对象：" + reference + "不存在数据！请检查这个对象详情！！！");
|
||||
return;
|
||||
}else{
|
||||
account.setField(dataField.getField(), referenceMap.get(0).get("new_id"));
|
||||
|
||||
@ -171,11 +171,13 @@ public class BulkUtil {
|
||||
|
||||
/**
|
||||
* 创建Bulk API 2.0查询作业
|
||||
** @param soqlQuery SOQL查询语句
|
||||
* @param url Salesforce实例URL
|
||||
* @param soqlQuery SOQL查询语句
|
||||
* @param connection BulkConnection连接对象
|
||||
* @return 作业ID
|
||||
* @throws Exception 创建作业时发生错误
|
||||
*/
|
||||
public static String createBulkV2Job(String url,String soqlQuery,BulkConnection connection) throws Exception {
|
||||
public static String createBulkV2QueryJob(String url, String soqlQuery, BulkConnection connection) throws Exception {
|
||||
// 构建创建查询作业的 JSON 请求体
|
||||
String jsonRequest = String.format("{\"operation\":\"query\", \"query\":\"%s\"}", soqlQuery.replace("\"", "\\\""));
|
||||
|
||||
@ -192,9 +194,7 @@ public class BulkUtil {
|
||||
// 从响应体中解析出 jobId
|
||||
String responseBody = EntityUtils.toString(response.getEntity());
|
||||
// 简单的JSON解析提取id，实际项目中建议使用JSON库如Jackson
|
||||
String jobId = responseBody.split("\"id\":\"")[1].split("\"")[0];
|
||||
System.out.println("Bulk API 2.0 Query job created with ID: " + jobId);
|
||||
return jobId;
|
||||
return responseBody.split("\"id\":\"")[1].split("\"")[0];
|
||||
} else {
|
||||
String errorBody = EntityUtils.toString(response.getEntity());
|
||||
throw new RuntimeException("Failed to create Bulk API 2.0 query job, status code: " + statusCode + ", error: " + errorBody);
|
||||
@ -206,11 +206,11 @@ public class BulkUtil {
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait for a Bulk API 2.0 job to complete by polling the job status.
|
||||
* 通过轮询作业状态等待 Bulk API 2.0 作业完成。
|
||||
*
|
||||
* @param jobId The ID of the job to check.
|
||||
* @param connection BulkConnection used to get session info.
|
||||
* @throws Exception If an error occurs during the polling process.
|
||||
* @param jobId 要检查的作业ID。
|
||||
* @param connection 用于获取会话信息的BulkConnection。
|
||||
* @throws Exception 轮询过程中发生错误时抛出异常。
|
||||
*/
|
||||
public static void waitForBulkV2JobCompletion(String url, String jobId, BulkConnection connection) throws Exception {
|
||||
CloseableHttpClient httpClient = HttpClients.createDefault();
|
||||
@ -228,17 +228,17 @@ public class BulkUtil {
|
||||
String responseBody = EntityUtils.toString(response.getEntity());
|
||||
// 简单解析响应体中的状态字段，实际项目中建议使用JSON库如Jackson
|
||||
if (responseBody.contains("\"state\":\"JobComplete\"")) {
|
||||
System.out.println("Job completed successfully.");
|
||||
System.out.println("作业成功完成。");
|
||||
jobCompleted = true;
|
||||
} else if (responseBody.contains("\"state\":\"Failed\"") || responseBody.contains("\"state\":\"Aborted\"")) {
|
||||
throw new RuntimeException("Job failed or was aborted. Response: " + responseBody);
|
||||
throw new RuntimeException("作业失败或已被中止。响应内容: " + responseBody);
|
||||
} else {
|
||||
System.out.println("Job is still running. Waiting...");
|
||||
System.out.println("作业仍在运行中。等待中...");
|
||||
Thread.sleep(10000); // 等待10秒后再次检查
|
||||
}
|
||||
} else {
|
||||
String errorBody = EntityUtils.toString(response.getEntity());
|
||||
throw new RuntimeException("Failed to get job status, status code: " + statusCode + ", error: " + errorBody);
|
||||
throw new RuntimeException("获取作业状态失败，状态码: " + statusCode + "，错误: " + errorBody);
|
||||
}
|
||||
} finally {
|
||||
response.close();
|
||||
@ -256,10 +256,10 @@ public class BulkUtil {
|
||||
* @return 查询结果的字符串内容
|
||||
* @throws Exception 如果在获取结果过程中发生错误
|
||||
*/
|
||||
public static String getBulkV2QueryJobResults(String url, String jobId, BulkConnection connection) throws Exception {
|
||||
public static BulkV2QueryResult getBulkV2QueryJobResults(String url, String jobId, BulkConnection connection) throws Exception {
|
||||
CloseableHttpClient httpClient = HttpClients.createDefault();
|
||||
String resultsUrl = url + "/services/data/v65.0/jobs/query/" + jobId + "/results" ;
|
||||
HttpPost httpGet = new HttpPost(resultsUrl);
|
||||
String resultsUrl = url + "/services/data/v56.0/jobs/query/" + jobId + "/results";
|
||||
HttpGet httpGet = new HttpGet(resultsUrl);
|
||||
httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
|
||||
httpGet.setHeader("Content-Type", "application/json");
|
||||
httpGet.setHeader("Accept", "text/csv");
|
||||
@ -269,13 +269,12 @@ public class BulkUtil {
|
||||
try {
|
||||
int statusCode = response.getStatusLine().getStatusCode();
|
||||
if (statusCode == 200) {
|
||||
|
||||
Header locatorHeader = response.getFirstHeader("Sforce-Locator");
|
||||
Header numRecordsHeader = response.getFirstHeader("Sforce-NumberOfRecords");
|
||||
Header numRecordsHeader = response.getFirstHeader("Sforce-NumberOfRecords");
|
||||
String newLocator = (locatorHeader != null) ? locatorHeader.getValue() : null;
|
||||
int numberOfRecords = (numRecordsHeader != null) ? Integer.parseInt(numRecordsHeader.getValue()) : 0;
|
||||
|
||||
return EntityUtils.toString(response.getEntity());
|
||||
String csvContent = EntityUtils.toString(response.getEntity());
|
||||
return new BulkV2QueryResult(csvContent, newLocator, numberOfRecords);
|
||||
} else {
|
||||
String errorBody = EntityUtils.toString(response.getEntity());
|
||||
throw new RuntimeException("Failed to get query job results, status code: " + statusCode + ", error: " + errorBody);
|
||||
@ -286,6 +285,41 @@ public class BulkUtil {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建Bulk API 2.0插入作业
|
||||
* @param url Salesforce实例URL
|
||||
* @param objectName 对象名称
|
||||
* @param connection BulkConnection连接对象
|
||||
* @return 作业ID
|
||||
* @throws Exception 创建作业时发生错误
|
||||
*/
|
||||
public static String createBulkV2InsertJob(String url, String objectName, BulkConnection connection) throws Exception {
|
||||
// 构建创建插入作业的 JSON 请求体
|
||||
String jsonRequest = String.format("{\"operation\":\"insert\", \"object\":\"%s\", \"contentType\":\"CSV\"}", objectName);
|
||||
|
||||
CloseableHttpClient httpClient = HttpClients.createDefault();
|
||||
HttpPost httpPost = new HttpPost(url + "services/data/v56.0/jobs/ingest/");
|
||||
httpPost.setHeader("Content-Type", "application/json");
|
||||
httpPost.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId());
|
||||
httpPost.setEntity(new StringEntity(jsonRequest, "UTF-8"));
|
||||
|
||||
CloseableHttpResponse response = httpClient.execute(httpPost);
|
||||
try {
|
||||
int statusCode = response.getStatusLine().getStatusCode();
|
||||
if (statusCode == 200) {
|
||||
// 从响应体中解析出 jobId
|
||||
String responseBody = EntityUtils.toString(response.getEntity());
|
||||
// 简单的JSON解析提取id，实际项目中建议使用JSON库如Jackson
|
||||
return responseBody.split("\"id\":\"")[1].split("\"")[0];
|
||||
} else {
|
||||
String errorBody = EntityUtils.toString(response.getEntity());
|
||||
throw new RuntimeException("Failed to create Bulk API 2.0 insert job, status code: " + statusCode + ", error: " + errorBody);
|
||||
}
|
||||
} finally {
|
||||
response.close();
|
||||
httpClient.close();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
|
||||
@ -455,7 +455,7 @@ public class DataUtil {
|
||||
case "datetime":
|
||||
return date;
|
||||
default:
|
||||
return "初始"+ UUID.randomUUID();
|
||||
return "初始"+ UUID.randomUUID().toString().substring(0, 8);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user