Compare commits
14 Commits
feat/dean/
...
ca7c5e7fa7
| Author | SHA1 | Date | |
|---|---|---|---|
| ca7c5e7fa7 | |||
| e92a56ff4f | |||
| d1a3c8cc2c | |||
| 5c47d111b1 | |||
| b940db8e73 | |||
| 4a3e299325 | |||
| c2d4d3a5f0 | |||
| ef31309f77 | |||
| e697867bb0 | |||
| b4fff05460 | |||
| 703c25aadf | |||
| be4840bb58 | |||
| b1765e9d1f | |||
| 0ff3f43a99 |
@@ -8,7 +8,7 @@ spring:
|
||||
active: prod # 사용할 프로파일 지정 (ex. dev, prod, test)
|
||||
|
||||
datasource:
|
||||
url: jdbc:postgresql://192.168.2.127:15432/kamco_cds
|
||||
url: jdbc:postgresql://172.16.4.56:15432/kamco_cds
|
||||
#url: jdbc:postgresql://localhost:5432/kamco_cds
|
||||
username: kamco_cds
|
||||
password: kamco_cds_Q!W@E#R$
|
||||
@@ -59,8 +59,8 @@ management:
|
||||
|
||||
file:
|
||||
#sync-root-dir: D:/kamco-nfs/images/
|
||||
sync-root-dir: /kamco-nfs/images/
|
||||
sync-tmp-dir: ${file.sync-root-dir}/tmp
|
||||
sync-root-dir: /data/images/
|
||||
sync-tmp-dir: /data/repo/tmp
|
||||
sync-file-extention: tfw,tif
|
||||
sync-auto-exception-start-year: 2025
|
||||
sync-auto-exception-before-year-cnt: 3
|
||||
|
||||
Binary file not shown.
@@ -1,12 +1,13 @@
|
||||
package com.kamco.cd.geojsonscheduler;
|
||||
|
||||
import com.kamco.cd.geojsonscheduler.config.DockerProperties;
|
||||
import com.kamco.cd.geojsonscheduler.config.TrainDockerProperties;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
|
||||
@SpringBootApplication
|
||||
@EnableConfigurationProperties(DockerProperties.class)
|
||||
@EnableConfigurationProperties({DockerProperties.class, TrainDockerProperties.class})
|
||||
public class GeoJsonSchedulerApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
|
||||
@@ -4,6 +4,7 @@ import com.kamco.cd.geojsonscheduler.service.DockerRunnerService;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.log4j.Log4j2;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.configuration.annotation.StepScope;
|
||||
import org.springframework.batch.core.scope.context.ChunkContext;
|
||||
import org.springframework.batch.core.step.tasklet.Tasklet;
|
||||
import org.springframework.batch.repeat.RepeatStatus;
|
||||
@@ -45,6 +46,7 @@ import org.springframework.stereotype.Component;
|
||||
*/
|
||||
@Log4j2
|
||||
@Component
|
||||
@StepScope
|
||||
@RequiredArgsConstructor
|
||||
public class DockerRunTasklet implements Tasklet {
|
||||
|
||||
|
||||
@@ -6,7 +6,9 @@ import java.util.List;
|
||||
import java.util.Objects;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.log4j.Log4j2;
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
import org.springframework.batch.core.JobParameters;
|
||||
import org.springframework.batch.core.JobParametersBuilder;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
@@ -16,35 +18,44 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
|
||||
import org.springframework.batch.repeat.RepeatStatus;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.support.DefaultTransactionDefinition;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
|
||||
/**
|
||||
* Child Job 실행 Tasklet (Parent Job용)
|
||||
*
|
||||
* <p>진행 중인 모든 분석 회차(AnalCntInfo)를 조회하여 각 회차마다 독립적인 Child Job을 실행합니다. 각 Child Job은 3개의
|
||||
* <p>
|
||||
* 진행 중인 모든 분석 회차(AnalCntInfo)를 조회하여 각 회차마다 독립적인 Child Job을 실행합니다. 각 Child Job은
|
||||
* 3개의
|
||||
* Step(makeGeoJson → dockerRun → zipResponse)을 순차적으로 실행합니다.
|
||||
*
|
||||
* <p><b>주요 기능:</b>
|
||||
* <p>
|
||||
* <b>주요 기능:</b>
|
||||
*
|
||||
* <ul>
|
||||
* <li>진행 중인 분석 회차 목록 조회 (tb_map_sheet_anal_inference, anal_state='ING')
|
||||
* <li>각 회차별 처리 필요 여부 판단 (all_cnt != file_cnt)
|
||||
* <li>회차마다 Child Job(processAnalCntInfoJob) 실행
|
||||
* <li>부분 실패 허용 (한 회차 실패해도 다른 회차 계속 처리)
|
||||
* <li>진행 중인 분석 회차 목록 조회 (tb_map_sheet_anal_inference, anal_state='ING')
|
||||
* <li>각 회차별 처리 필요 여부 판단 (all_cnt != file_cnt)
|
||||
* <li>회차마다 Child Job(processAnalCntInfoJob) 실행
|
||||
* <li>부분 실패 허용 (한 회차 실패해도 다른 회차 계속 처리)
|
||||
* </ul>
|
||||
*
|
||||
* <p><b>실행 조건:</b>
|
||||
* <p>
|
||||
* <b>실행 조건:</b>
|
||||
*
|
||||
* <ul>
|
||||
* <li>tb_map_sheet_anal_inference.anal_state = 'ING' (진행 중)
|
||||
* <li>검수 완료 건수(complete_cnt) > 0
|
||||
* <li>all_cnt != file_cnt (아직 파일 생성이 완료되지 않음)
|
||||
* <li>tb_map_sheet_anal_inference.anal_state = 'ING' (진행 중)
|
||||
* <li>검수 완료 건수(complete_cnt) > 0
|
||||
* <li>all_cnt != file_cnt (아직 파일 생성이 완료되지 않음)
|
||||
* </ul>
|
||||
*
|
||||
* <p><b>실패 정책:</b>
|
||||
* <p>
|
||||
* <b>실패 정책:</b>
|
||||
*
|
||||
* <ul>
|
||||
* <li>현재: 부분 실패 허용 (일부 Child Job 실패해도 Parent Job 성공)
|
||||
* <li>변경 가능: 87-89라인 주석 해제 시 하나라도 실패하면 Parent Job 실패
|
||||
* <li>현재: 부분 실패 허용 (일부 Child Job 실패해도 Parent Job 성공)
|
||||
* <li>변경 가능: 87-89라인 주석 해제 시 하나라도 실패하면 Parent Job 실패
|
||||
* </ul>
|
||||
*
|
||||
* @author KAMCO Development Team
|
||||
@@ -59,17 +70,23 @@ public class LaunchChildJobsTasklet implements Tasklet {
|
||||
/** 분석 회차 정보 조회를 위한 Repository */
|
||||
private final TrainingDataReviewJobRepository repository;
|
||||
|
||||
/** Child Job을 실행하기 위한 JobLauncher */
|
||||
private final JobLauncher jobLauncher;
|
||||
/** Child Job을 실행하기 위한 비동기 JobLauncher (트랜잭션 충돌 방지) */
|
||||
@Qualifier("asyncJobLauncher")
|
||||
private final JobLauncher asyncJobLauncher;
|
||||
|
||||
/** 실행할 Child Job (processAnalCntInfoJob) */
|
||||
@Qualifier("processAnalCntInfoJob")
|
||||
private final Job processAnalCntInfoJob;
|
||||
|
||||
/** 트랜잭션 매니저 (트랜잭션 제어용) */
|
||||
private final PlatformTransactionManager transactionManager;
|
||||
|
||||
/**
|
||||
* Parent Job의 메인 로직 실행
|
||||
*
|
||||
* <p>진행 중인 모든 분석 회차를 조회하여 각 회차마다 Child Job을 실행합니다. 한 회차가 실패해도 다른 회차는 계속 처리되며, 최종적으로 통계를
|
||||
* <p>
|
||||
* 진행 중인 모든 분석 회차를 조회하여 각 회차마다 Child Job을 실행합니다. 한 회차가 실패해도 다른 회차는 계속 처리되며,
|
||||
* 최종적으로 통계를
|
||||
* 로깅합니다.
|
||||
*
|
||||
* @param contribution Step 실행 정보를 담는 객체
|
||||
@@ -130,29 +147,80 @@ public class LaunchChildJobsTasklet implements Tasklet {
|
||||
|
||||
try {
|
||||
// Child Job Parameters 생성
|
||||
JobParameters jobParameters =
|
||||
new JobParametersBuilder()
|
||||
.addLong("analUid", info.getAnalUid())
|
||||
.addString("resultUid", info.getResultUid())
|
||||
.addLong("timestamp", System.currentTimeMillis()) // JobInstance 고유성 보장
|
||||
.toJobParameters();
|
||||
JobParameters jobParameters = new JobParametersBuilder()
|
||||
.addLong("analUid", info.getAnalUid())
|
||||
.addString("resultUid", info.getResultUid())
|
||||
.addLong("timestamp", System.currentTimeMillis()) // JobInstance 고유성 보장
|
||||
.toJobParameters();
|
||||
|
||||
log.info("[Child Job 실행] processAnalCntInfoJob 시작...");
|
||||
log.info(" - JobParameters: analUid={}, resultUid={}", info.getAnalUid(),
|
||||
info.getResultUid());
|
||||
|
||||
// Child Job 실행 (동기 방식)
|
||||
// Child Job 실행 (비동기 방식 - 트랜잭션 충돌 방지)
|
||||
// asyncJobLauncher를 사용하여 별도 쓰레드에서 실행
|
||||
// 내부적으로 makeGeoJsonStep → dockerRunStep → zipResponseStep 순차 실행
|
||||
long startTime = System.currentTimeMillis();
|
||||
jobLauncher.run(processAnalCntInfoJob, jobParameters);
|
||||
|
||||
// 트랜잭션 일시 정지 후 실행 (Existing transaction detected 에러 방지)
|
||||
JobExecution jobExecution = new TransactionTemplate(transactionManager,
|
||||
new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_NOT_SUPPORTED))
|
||||
.execute(status -> {
|
||||
try {
|
||||
return asyncJobLauncher.run(processAnalCntInfoJob, jobParameters);
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
});
|
||||
|
||||
if (jobExecution == null) {
|
||||
throw new RuntimeException("JobExecution is null after launching child job");
|
||||
}
|
||||
|
||||
// Child Job 완료 대기 (비동기 실행이므로 완료를 폴링)
|
||||
log.info("[Child Job 대기] 실행 완료 대기 중... (JobExecutionId={})",
|
||||
jobExecution.getId());
|
||||
while (jobExecution.isRunning()) {
|
||||
try {
|
||||
Thread.sleep(1000); // 1초마다 상태 확인
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new RuntimeException("Child Job 대기 중 인터럽트 발생", e);
|
||||
}
|
||||
}
|
||||
|
||||
long duration = System.currentTimeMillis() - startTime;
|
||||
BatchStatus status = jobExecution.getStatus();
|
||||
|
||||
log.info("[Child Job 완료] ✓ 정상 종료");
|
||||
log.info(" - AnalUid: {}", info.getAnalUid());
|
||||
log.info(" - ResultUid: {}", info.getResultUid());
|
||||
log.info(" - 실행 시간: {} ms ({} 초)", duration, duration / 1000);
|
||||
if (status == BatchStatus.COMPLETED) {
|
||||
log.info("[Child Job 완료] ✓ 정상 종료");
|
||||
log.info(" - AnalUid: {}", info.getAnalUid());
|
||||
log.info(" - ResultUid: {}", info.getResultUid());
|
||||
log.info(" - 실행 시간: {} ms ({} 초)", duration, duration / 1000);
|
||||
log.info(" - 최종 상태: {}", status);
|
||||
processedCount++;
|
||||
} else {
|
||||
// Child Job 실패
|
||||
log.error("[Child Job 실패] ✗ 비정상 종료");
|
||||
log.error(" - AnalUid: {}", info.getAnalUid());
|
||||
log.error(" - ResultUid: {}", info.getResultUid());
|
||||
log.error(" - 실행 시간: {} ms ({} 초)", duration, duration / 1000);
|
||||
log.error(" - 최종 상태: {}", status);
|
||||
log.error(" - Exit 상태: {}", jobExecution.getExitStatus());
|
||||
|
||||
processedCount++;
|
||||
// 실패 예외 정보 로깅
|
||||
if (!jobExecution.getAllFailureExceptions().isEmpty()) {
|
||||
log.error(" - 실패 예외:");
|
||||
for (Throwable t : jobExecution.getAllFailureExceptions()) {
|
||||
log.error(" * {}: {}", t.getClass().getSimpleName(), t.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
failedCount++;
|
||||
// 실패해도 다음 회차 계속 처리
|
||||
log.info("[계속 진행] 다음 회차 처리를 계속합니다.");
|
||||
continue; // 다음 for 루프로
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
// Child Job 실행 실패 시 (Step 실패 또는 예외 발생)
|
||||
@@ -181,8 +249,7 @@ public class LaunchChildJobsTasklet implements Tasklet {
|
||||
|
||||
// 성공률 계산
|
||||
if (analList.size() > 0) {
|
||||
double successRate =
|
||||
(double) processedCount / (analList.size() - skippedCount) * 100;
|
||||
double successRate = (double) processedCount / (analList.size() - skippedCount) * 100;
|
||||
log.info(" - 성공률: {}% (건너뛴 회차 제외)", String.format("%.2f", successRate));
|
||||
}
|
||||
|
||||
@@ -198,8 +265,8 @@ public class LaunchChildJobsTasklet implements Tasklet {
|
||||
// 실패가 있어도 Parent Job은 성공으로 처리 (부분 성공 정책)
|
||||
// 만약 하나라도 실패하면 Parent Job도 실패로 처리하려면 아래 주석 해제
|
||||
// throw new RuntimeException(
|
||||
// String.format("%d 개의 Child Job 실행이 실패했습니다. (성공: %d, 실패: %d)",
|
||||
// failedCount, processedCount, failedCount));
|
||||
// String.format("%d 개의 Child Job 실행이 실패했습니다. (성공: %d, 실패: %d)",
|
||||
// failedCount, processedCount, failedCount));
|
||||
} else {
|
||||
log.info("[완료] 모든 Child Job이 정상적으로 완료되었습니다.");
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ import java.util.List;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.log4j.Log4j2;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.configuration.annotation.StepScope;
|
||||
import org.springframework.batch.core.scope.context.ChunkContext;
|
||||
import org.springframework.batch.core.step.tasklet.Tasklet;
|
||||
import org.springframework.batch.repeat.RepeatStatus;
|
||||
@@ -46,6 +47,7 @@ import org.springframework.stereotype.Component;
|
||||
*/
|
||||
@Log4j2
|
||||
@Component
|
||||
@StepScope
|
||||
@RequiredArgsConstructor
|
||||
public class MakeGeoJsonTasklet implements Tasklet {
|
||||
|
||||
@@ -140,7 +142,7 @@ public class MakeGeoJsonTasklet implements Tasklet {
|
||||
|
||||
try {
|
||||
// Step 2-6: 디렉토리 생성 (존재하지 않으면)
|
||||
log.info(" [2-6] 디렉토리 생성 중...");
|
||||
log.info(" [2-6] 디렉토리 생성 중... {}", outputPath.getParent());
|
||||
Files.createDirectories(outputPath.getParent());
|
||||
log.info(" [2-6] 디렉토리 생성 완료: {}", outputPath.getParent());
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ import java.util.zip.ZipOutputStream;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.log4j.Log4j2;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.configuration.annotation.StepScope;
|
||||
import org.springframework.batch.core.scope.context.ChunkContext;
|
||||
import org.springframework.batch.core.step.tasklet.Tasklet;
|
||||
import org.springframework.batch.repeat.RepeatStatus;
|
||||
@@ -53,6 +54,7 @@ import org.springframework.stereotype.Component;
|
||||
*/
|
||||
@Log4j2
|
||||
@Component
|
||||
@StepScope
|
||||
@RequiredArgsConstructor
|
||||
public class ZipResponseTasklet implements Tasklet {
|
||||
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
package com.kamco.cd.geojsonscheduler.config;
|
||||
|
||||
import org.springframework.batch.core.launch.JobLauncher;
|
||||
import org.springframework.batch.core.launch.support.TaskExecutorJobLauncher;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.task.SimpleAsyncTaskExecutor;
|
||||
|
||||
/**
|
||||
* 비동기 JobLauncher 설정
|
||||
*
|
||||
* <p>Parent Job 내에서 Child Job을 실행할 때 트랜잭션 충돌을 방지하기 위해 비동기 JobLauncher를 생성합니다.
|
||||
*
|
||||
* <p><b>문제:</b> Parent Job의 Step이 트랜잭션 내에서 실행되는데, 그 안에서 동기 JobLauncher로 Child Job을
|
||||
* 실행하면 "Existing transaction detected in JobRepository" 에러 발생
|
||||
*
|
||||
* <p><b>해결:</b> 비동기 TaskExecutor를 사용하는 별도의 JobLauncher를 생성하여 트랜잭션 분리
|
||||
*
|
||||
* @author KAMCO Development Team
|
||||
* @since 1.0.0
|
||||
*/
|
||||
@Configuration
|
||||
public class AsyncJobLauncherConfig {
|
||||
|
||||
/**
|
||||
* 비동기 JobLauncher 생성
|
||||
*
|
||||
* <p>SimpleAsyncTaskExecutor를 사용하여 Child Job을 별도 쓰레드에서 실행합니다. 이렇게 하면 Parent Job의
|
||||
* 트랜잭션과 분리되어 트랜잭션 충돌이 발생하지 않습니다.
|
||||
*
|
||||
* @param jobRepository JobRepository
|
||||
* @return 비동기 JobLauncher
|
||||
* @throws Exception JobLauncher 초기화 실패 시
|
||||
*/
|
||||
@Bean(name = "asyncJobLauncher")
|
||||
public JobLauncher asyncJobLauncher(JobRepository jobRepository) throws Exception {
|
||||
TaskExecutorJobLauncher jobLauncher = new TaskExecutorJobLauncher();
|
||||
jobLauncher.setJobRepository(jobRepository);
|
||||
jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor()); // 비동기 실행
|
||||
jobLauncher.afterPropertiesSet();
|
||||
return jobLauncher;
|
||||
}
|
||||
}
|
||||
@@ -144,7 +144,7 @@ public class DockerRunnerService {
|
||||
* @param resultUid 결과물 고유 ID (입력/출력 폴더 경로에 사용)
|
||||
* @return Docker 명령어 문자열 리스트 (ProcessBuilder 실행용)
|
||||
*/
|
||||
private List<String> cmd buildCommand(String resultUid) {
|
||||
private List<String> buildCommand(String resultUid) {
|
||||
log.debug("Docker 명령어 파라미터 구성 중...");
|
||||
|
||||
List<String> cmd = new ArrayList<>();
|
||||
|
||||
@@ -6,6 +6,9 @@ spring:
|
||||
hikari:
|
||||
minimum-idle: 2
|
||||
maximum-pool-size: 5
|
||||
batch:
|
||||
job:
|
||||
name: exportGeoJsonJob # 기본 실행 Job 지정
|
||||
|
||||
training-data:
|
||||
geojson-dir: /kamco-nfs/dataset
|
||||
|
||||
@@ -3,6 +3,9 @@ spring:
|
||||
url: jdbc:postgresql://localhost:5432/kamco_cds
|
||||
username: kamco_cds
|
||||
password: kamco_cds
|
||||
batch:
|
||||
job:
|
||||
name: exportGeoJsonJob # 기본 실행 Job 지정
|
||||
|
||||
training-data:
|
||||
geojson-dir: /tmp/geojson
|
||||
|
||||
@@ -6,6 +6,26 @@ spring:
|
||||
hikari:
|
||||
minimum-idle: 2
|
||||
maximum-pool-size: 5
|
||||
batch:
|
||||
job:
|
||||
name: exportGeoJsonJob # 기본 실행 Job 지정
|
||||
|
||||
training-data:
|
||||
geojson-dir: /kamco-nfs/dataset
|
||||
|
||||
# Train Model Docker Configuration
|
||||
train-data:
|
||||
docker:
|
||||
image: kamco-cd-train:latest
|
||||
data-volume: /kamco-nfs/dataset:/data
|
||||
checkpoints-volume: /kamco-nfs/checkpoints:/checkpoints
|
||||
dataset-folder: /data/dataset
|
||||
output-folder: /data/output
|
||||
input-size: "512"
|
||||
crop-size: "256"
|
||||
batch-size: 8
|
||||
gpu-ids: "0,1,2,3"
|
||||
gpus: 4
|
||||
lr: "0.001"
|
||||
backbone: resnet50
|
||||
epochs: 100
|
||||
|
||||
@@ -21,6 +21,7 @@ spring:
|
||||
batch:
|
||||
job:
|
||||
enabled: true
|
||||
name: exportGeoJsonJob # 기본 실행 Job 지정
|
||||
jdbc:
|
||||
initialize-schema: always
|
||||
|
||||
|
||||
Reference in New Issue
Block a user