우선 JobBuilderFactory와 StepBuilderFactory를 더이상 사용할 수 없는데, JobBuilder와 StepBuilder를 사용하면 된다. 아래는 스텝로그를 찍는 기본 예제이다.

@Configuration
public class HelloJobConfiguration {

    @Bean
    public Job helloJob(JobRepository jobRepository, Step helloStep1, Step helloStep2) {
        return new JobBuilder("helloJob", jobRepository)
                .start(helloStep1)
                .next(helloStep2)
                .build();
    }

    @Bean
    public Step helloStep1(JobRepository jobRepository, PlatformTransactionManager tx) {
        return new StepBuilder( "helloStep1", jobRepository)
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("====================================");
                    System.out.println(" helloStep1 executed ");
                    System.out.println("====================================");
                    return RepeatStatus.FINISHED;
                }, tx).build();
    }

    @Bean
    public Step helloStep2(JobRepository jobRepository, PlatformTransactionManager tx) {
        return new StepBuilder( "helloStep2", jobRepository)
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("====================================");
                    System.out.println(" helloStep2 executed ");
                    System.out.println("====================================");
                    return RepeatStatus.FINISHED;
                }, tx).build();
    }


}

참고로 @EnableBatchProcessing 어노테이션을 제거해야 제대로 로그가 출력된다. 메인 어플리케이션 클래스도 첨부한다.

@SpringBootApplication
public class SpringBatchApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringBatchApplication.class, args);
    }

}

이 링크는 스프링 공식 깃헙에서 기존 코드를 배치5 버전 마이그레이션 가이드이다.

나는 편하게 개발만 하고싶지만!!! 늘 그렇듯 개발환경세팅이 우리들을 괴롭히죠..

저는 자바 개발자이기 때문에 업무 할때도 항상 인텔리제이 엔터프라이즈 버젼을 써왔는데 퇴사 후 개인 노트북으로는 커뮤니티버젼으로 다운그레이드 되어 ㅋㅋ 하지만 JPA 개발할때에는 큰 문제가 없었습니다. 하지만.. JSP 를 다시 공부하면서 복병을 만났죠.

우선 강사님은 이클립스를 사용하셨지만 저는 젯브레인의 골수팬이기에 인텔리제이에서 톰캣도 다운로드하고(지금 생각해보니 스프링이 참 대단한 녀석이었음을 새삼 깨닫습니다..) JSP 파일 작성해서 로컬에 서버까지는 무난무난 잘 되었습니다. 그런데 이제 서블릿을 생성하려고 하니

눈을 씻고 찾아봐도 서블릿은 없읍니다..

그런데 구글링 열심히 해보니 저기에 서블릿을 추가하는 방법이 있는것 같더라구요?

Project Structure 에서 서블릿 api 를 추가하면 된다는 글을 다수 발견.

버젼도 확인해보고 다 해봤지만 역시나 서블릿을 생성할수 없었습니다.

그 다음시도는 preferences 에서 file and code templates 에서 서블릿 추가 도전

하지만 + 버튼 deactive ㅜㅜ

그래서 live 로 해보았으나 역시 실패

..

처음부터 JAVA EE 가 아닌 메이븐으로 빌드 했는데 그거시 문제인가 싶어 새로운 마음으로 프로젝트 새로 생성.

불길하게 JAVA EE 가 없슴 ,,

커뮤니티버젼에서는 JAVA EE 가 생성이 안되는 거였습니다..

인내심의 한계를 느끼고 내일 다시 일어나서 이클립스로 시작하려 합니다.

여러분.. 서블릿 개발하시는 분들 인텔리제이 커뮤니티 버젼이라면 이클립스로 가십시오... 그럼 20000...

누추하게 나마 그림을 그려보았습니다.. (공대오길 잘했다..)

실질적인 데이터 처리는 커넥터 에서 담당하는데, 사용자가 커넥트에 커넥터 생성 명령을 내리면 커넥트가 내부에 커넥터와 태스크를 생성한다.

여기서 커넥트와 커넥터 용어가 헷갈릴 수 있는데, 커넥트는 config등 환경을 설정해주는 역할이고 데이터 처리는 커넥터 에서 처리된다.

1. 소스커넥터 & 싱크커넥터

MySQL, S3, MongoDB 등과 같은 저장소가 대표적인 소스 어플리케이션, 싱크 어플리케이션인데, 소스 애플리케이션에서 가져온 데이터를 커넥트를 통해 카프카 토픽으로 저장하고, 필요에 따라 가공 후 싱크 애플리케이션으로 저장할 수 있다.

• Source Connector

  • 프로듀서 역할을 한다. 말그대로 카프카로 데이터를 가져오는 역할을 한다.

• Sync Connector

  • 컨슈머 역할을 한다. 데이터를 저장함.

2. 컨버터, 트랜스폼

커넥터를 사용해서 파이프라인을 생성할 때 사용할 수 있는 기능들이 있는데 첫번째는 converter, 두번째는 transform 이다.

converter: 데이터 처리 전 스키마 변경시 사용 transform: 데이터 처리 시 각 메세지 단위로 데이터를 변환하기 위한 용도

3. 커넥트 실행 모드 (Standalone, Distributed)

커넥트는 두가지 모드로 실행될 수 있는데, 단일 모드(Standalone) 와 분산모드(Distributed) 이다.

단일모드 커넥트

• 환경 설정은 _**connect-standalone.properties
• *_
• 1개의 프로세스만 실행, 단일장애점(SPOF)

분산모드 커넥트

• 환경설정은 _**connect-distributed.properties
• *_
• 2개 이상의 서버에서 클러스터형태로 운영, 고가용성
• 2개 이상의 프로세스가 1개의 그룹으로 묶여서 사용된다. (group.id)
• 운영환경에서는 거의 분산모드를 사용함

예제 코드는,, 나중에 여유가 되면 추가예정..

1. Push Query
2. Pull Query

에 대하여 정리 해보겠습니다..

1. Push Query

• ksql 서버에서 진행중인 스트리밍 이벤트에 대한 실시간 처리를 요청하는 쿼리
• 연속된 stream을 반환한다.
• 스트리밍 데이터의 변경사항이 발생할 때마다 자동으로 결과를 업데이트 하는 방식

(아래 내용은 하이퍼커넥트 기술블로그에서 참조 했습니다)

Push Query 는 EMIT CHAGES 혹은 EMIT FINAL 구문을 쿼리의 마지막에 붙여줘야 하는데, EMIT CHANGES 를 사용하면 Stream 의 변경 사항들을 연속적으로 반환하는 것이고, EMIT FINAL 은 ‘windowed aggregation’ 에서만 사용할 수 있습니다. 이를 사용하면 ‘마지막’ 윈도우의 결과를 반환해줍니다.

• push 쿼리는 실시간 스트리밍 데이터를 처리하므로 변경사항에 대한 처리가 필요한데 EMIT CHAGES; 쿼리 구문을 보고 push query 임을 알수 있다 (당연한 소리?이다)

EMIT CHAGES;

2. Pull Query

• pull query는 스트리밍 데이터가 아닌, 현재 상태를 기준으로 사용자가 필요할때 쿼리를 통하여 결과를 가져오는 방식이다.
• 특정 시점에서의 상태를 조회할 때 사용한다.
• 일반적인 SQL 조회 쿼리라고 생각하면 쉽다.
• KSQL은 Materialized agg table에서만 pull Query를 지원한다.
• Pull Query 는 반드시 파티셔닝의 기준이 되는 ‘KEY’ 를 기준으로 쿼리를 해야 한다(rowkey). 아래 예제 참고

  SELECT * FROM table_name WHERE rowkey=id;

• Pull Query 는 스트림데이터를 다루는 KStream 이나 KTable 에서는 사용할 수 없다!