[Spring Batch] Data 를 CSV 파일로 만들기

회사에서 Spring Batch 를 통해서 DB Column 들을 CSV 로 전환해서 뽑을 필요가 있었다.

일단 간단하게 DB 에서 값을 읽어와서 CSV 로 만드는 작업을 해보자.

 

@Entity
@Table(name = "user")
class User(
  name: String,
  email: String
) {
  @Id
  @GeneratedValue(strategy = GenerationType.IDENTITY)
  var id: Long? = null
  @Column(name = "name", length = 20)
  var name: String = name
  @Column(name = "email", length = 30)
  var email: String = email

  override fun toString(): String {
    return "User(id=$id, name='$name', email='$email')"
  }

}

위와 같이 일단 User 테이블을 만들어주자.

그리고 엑셀로 만들어야할 데이터를 대략적으로 추가해보자.

INSERT INTO user (name, email) values ('로치', 'dev0jsh@gmail.com');
INSERT INTO user (name, email) values ('두더지', 'doo@gmail.com');
INSERT INTO user (name, email) values ('푸', 'foo@gmail.com');
INSERT INTO user (name, email) values ('포우', 'pow@gmail.com');
INSERT INTO user (name, email) values ('태리', 'teri@gmail.com');
INSERT INTO user (name, email) values ('호돌', 'hodol@gmail.com');
INSERT INTO user (name, email) values ('도도', 'dodo@gmail.com');

이제 쉽게 사용할 JdbcItemReader 를 하나 만들자.

@Configuration
class JdbcCustomCursor(
  val dataSource: DataSource
) {

  fun <T> jdbcCustomItemReader(
    fetchSize: Int = 10,
    sql: String,
    clazz: Class<T>
  ): JdbcCursorItemReader<T> {
    return JdbcCursorItemReaderBuilder<T>()
      .fetchSize(fetchSize)
      .dataSource(dataSource)
      .rowMapper(BeanPropertyRowMapper<T>(clazz))
      .sql(sql)
      .name(JDBC_CURSOR_BEAN_NAME)
      .build()
  }

  companion object {
    const val JDBC_CURSOR_BEAN_NAME = "jdbcCursorItemReader"
  }

}

이제 DB 에 대략적으로 붙으면서 사용할 수 있는 ItemReader 를 생성했다.

위의 코드가 궁금하다면 개인적으로 ItemReader 관련해서 공부해보길 바란다.

이번 시간엔 실습 코드 위주이므로 기본적인 코드에 대한 내용은 별로 설명하진 않겠다.

 

데이터를 CSV 파일로 만들기 위해서는 구분자와 어떤 값들을 필드로 할것인지 정해주어야 한다.

따라서 FlatItemFileWriter 에는 LineAggregator 라는 인터페이스를 필수로 구현해야 한다.

/**
 * Interface used to create string representing object.
 * 
 * @author Dave Syer
 */
public interface LineAggregator<T> {
	
	/**
	 * Create a string from the value provided.
	 * 
	 * @param item values to be converted
	 * @return string
	 */
	String aggregate(T item);
}

기본적으로 많이 쓰는 구현체로는 DelimitedLineAggregator 가 있다.

public class DelimitedLineAggregator<T> extends ExtractorLineAggregator<T> {

	private String delimiter = ",";

	/**
	 * Public setter for the delimiter.
	 * @param delimiter the delimiter to set
	 */
	public void setDelimiter(String delimiter) {
		this.delimiter = delimiter;
	}

	@Override
	public String doAggregate(Object[] fields) {
		return StringUtils.arrayToDelimitedString(fields, this.delimiter);
	}

}

일단 기본적으로 ExtractorLineAggregator 라는 추상클래스를 상속받고 있는데 가장 중요한 것은 아래의 Extractor 이다.

public abstract class ExtractorLineAggregator<T> implements LineAggregator<T> {

	private FieldExtractor<T> fieldExtractor = new PassThroughFieldExtractor<>();

	/**
	 * Public setter for the field extractor responsible for splitting an input
	 * object up into an array of objects. Defaults to
	 * {@link PassThroughFieldExtractor}.
	 * 
	 * @param fieldExtractor The field extractor to set
	 */
	public void setFieldExtractor(FieldExtractor<T> fieldExtractor) {
		this.fieldExtractor = fieldExtractor;
	}

	/**
	 * Extract fields from the given item using the {@link FieldExtractor} and
	 * then aggregate them. Any null field returned by the extractor will be
	 * replaced by an empty String. Null items are not allowed.
	 * 
	 * @see org.springframework.batch.item.file.transform.LineAggregator#aggregate(java.lang.Object)
	 */
    @Override
	public String aggregate(T item) {
		Assert.notNull(item, "Item is required");
		Object[] fields = this.fieldExtractor.extract(item);

		//
		// Replace nulls with empty strings
		//
		Object[] args = new Object[fields.length];
		for (int i = 0; i < fields.length; i++) {
			if (fields[i] == null) {
				args[i] = "";
			}
			else {
				args[i] = fields[i];
			}
		}

		return this.doAggregate(args);
	}

	/**
	 * Aggregate provided fields into single String.
	 * 
	 * @param fields An array of the fields that must be aggregated
	 * @return aggregated string
	 */
	protected abstract String doAggregate(Object[] fields);
}

 

밑을 보면 Extractor 의 extract method 를 호출하여 fields 를 얻어내고, 해당 field 들을 aggregate 라는 메소드를 통해서 하나의 문자열로 만들어낸다. 우리는 DelimitedLineAggregator 를 쓸것이므로 예를 들어 '"아이디","이름","이메일" ' 의 하나의 문자열로 탄생될 것이다.

 

따라서 우리는 Extractor 를 구현하거나 기존의 다른 것들을 이용해야 하는데 

이번 실습때는 한번 User 의 extract method 를 직접 구현해보려고 한다.

위에서 설명했듯이 extract method 는 어떤 데이터를 필드로 할지 알려주면 된다.

 

class UserExtractor: FieldExtractor<User> {

  override fun extract(item: User): Array<Any> {
    return arrayOf(item.id.toString(), item.name, item.email)
  }

}

실습에서는 "아이디", "이름", "이메일" 순으로 진행하기 위해서 위와 같이 Extract method 를 구현했다.

이제 이렇게 했을때 잘 CSV 가 출력될텐데 우리는 CSV 의 header 이름을 정해줘야 할 필요가 있다.

따라서 Writer 에 HeaderCallback 을 추가해줘야 한다.

  @Bean
  fun userItemToCsvWriter(): FlatFileItemWriter<User> {
    return FlatFileItemWriterBuilder<User>()
      .resource(FileSystemResource("src/main/resources/$CSV_FILE_NAME"))
      .append(true)
      .lineAggregator(userDataCsvLineAggregator())
      .headerCallback {
        val headers = listOf("아이디", "이름", "이메일")
        it.write(headers.joinToString(","))
      }
      .name(USER_ITEM_TO_CSV_WRITER)
      .build()
  }

이렇게만 해준다면 전체적으로 데이터들이 CSV 로 잘 바뀌어서 출력 될 것이다.

 

아래는 전체 코드이다.

@Configuration
class JobConfiguration(
  val jobBuilderFactory: JobBuilderFactory,
  val stepBuilderFactory: StepBuilderFactory,
  val dataSource: DataSource
) {

  @Bean
  fun userItemToCSVJob(): Job {
    return jobBuilderFactory
      .get(USER_ITEM_TO_CSV_JOB)
      .start(userItemToCsvStep())
      .incrementer(RunIdIncrementer())
      .build()
  }

  @Bean
  fun userItemToCsvStep(): Step {
    return stepBuilderFactory
      .get(USER_ITEM_TO_CSV_STEP)
      .chunk<User, User>(chunkSize)
      .reader(JdbcCustomCursor(dataSource).jdbcCustomItemReader(
        fetchSize = chunkSize,
        sql = FIND_ALL_USERS_QUERY,
        clazz = User::class.java
      ))
      .writer(userItemToCsvWriter())
      .build()
  }

  @Bean
  fun userItemToCsvWriter(): FlatFileItemWriter<User> {
    return FlatFileItemWriterBuilder<User>()
      .resource(FileSystemResource("src/main/resources/$CSV_FILE_NAME"))
      .append(true)
      .lineAggregator(userDataCsvLineAggregator())
      .headerCallback {
        val headers = listOf("아이디", "이름", "이메일")
        it.write(headers.joinToString(","))
      }
      .name(USER_ITEM_TO_CSV_WRITER)
      .build()
  }

  private fun userDataCsvLineAggregator(): LineAggregator<User> {
    val aggregator = DelimitedLineAggregator<User>()
    aggregator.setDelimiter(",")
    aggregator.setFieldExtractor(UserExtractor())
    return aggregator
  }

  companion object {
    const val USER_ITEM_TO_CSV_JOB = "USER_ITEM_TO_CSV_JOB"
    const val USER_ITEM_TO_CSV_STEP = "USER_ITEM_TO_CSV_STEP"
    const val USER_ITEM_TO_CSV_WRITER = "USER_ITEM_TO_CSV_WRITER"
    const val chunkSize = 10
    const val FIND_ALL_USERS_QUERY = "SELECT id, name, email FROM user"
    const val CSV_FILE_NAME = "user_data.csv"
  }

https://github.com/tmdgusya/batch-csv-example

GitHub - tmdgusya/batch-csv-example