시작하기 전에...
여기 쓰인 예제는 연습삼아 만들어본 도로명주소 데이터 덤프 툴 중 일부입니다.전체 소스는 공개할 생각이 없으니 알아서 공부해서 만드십쇼.
여기에 XML 설정은 눈 씻고 봐도 찾아볼 수 없습니다. XML 설정은 구글신께 신탁을 부탁합시다.
2015-07-31에 추가:
아주 마아아아아안약에 사이트에 도로명주소+기초구역코드(새 우편번호) 적용을 하고 싶은데 나온 검색 결과가 이거라면, 이걸 시도하기 전에 (즉, 자체 디비구축이라는 삽질을 하기 전에) 다음카카오를 만나세요.
여기로 가시면 자바스크립트로 한방에!
사용법
일단 이거부터...@EnableBatchProcessing그리고...
- 일(Job)을 만든다.
- 단계(Step)를 만든다.
일단 일(Job)부터 만들자
@Bean
public Job job() {
return jobBuilderFactory.get("address-import")
.start(streetStep())
.next(partitionedStreetDetailStep())
.next(partitionedStreetExtraStep())
.next(partitionedLotStep()).build();
}- Job을 만든다.
- 시작할 단계를 설정한다.
- 다음 단계... 다음 단계... 다음 단계...
- 설정 끗!
Job은 이게 끗이다. 진짜로...
도로명주소는 넣는 순서가 중요해서 설정하지 않았는데, 여러 개의 단계를 동시에 돌릴 수도 있다.
그런데 자바 설정으로는 단계를 멀티쓰레드화 하는게 쉽지 않았다.
일할 단계(Step)도 만들자
각 단계는 다음 순서대로 이루어진다.- 읽는다.
- 가공한다.
- 쓴다.
실제로는 쓰기 작업이 있으면 가공 작업은 생략이 가능하고, 가공 작업이 있으면 쓰기 작업을 생략할 수 있다.
@Bean
public Step streetStep() {
return stepBuilderFactory.get("street-address")
.<AddressStreet, AddressStreet> chunk(getChunkSize())
.reader(streetReader())
.writer(streetWriter())
.taskExecutor(taskExecutor)
.throttleLimit(THROTTLE_LIMIT)
.build();
}chunk는 읽기/쓰기 단위를 설정한다. 100을 설정하면 읽고 쓰는걸 100개 단위로 하게 된다.taskExecutor, throttleLimit는 밑에서 설명을...읽기
여기서는 주어지는 대상이 파일이니 파일을 읽는 것으로...@Bean
public FlatFileItemReader<AddressStreet> streetReader() {
DefaultLineMapper<AddressStreet> lineMapper = new DefaultLineMapper<>();
lineMapper.setLineTokenizer(lineTokenizer);
lineMapper.setFieldSetMapper(new FieldSetMapper<AddressStreet>() {
@Override
public AddressStreet mapFieldSet(FieldSet fieldSet) throws BindException {
AddressStreet street = new AddressStreet();
street.setId(String.format("%12s%2s", fieldSet.readString(0), fieldSet.readString(3)));
street.setCode(fieldSet.readString(0));
street.setCodeIndex(fieldSet.readString(3));
street.setName(fieldSet.readString(1));
street.setRegion(fieldSet.readString(4));
street.setCity(fieldSet.readString(6));
street.setTown(fieldSet.readString(8));
street.setTownType(fieldSet.readInt(10, 2));
street.setTownCode(fieldSet.readString(11));
street.setDisabled(fieldSet.readBoolean(12, "1"));
return street;
}
});
FlatFileItemReader<AddressStreet> reader = new FlatFileItemReader<>();
reader.setLineMapper(lineMapper);
reader.setResource(new PathResource(environment.getRequiredProperty("street")));
return reader;
}LineMapper를 만들고FileItemReader를 맹근 다음에FileItemReader에LineMapper와 읽을 파일을 설정
파일을 여러 개 읽으려면
FlatFileItemReader대신 MultiResourceItemReader를 사용하면 된다.쓰기
파일을 읽었으면 디비에 쏴줘야지...@Bean
public ItemWriter<AddressStreet> streetWriter() {
JdbcBatchItemWriter<AddressStreet> writer = new JdbcBatchItemWriter<>();
writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<AddressStreet>());
writer.setSql("INSERT INTO ADDRESS_STREET (ID, CODE, CODE_INDEX, NAME, REGION, CITY, TOWN, TOWN_CODE, TOWN_TYPE, IS_DISABLED) VALUES (:id, :code, :codeIndex, :name, :region, :city, :town, :townCode, :townType, :disabled)");
writer.setJdbcTemplate(jdbcTemplate);
return writer;
}만약에 본인처럼 Bean을 쓰는게 아니라면
BeanPropertyItemSqlParameterSourceProvider는 다른걸로 대체하거나 직접 맹글어야 한다?
멀티쓰레딩 - TaskExecutor와 ThrottleLimit
위에서 대충 넘겼던거 여기와서 설명을...앞뒤 다 자르고 설명하면... 작업 단위를 멀티쓰레드로 실행한다.
TaskExecutor는 다들 알다시피(?) 쓰레드 풀, ThrottleLimit는 동시실행 쓰레드 갯수 제한이다.
TaskExecutor, 뭘 골라야 쓰겄소?
TaskExecutor의 구현이 여럿 있는데, 일반적으로 원샷 실행일 경우 SimpleAsyncTaskExecutor를 많이 쓰는것 같다.왜인지 모르게
SimpleThreadPoolTaskExecutor와 ThreadPoolTaskExecutor가 의도한 대로 동작하지 않는데 (싱글 쓰레드처럼 동작한다.), 그냥 Executor를 래핑하는 ConcurrentTaskExecutor를 만들어 쓰는게 더 낫다.
ConcurrentTaskExecutor에 Executor를 넣지 않으면 Executors.newSingleThreadExecutor()를 설정한다. 파티셔닝 - 여러 개의 I/O를 멀티쓰레드로
위에 설명한게 I/O 단위 하나를 멀티쓰레드로 돌린다면, 이건 작업 하나에 여러 개의 I/O가 있을 때 이걸 멀티쓰레드로 돌리게 해준다.@Bean
public Step partitionedStreetDetailStep() {
return stepBuilderFactory.get("partition.street-address-detail")
.partitioner(streetDetailStep())
.partitioner("partition.street-address-detail.", streetDetailPartitioner())
.taskExecutor(taskExecutor)
.build();
}이 때, 내부 step은
Partitioner가 쪼개는 갯수만큼 생성된다.파일 여러개 읽기
파티셔닝 모드에서 파일 여러개 읽기는 손이 많이 가는 작업이다.- 읽을 파일 목록을
Partitioner로 맹근다. ItemReader를 맹근다.- 파티션 Step에서 사용하는 내부 Step에
ItemReader를 설정한다.
일단 파일 목록 파티션부터 맹근다.
@Bean
public Partitioner streetDetailPartitioner() {
MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
List<Resource> resources = Arrays.asList(environment.getProperty("street-detail", String[].class))
.stream()
.map(PathResource::new)
.collect(Collectors.toList());
partitioner.setResources(resources.toArray(new Resource[resources.size()]));
return partitioner;
}다음
ItemReader도 맹근 다음...@Bean
@StepScope
public FlatFileItemReader<AddressStreetDetail> streetDetailReader(@Value("#{stepExecutionContext['fileName']}") String path) {
DefaultLineMapper<AddressStreetDetail> lineMapper = new DefaultLineMapper<>();
lineMapper.setLineTokenizer(lineTokenizer);
lineMapper.setFieldSetMapper(new FieldSetMapper<AddressStreetDetail>() {
@Override
public AddressStreetDetail mapFieldSet(FieldSet fieldSet) throws BindException {
AddressStreetDetail detail = new AddressStreetDetail();
detail.setId(fieldSet.readString(0));
detail.setUnderground(fieldSet.readBoolean(3, "1"));
detail.setBuildingNo(fieldSet.readInt(4));
detail.setBuildingNoSub(fieldSet.readInt(5));
detail.setBlockNo(fieldSet.readString(6));
detail.setExtra(fieldSet.readBoolean(10, "1"));
AddressStreet street = new AddressStreet();
street.setId(String.format("%12s%2s", fieldSet.readString(1), fieldSet.readString(2)));
detail.setStreet(street);
return detail;
};
});
FlatFileItemReader<AddressStreetDetail> reader = new FlatFileItemReader<>();
reader.setLineMapper(lineMapper);
reader.setResource(new PathResource(URI.create(path)));
return reader;
}내부 Step도 맹근다.
@Bean
public Step streetDetailStep() {
return stepBuilderFactory.get("street-address-detail")
.<AddressStreetDetail, AddressStreetDetail> chunk(getChunkSize())
.reader(streetDetailReader("classpath:/empty.csv"))
.writer(streetDetailWriter())
.taskExecutor(taskExecutor)
.throttleLimit(THROTTLE_LIMIT)
.build();
}@StepScope라는 놈이 지금 막 등장했는데, 매우 중요하다. 이게 없으면 파티션된 작업에서 읽을 파일을 제대로 설정할 수가 없다! 그리고 읽을 파일명은 stepExecutionContext에서 받아오게 된다. (Job에는 같은 역할을 하는 jobExecutionContext가 있다.)아쉽게도 자바 설정으로
FlatFileItemReader를 생성할 때 읽을 파일을 뭐라도 설정하지 않으면 나중에 실제로 읽을 파일을 설정해 줘도 파일을 읽을 생각을 안한다. 따라서 bean을 만들 때는 아무거나 하나 던져준다. (이 때는 실제 파일이 없어도 신경을 쓰지 않지 시포요. 아니면 아무 내용도 없는 더미 파일을 하나 맹글어서 던져주든가...)삽질기 - 디비가 지저분해졌어요!
기본적으로 spring-boot는DataSource bean이 있고 @EnableBatchProcessing이 설정되어있다면 배치 실행 정보를 디비에 우겨넣는다. 원샷 실행이고 실행 정보를 저장해야 할 필요따원 없는데 이러면 참 골치가 아프더라.원래는
spring.batch.initializer.enabled 속성이 false이면 테이블을 생성하지 않고 잘 돌아야 하는데... 실상은 에러나 막 떨궈댄다.원인은
DefaultBatchConfigurer가 DataSource를 주입받기 때문에...그러니까
DataSource를 무시하도록 직접 BatchConfigurer를 만들면 된다.@Bean
public BatchConfigurer batchConfigurer() {
BatchConfigurer configurer = new BatchConfigurer() {
private PlatformTransactionManager transactionManager;
private JobRepository jobRepository;
private JobLauncher jobLauncher;
private JobExplorer jobExplorer;
@Override
public PlatformTransactionManager getTransactionManager() throws Exception {
return transactionManager;
}
@Override
public JobRepository getJobRepository() throws Exception {
return jobRepository;
}
@Override
public JobLauncher getJobLauncher() throws Exception {
return jobLauncher;
}
@Override
public JobExplorer getJobExplorer() throws Exception {
return jobExplorer;
}
@PostConstruct
public void initialize() {
if (this.transactionManager == null) {
this.transactionManager = new ResourcelessTransactionManager();
}
try {
MapJobRepositoryFactoryBean jrf = new MapJobRepositoryFactoryBean(this.transactionManager);
jrf.afterPropertiesSet();
this.jobRepository = jrf.getObject();
MapJobExplorerFactoryBean jef = new MapJobExplorerFactoryBean(jrf);
jef.afterPropertiesSet();
this.jobExplorer = jef.getObject();
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(jobRepository);
jobLauncher.afterPropertiesSet();
this.jobLauncher = jobLauncher;
} catch (Exception e) {
throw new BatchConfigurationException(e);
}
}
};
return configurer;
}