Story Of ace-T

2018/10/12 - [OpenSource/Spring Batch] - Spring Boot + Spring Batch 분석_01

 저번 분석에서는 Job에 대해서 알아보았습니다. 

이제 Step에 대해 알아보도록 하겠습니다.

설명하고 있는 소스는 https://spring.io/guides/gs/batch-processing/ 를 참고!

// tag::jobstep[]
    @Bean
    public Job importUserJob(JobCompletionNotificationListener listener, Step step1) {
        return jobBuilderFactory.get("importUserJob")
            .incrementer(new RunIdIncrementer())
            .listener(listener)
            .flow(step1)
            .end()
            .build();
    }

    @Bean
    public Step step1(JdbcBatchItemWriter<Person> writer) {
        return stepBuilderFactory.get("step1")
            .<Person, Person> chunk(10)
            .reader(reader())
            .processor(processor())
            .writer(writer)
            .build();
    }
    // end::jobstep[]

• Step의 메소드명은 step1
• 파라미터는 JdbcBatchItemWriter<Person> writer 
• return은 stepBuilder
• builder에는 아래와 같이 구성.
• stepBuilderFactory.get
• chunk
• reader
• processor
• writer
• 내용을 파악하기 전에 SpringBatch의 Flow를 파악해보면 좋을것 같다.
  왜? 나는 chunk, reader, processor, writer를 알아야하는가?에 대한 답변이 될꺼라 생각한다.
• SpringBatch Flow

• 이제 하나하나씩 알아가보자.
  

• chunk를 보자.
• 우선 chunk(청크)란 배치선 덩어리 데이터들을 단위별로 묶어서 처리되어지는 수!
  즉, 위에 10이라는 뜻은 10개를 모아서 Chuck단위(10)로 트랜잭션을 이루고 있다고 보면 된다.
  트랜잭션 수행이기 때문에 실패 시 Chunk단위로 롤백이되고, 이전에 커밋된 트랜잭션까지는 반영이 된다는 뜻이다.
• <Person, Person> chunk(10)에서 <Person, Person>은 제네릭 타입으로
  각 청크의 input과 output type을 나타낸다. 
• ItemReader<Person>과 ItemWriter<Person>로 생각하면 된다.
• reader(reader())를 보자.
• reader? 뜻 그대로 읽어들이는 것이다. 무엇을? 데이터를~(txt, api, db 등등등)
  

   @Bean
      public FlatFileItemReader<Person> reader() {
          return new FlatFileItemReaderBuilder<Person>()
              .name("personItemReader")
              .resource(new ClassPathResource("sample-data.csv"))
              .delimited()
              .names(new String[]{"firstName", "lastName"})
              .fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                  setTargetType(Person.class);
              }})
              .build();
      }
  

• 위의 FlatFileItemReader는 SpringBatch에서 제공해주는 녀석이다.
  json, xml 형태가 아닌 말그대로 flat file을 읽어들인다.
  

  import org.springframework.batch.item.file.FlatFileItemReader;
  import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;

• FlatFileItemReaderBuilder! 이녀석 또한 Builder Pattern. 
• name : reader instance명 
• resource : 근원지 즉, read할 flat file.
• delimited() : 명시적이지 않을 때 사용!
  DelimitedLineTokenizer를 작성하기위한 FlatFileItemReaderBuilder.DelimitedBuilder의 인스턴스를 리턴합니다. 이 빌더에 의해 구성된 DelimitedLineTokenizer는 lineTokenizer를 통해 명시적으로 구성되지 않은 경우에만 사용됩니다.
• names : sample-data.csv의 내용과 names에서 선언한 포맷에 맞게 들어오게 되며 이 데이터는
  
• fieldSetMapper()를 통해 DTO에 매핑 되어진다.
• 예를 들어 아래의 데이터 중 Jane, Doe라고 한다면  String[]{"firstName", "lastName"}에 담긴다는 것이고 이것은 fieldSetMapper를 통해 객체에 담기게 된다. 
  

• 제공 되는 구현 클래스들
  AddressFieldSetMapper, AggregateItemFieldSetMapper, ArrayFieldSetMapper, 
  BeanWrapperFieldSetMapper, BillingFieldSetMapper, CustomerCreditFieldSetMapper,
  CustomerFieldSetMapper, CustomerUpdateFieldSetMapper, GameFieldSetMapper, 
  HeaderFieldSetMapper, OrderItemFieldSetMapper, PassThroughFieldSetMapper, 
  PlayerFieldSetMapper, ShippingFieldSetMapper, TradeFieldSetMapper
• 이중 우리는 BeanWrapperFieldSetMapper 을 사용.
• setTargetType : property가 아닌 Bean을 설정. 여기에서는 Person.class
  

  public class Person {
  
      private String lastName;
      private String firstName; 
      이하 생략..

• processor(processor)
• PersonItemProcessor는 ItemProcessor의 구현체이다.
• Person객체를 input으로 하여 가공한 뒤에 output으로 Person을 리턴한다. 
  

  @Bean
      public PersonItemProcessor processor() {
          return new PersonItemProcessor();
      }

  package hello;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.springframework.batch.item.ItemProcessor;
  
  public class PersonItemProcessor implements ItemProcessor<Person, Person> {
  
      private static final Logger log = LoggerFactory.getLogger(PersonItemProcessor.class);
  
      @Override
      public Person process(final Person person) throws Exception {
          final String firstName = person.getFirstName().toUpperCase();
          final String lastName = person.getLastName().toUpperCase();
  
          final Person transformedPerson = new Person(firstName, lastName);
  
          log.info("Converting (" + person + ") into (" + transformedPerson + ")");
  
          return transformedPerson;
      }
  
  }

• 정리하면 processor의 역할은 read한 것을 가공하여 write에 주는 것이다.

• writer(writer)
• 이제 마지막으로 writer입니다.
  여기에서는 데이터베이스 관련 writer중 JdbcBatchItemWriter가 사용되어집니다.

@Bean
    public JdbcBatchItemWriter<Person> writer(DataSource dataSource) {
        return new JdbcBatchItemWriterBuilder<Person>()
            .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
            .sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
            .dataSource(dataSource)
            .build();
    }

• JdbcBatchItemWriterBuilder가 사용.
• itemSqlParameterSourceProvider()
• 작성자가 사용할 ItemSqlParameterSourceProvider를 구성합니다.
  beanMapped ()가 호출되지 않은 경우에만 사용해야합니다.
• BeanPropertyItemSqlParmeterSourceProvider
• SQL구문에서 파라미터에 사용되어지는 명명 중 JavaBean(Person)에 속성에 해당되면 
  제공하여준다.
• sql : sql구문
• dataSource : @EnableBatchProcessing에 의해 생성.

• good link
• https://docs.spring.io/spring-batch/3.0.x/reference/html/listOfReadersAndWriters.html

주저리 

   spring boot를 사용하여 
   spring batch를 해보려는데 막상 spring batch를 잘모르면 제대로 못할것 같은 생각이 들었다.
   springframework를 spring boot기반으로 해볼때도 마찬가지였다.

구성

• Job과 Step으로 구성.
• Job
• 여러개의 Step으로 구성.
• Step은 ItemReader, ItemProcessor, ItemWriter의 구조이거나 
  이런 전형적인 구조가 아닐 경우 Tasklet 구조.

org.springframework.boot:spring-boot-starter-batch 디펜던시는 요녀석!

Batch Job을 구성할 때 
Spring Batch는 따로 작성할 필요없이 비즈니스 로직에 집중할 수 있도록 많은 유틸리티를 제공한다.

BatchConfiguration! 이부분은 옛날 Spring Batch에서 xml로 구성되던 녀석이다.

@Configuration
@EnableBatchProcessing
public class BatchConfiguration {

하나하나씩 보도록 하자. 우선 Factory를 DI 하는 것을 볼수 있다.

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

// tag::jobstep[]
    @Bean
    public Job importUserJob(JobCompletionNotificationListener listener, Step step1) {
        return jobBuilderFactory.get("importUserJob")
            .incrementer(new RunIdIncrementer())
            .listener(listener)
            .flow(step1)
            .end()
            .build();
    }

    @Bean
    public Step step1(JdbcBatchItemWriter<Person> writer) {
        return stepBuilderFactory.get("step1")
            .<Person, Person> chunk(10)
            .reader(reader())
            .processor(processor())
            .writer(writer)
            .build();
    }
    // end::jobstep[]

그리고 Step에서 사용되어지는 reader, processor, writer도 함께 구성.

    // tag::readerwriterprocessor[]
    @Bean
    public FlatFileItemReader<Person> reader() {
        return new FlatFileItemReaderBuilder<Person>()
            .name("personItemReader")
            .resource(new ClassPathResource("sample-data.csv"))
            .delimited()
            .names(new String[]{"firstName", "lastName"})
            .fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }})
            .build();
    }

    @Bean
    public PersonItemProcessor processor() {
        return new PersonItemProcessor();
    }

    @Bean
    public JdbcBatchItemWriter<Person> writer(DataSource dataSource) {
        return new JdbcBatchItemWriterBuilder<Person>()
            .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
            .sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
            .dataSource(dataSource)
            .build();
    }
    // end::readerwriterprocessor[]

구조는 이러하고..이제 조금 더 자세하게 보도록 하자.

More and More

BatchConfiguration.java

1. @Configuration : 설정 파일임을 알려준다. @MVC에서 @service @controller와 같다고 보면 된다.
2. @EnableBatchProcessing : Batch 관련 유용한 유틸을 제공가능케 한다. 영어 고대로 Enable하겠다는 뜻.
3. 위에서 언급한 내용처럼 Factory를 DI 하여 Job과 Step을 구성한다.
1. Job은 Step으로 구성하여 만들어지고 Step은 reader, processor, writer를 구성할 수 있다.

• 어노테이션 @Autowired로 DI 해주는 부분.

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

• Job과 Step을 설정 해주는 부분.

    @Bean
    public Job importUserJob(JobCompletionNotificationListener listener, Step step1) {
        return jobBuilderFactory.get("importUserJob")
            .incrementer(new RunIdIncrementer())
            .listener(listener)
            .flow(step1)
            .end()
            .build();
    }

    @Bean
    public Step step1(JdbcBatchItemWriter<Person> writer) {
        return stepBuilderFactory.get("step1")
            .<Person, Person> chunk(10)
            .reader(reader())
            .processor(processor())
            .writer(writer)
            .build();
    } 

• Job은 아래와 같이 importUserJob이라는 함수명을 가지며, 파라미터로는 listener, step을
• return으로는 JobBuilderFactory를 이용하여 Job을 리턴하는 구조이다.
• jobBuilderFactory.get은 job builder를 생성하고 JobRepository를 초기화 시킨다.
  

[ Spring Batch 흐름도]

• 아래를 좀 더 심화있게 이해하려면 builder pattern을 이해해야한다. 
• 하지만 우선적으로 Spring Batch에 대해서 파악하는게 우선이다. 
• 이제 돌아와서 incremeter를 보자.
• JobBuilderHelper Class에 속해있는 method이다. 
• JobBuilderHelper는 JobBuilder의 부모 클래스이다.
• 포맷은 아래와 같다.
• public B incrementer(JobParametersIncrementer jobParametersIncrementer)
• 파라미터로 JobPrametersIncrementer를 가지는데 이것은 Interface 형태이다.
• 이 Interface를 구현한 녀석이 위에 사용되어진 RunIdIncrementer Class 이다.
• 그리고 incrementer 이녀석이 하는 역할은 JobParametersIncrementer 인터페이스의 
  역할을 보면 알수 있다. JobParameters를 취득하기 위한 인터페이스이기 때문이다.
• JobParameters..! 이녀석을 이해하려면 아래의 그림을 이해해야한다.
  
• 동시 다발적으로 돌아가는 Job들의 차이는 어떻게 알수가 있나? Job들 간의 구별? 그 해답은 JobParameters에 있다. 그래서 batch job을 시작할 때 이처럼 JobParameters를 취득한다.
  더욱 더 자세한 내용은 해당 링크를 참조. 

• 그 다음으로 listener.
• JobBuilderHelper Class에 속해 있는 Method이며, format은 아래와 같다.
  public B listener(JobExecutionListener listener)
• JobExecutionListener도 Interface이며 지금 설명하는 소스에서는 
  JobCompletionNotificationListener Class에서 extends하고 있다.
• JobCompletionNotificationListener.java
  

  package hello;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.batch.core.BatchStatus;
  import org.springframework.batch.core.JobExecution;
  import org.springframework.batch.core.listener.JobExecutionListenerSupport;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.jdbc.core.JdbcTemplate;
  import org.springframework.stereotype.Component;
  
  @Component
  public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
  
  	private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class);
  
  	private final JdbcTemplate jdbcTemplate;
  
  	@Autowired
  	public JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) {
  		this.jdbcTemplate = jdbcTemplate;
  	}
  
  	@Override
  	public void afterJob(JobExecution jobExecution) {
  		if(jobExecution.getStatus() == BatchStatus.COMPLETED) {
  			log.info("!!! JOB FINISHED! Time to verify the results");
  
  			jdbcTemplate.query("SELECT first_name, last_name FROM people",
  				(rs, row) -> new Person(
  					rs.getString(1),
  					rs.getString(2))
  			).forEach(person -> log.info("Found <" + person + "> in the database."));
  		}
  	}
  }

• 위의 소스는 job이 BatchStatus.COMPLETED인가에 대해서 listen하고 있다. 그리고나서 JdbcTemplate를 사용한다.

• 다음으로 flow.
• JobBuilder Class에 속해 있는 Method이며, format은 아래와 같다.
  public JobFlowBuilder flow(Step step)
  딱 봐도 job과 step의 연결 & 수행. return으로는 a SimpleJobBuilder.
• 다음으로 end. 즉, builder를 종료.
• 다음으로 build() : 호출하여 최종적으로 a job builder를 리턴. 
• 이것은 bulider pattern을 참고.(잘 정리 해놓은 블로그! thx.)

지금까지 Job 관련 내용들을 살펴 보았습니다.

다음 포스팅에서는 비즈니스 로직을 당담하고 있는 Step에 대해서 알아보도록 하겠습니다.

2018/10/23 - [OpenSource/Spring Batch] - Spring Boot + Spring Batch 분석_02

스프링배치 버전이 낮아서 이참에 스프링부트+스프링배치로 변경해보려고 한다.

"라스트 제다이"가 아닌 "라스트 스프링" ㅋㅋ 

설명에 앞서 주저리를 하자면..요즘 스프링 관련 업무가 없어서..너무 뒤쳐진거 같다..ㅜ.ㅜ..마지막 남은 스프링인가? ㅋㅋ 

예제는 아래 사이트를 참고
  - https://spring.io/guides/gs/batch-processing/ 

결과는 아래와 같다.

위 사이트 내용을 요약해서 보면

간단한 배치 서비스를 만들껀데
이 서비스는 csv파일의 데이터를 가져와서 입맛에 맞게 변경 한 뒤 데이터베이스에 저장하는게 목표이다.

정형적인 배치서비스를 축약한 내용의 소스인것 같아서 선택했다.

데이터를 수집해서 변경 후에 데이터베이스에 저장(insert) or 변경(update)하는 style이 되겠다.

위 소스를 위해서 필요한것은

• IDE(Spring Tool Suite (STS)  or  IntelliJ IDEA  )
• JDK 1.8 or later
• Gradle 4+ or Maven 3.2+
• 마지막으로 15분이 필요하다 ㅋㅋ 

kafka install

brew로 인스톨 해준다.
brew install kafka

명령어를 날리려는데 셋팅 된 위치를 모르겠다..ㅋㅋ

위치는 요기!!
/usr/local/Cellar/kafka/0.11.0.0/bin

명령은!

./kafka-console-consumer --zookeeper xxx.xxx.com:1111/acet-kafka --topic ace-taeha

끝~

로컬이 아닌 서버에 kibana를 다운 받고 run을 한 뒤!

웹브라우저로 접근을 하려는데..접근이 fail...

원인은 server.host가 주석처리가 되어있었는데 default가 localhost라서..접근이 안되었던 것!!!

docker image와는 다른 default~!!

엄청나게 쉽게 될 줄 알았는데 삽질 함..-_-;;

다른분들은 금방 해결 되길 바라며..포스팅~촤촤촤! ㅎㅎ

주저리 > 

보통 Web Project를할 때 MVC 패턴과 더불어 SQL Mapper인 ibatis, mybatis등을 많이 사용하였습니다.

지금 회사에서도 mybatis가 사용된 프로젝트들이 대다수 입니다.

필드 하나가 추가 되면 dto(vo)를 비롯한 sql등을 수정해주어야하는..

 즉, MVC pattern : View - Controller - Service - Repository(Dao) - SQL까지 전체적인 수정을 요합니다.

이런 노가다적이고 복잡한 것을 JPA에서는 객체모델링을 기반으로 풀어나간다고 합니다.

db 모델링 기반이 아닌 object 모델링!! 

정말 가능한지 한번 알아보겠습니다.

JPA는 무엇인가?

풀어서 알아보면! Java Persistence Api~~자바 진영의 ORM 기술 표준이다.

출처 : https://www.tutorialspoint.com/jpa/jpa_introduction.htm

여기서 ORM이란? Object Relational Mapping 말그대로 관계형 데이터베이스와 객체를 매핑 시켜준다는 것이다.

ORM하면 하이버네이트가 떠오를 것이다.

하이버네이트를 기반으로 새로운 자바 ORM기술 표준이 만들어졌는데 이것이 바로 JPA이다!

그림 - JPA 표준 인터페이스와 구현체

출처 : 책 - 자바 ORM 표준 JPA 프로그래밍

위의 UML을 보면 감이 오는가?

JPA는 왜 사용해야하는가?

1. 생산성

2. 유지보수

3. 패러다임의 불일치 해결

4. 성능

5. 데이터 접근 추상화와 벤더 독립성(Dialect) 

라고 한다. 앞으로 정말 그러한지 알아보도록 하자!