스프링 배치(Spring Batch)를 이용한 데이터 마이그레이션

회사 업무 중에 데이터 마이그레이션을 해야했는데 약 9만건의 데이터와 13만건의 데이터를 옮기는 작업을 해야했다.

Connection는 총 3개고 A Database에서 각각 B, C Database로 마이그레이션을 하는 상황이다.

업무를 하면 작성한 코드를 기준으로 예제를 정리해보겠다.

 

해당 예제는 Spring Boot, Gradle을 기준으로 하겠다.

Spring Batch의 개념 및 설정하는 방법은 얼마전에 작성한 블로그 글을 참고 바란다.

 

1. application.yml

Docker를 통해 mysql를 생성하고 hikariCP와 JPA를 사용해 마이그레이션 하려고 한다.

spring:
  datasource:
    hikari:
      a-database:
        jdbc-url: jdbc:mysql://localhost:3306/spring_batch
        username: root
        password: 1q2w3e!
        driver-class-name: com.mysql.jdbc.Driver
      b-database:
        jdbc-url: jdbc:mysql://localhost:12020/spring_batch
        username: root
        password: 1q2w3e!
        driver-class-name: com.mysql.jdbc.Driver
      c-database:
        jdbc-url: jdbc:mysql://localhost:12021/spring_batch
        username: root
        password: 1q2w3e!
        driver-class-name: com.mysql.jdbc.Driver

  jpa:
    database: mysql
    database-platform: org.hibernate.dialect.MySQL5InnoDBDialect
    generate-ddl: false
    open-in-view: false
    hibernate:
      ddl-auto: validate
    show-sql: true
    properties:
      hibernate:
        enable_lazy_load_no_trans: true
        format_sql: true

 

2. 엔티티 생성

마이그레이션 할 엔티티를 만들어 준다.

폴더 구조

// AUser 엔티티
@Getter
@Entity
@Table(name = "users")
@AllArgsConstructor
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class AUser {

    @Id
    @Column(name = "id")
    private Long id;

    @Column(name = "email", nullable = false)
    private String email;

    @Column(name = "name", nullable = false)
    private String name;

}

// AUserRepository
public interface AUserRepository extends JpaRepository<AUser, Long> {
}

BUser, CUser도 마찬가지로 만들어준다.

 

3. DataSource, EntityManagerFactory Bean 등록

각각 DataSource와 EntityManagerFactory를 Bean으로 등록해준다.

@Configuration
@EnableConfigurationProperties
public class DataSourceProperties {

    @Primary
    @Bean(name = "ADataSource")
    @ConfigurationProperties(prefix = "spring.datasource.hikari.a-database")
    public DataSource ADataSource() {
        return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }

    @Bean(name = "BDataSource")
    @ConfigurationProperties(prefix = "spring.datasource.hikari.b-database")
    public DataSource BDataSource() {
        return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }

    @Bean(name = "CDataSource")
    @ConfigurationProperties(prefix = "spring.datasource.hikari.c-database")
    public DataSource CDataSource() {
        return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }

}

@Configuration
@EnableTransactionManagement
@EnableJpaRepositories(
    entityManagerFactoryRef = "AEntityManagerFactory", // 등록한 Bean Id
    basePackages = { "com.devbeekei.springbatch.a.repository" } // 리포지토리 패키지
)
class AEntityManagerFactoryConfiguration {

    @Autowired
    @Qualifier("ADataSource")
    private DataSource ADataSource;

    @Primary
    @Bean(name = "AEntityManagerFactory")
    public LocalContainerEntityManagerFactoryBean AEntityManagerFactory(EntityManagerFactoryBuilder builder) {
        return builder
            .dataSource(ADataSource)
            .packages("com.devbeekei.springbatch.a.entity") // 엔티티 패키지
            .persistenceUnit("A")
            .build();
    }

}

마찬가지로 BEntityManagerFactoryConfiguration, CEntityManagerFactoryConfiguration도 구현해준다.

리포지토리 패키지와 엔티티 패키지는 2. 엔티티 생성에서 생성한 폴더 경로를 설정하면 된다.

 

4. Batch Job 실행

설명은 주석으로 작성했다.

@RequiredArgsConstructor
@Configuration
public class MigrationJobConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
	
    // Bean으로 등록해준 EntityManagerFactory들을 불러온다.
    @Autowired
    @Qualifier("AEntityManagerFactory")
    private EntityManagerFactory AEntityManagerFactory;

    @Autowired
    @Qualifier("BEntityManagerFactory")
    private EntityManagerFactory BEntityManagerFactory;

    @Autowired
    @Qualifier("CEntityManagerFactory")
    private EntityManagerFactory CEntityManagerFactory;

    private int chunkSize; // 한번에 조회할 데이터 수
    @Value("${chunkSize:1000}") // 미 설정 시 1000
    public void setChunkSize(int chunkSize) {
        this.chunkSize = chunkSize;
    }

    @Bean
    public Job migrationJob() { // Job 생성
        return jobBuilderFactory.get("migrationJob")
            .start(migrationBStep())
            .next(migrationCStep())
            .build();
    }


    @Bean
    public Step migrationBStep() {
        return stepBuilderFactory.get("migrationBStep")
            // JPA를 사용하기 때문에 transactionManager를 설정해줘야 transaction이 동작하게 된다.
            .transactionManager(new JpaTransactionManager(BEntityManagerFactory))
            // chunk 방식으로 Step을 구현하고 chunkSize를 설정한다.
            .<AUser, BUser>chunk(chunkSize) 
            .reader(reader()) // Reader
            .processor(bProcessor()) // Processor
            .writer(bWriter()) // writer
            .build();
    }

    @Bean
    public Step migrationCStep() {
        return stepBuilderFactory.get("migrationCStep")
            .transactionManager(new JpaTransactionManager(CEntityManagerFactory))
            .<AUser, CUser>chunk(chunkSize)
            .reader(reader())
            .processor(cProcessor())
            .writer(cWriter())
            .build();
    }

    @Bean
    public ItemReader<AUser> reader() {
    	// 데이터 조회
        return new JpaPagingItemReaderBuilder<AUser>()
            .name("reader")
            .entityManagerFactory(AEntityManagerFactory)
            .pageSize(chunkSize)
            .queryString("SELECT u FROM AUser u")
            .build();
    }

    @Bean
    public ItemProcessor<AUser, BUser> bProcessor() {
    	// 조회한 데이터를 Writer에 맞게 가공처리
        // 만약 Read하는 엔티티와 Write하는 엔티티가 같다면 Processor는 설정하지 않아도 된다.
        return aUser -> BUser.builder() 
            .id(aUser.getId())
            .email(aUser.getEmail())
            .name(aUser.getName())
            .build();
    }

    @Bean
    public JpaItemWriter<BUser> bWriter() {
        // JpaItemWriter는 JPA EntityManager를 이용해 Entity를 DB에 Insert해준다.
        JpaItemWriter<BUser> writer = new JpaItemWriter<>();
        // EntityManagerFactory 설정
        writer.setEntityManagerFactory(BEntityManagerFactory);
        // true로 설정 시 Insert만 실행하고, false로 설정 시 데이터가 존재하면 Update
        writer.setUsePersist(true);
        return writer;
    }

    @Bean
    public ItemProcessor<AUser, CUser> cProcessor() {
        return aUser -> CUser.builder() 
            .id(aUser.getId())
            .email(aUser.getEmail())
            .name(aUser.getName())
            .build();
    }

    @Bean
    public JpaItemWriter<CUser> cWriter() {
        JpaItemWriter<CUser> writer = new JpaItemWriter<>();
        writer.setEntityManagerFactory(CEntityManagerFactory);
        writer.setUsePersist(true);
        return writer;
    }

}

JpaItemWriter를 사용할 때 setUsePersist를 false로 설정하면 해당 엔티티를 한번 조회 후 존재하면 Update, 존재하지 않으면 Insert 한다. 당연히 조회 후 Update 또는 Insert하기 때문에 마이그레이션 시간은 더 오래 걸린다.

 

 

Job을 실행해보면 설정한 chunkSize만큼 Select 후 조회된 데이터를 Insert 시켜준다.

데이터 16만개 기준으로 테스트 해봤을때 3분정도가 걸렸었다. 물론 Join하는 테이블이 있거나 컬럼이 많을 수록 더 오래걸린다.

 

각각 데이터베이스를 확인해보면 정상적으로 데이터가 Insert 된 것을 확인할 수 있다.

이처럼 Spring Batch를 사용하면 간단한 코드로 데이터 마이그레이션이 가능하다.

 

 

---

※ 전체 코드는 GitHub에서 확인하실 수 있습니다.