Spring XD - Batch Jobs

소개

Spring Batch 기반의 배치 작업을 구동하고 모니터링 하는 기능을 Spring XD에서 제공한다. Spring Batch는 2007년 시작된 프로젝트로서 SpringSource와 Accenture가 함께 협력하였다. 강력한 batch 어플리케이션 개발을 지원하기 위해 이해하기 쉬운 프레임워크를 제공하자는 취지로 시작되었다. 배치 작업은 나름의 best practice와 domain 개념을 가지고 있었고, Accenture의 컨설팅 비즈니스 기반에 Spring Batch는 구축 되었다. Spring Batch가 사용되기 시작한 이후로 수천개의 엔터프라이즈 어플리케이션에 적용되었고 JSR-352(배치 처리를 위한 JSR 표준)의 기반이 되었다.

Spring XD는 파일이나 데이터베이스로부터 데이터를 이동시키는 전통적인 유스케이스 뿐만 아니라 하둡의 사용하는 유스케이스인 하둡클러스터에 분석 로직을 일련의 스텝으로 분할하는 등의 배치 워크플로우를 단순하게 만들 수 있는 솔루션을 제공한다. 워크플로우상 하둡의 Step은 MapReduce 작업이거나 Hive/Pig 스크립트나 HDFS 명령일 수 있다.

WorkFlow

워크플로우에 명시된 Job을 MapReduce job이라 혼동하지 말자. Job는 directed 그래프이다. 그래프의 각 노드는 처리 Step이다. Step은 설정에 따라 순차적으로 처리될 수도 있도, 병렬로 처리될 수도 있다. 작업은 시작될 수 있고, 중지 될 수 있고 재시작 될 수 있다. 작업을 재시작하는 것은 작업과 수행 스텝들이 JobRepository를 저장되어야 가능하다. 다음 그림은 workflow의 기본 컴포넌트들을 보여준다.

하둡처리를 위한 Step을 명시한 작업은 다음과 같다.

Features

Spring XD는 작업을 생성하고 구동하는 것이 가능하다. 작업을 구동하는 것은 data stream에 대한 반응일수도 있고, cron 표현식을 triggering일 수 있다. 작업이 수행될 때 작업은 stream에 의해 구독될 이벤트 데이타의 소스일 수도 있다. 작업의 수행 중 전송된 이벤트에는 몇가지 타입이 있다. 가장 일반적인 경우는 작업으로 획득된 작업과 스텝의 상태일 수 있다. Stream과 배치 처리의 상호 커뮤니케이션을 위해 더 복잡한 처리 순서를 생성할 수 있도록 제공된다. 시작점으로서 다음와 같은 케이스의 작업들은 기본 제공된다.

- Poll a Directory and import CSV files to HDFS

- Import CSV files to JDBC

- HDFS to JDBC Export

- JDBC to HDFS Import

- HDFS to MongoDB Export

Lifecycle

Spring XD에서 배치 작업의 라이프사이클은 다음과 같다.

1. Register a Job Module - 작업 모듈을 Module Registry에 등록하기 위해 XML과 jar 파일을 $XD_HOME/modules/job 디렉토리에 등록하는 것

2. Create a Job Definition - 작업 정의를 생성하는 것. 모든 JobInstance에 적용될 프로퍼티 및 작업 정의명을 전달.이 시점에 작업이 배포된 것은 아님.

3. Deploy a Job - 하나 또는 여러 개의 Spring XD 컨테이너에 작업을 배포. 작업정의는 개별 인스턴스에서 초기화. 작업은 라이브 상태가 된다.

4. Launch a Job - 작업 큐에 작업 파라미터와 함께 메시지를 전송함으로써 작업이 구동. 작업인스턴스는 작업정의 + 실행시 작업 파라미터임. 주어진 작업명으로 작업인스턴스를 조회할 수 있음.

5. Job Execution - JobExecution 객체를 생성하기 위해 작업이 수행됨. JobExecution이 작업의 성공/실패를 포함한 작업의 스냅샷임. 주어진 작업명으로 JobExecution을 조회할 수 있음.

6. Un-deploy a Job - 새로운 작업 인스턴스가 수행되지 않도록 Spring XD 컨테이너에서 작업을 제거. 리포팅 목적으로 작업의 실행 이력은 조회가 가능함.

7. Destroy a Job Definition - 작업 정의 자체를 삭제함.

작업을 생성할때 모든 작업 정의는 다음과 같은 옵션이 사용가능하다.

dateFormat - 작업 파라미터를 위한 date format(기본값:yyyy-MM-dd)

numberFormat - 숫자형 파라미터를 파싱할떄 사용될 number format (기본값:NumberFormat.getInstance(Locale.US))

makeUnique - job parameters를 유니크하게 만들 것인가? (기본값:true)

Stream create command와 비슷하게 job create command도 작업을 생성하고 바로 디플로이하기 위해 --deploy 옵션을 명시할 수 있다. 해당 옵션의 기본값은 false이다.

job create myjob --definition "fooJob --makeUnique=false"

Deployment manifest support for job

배치작업을 배포할때, deployment manifest를 제공할 수 있다. 작업이나 stream을 위한 manifest 프로퍼티 배포는 동일하며, 다음과 같은 것들을 정의할 수 있다.

- 디플로이될 작업모듈 수

- 작업에 사용가능한 컨테이너를 매칭한 기준 표현식

예를들면

job create myjob --definition "fooJob --makeUnique=false"

job deploy myjob --properties

 "module.fooJob.count=3,module.fooJob.criteria=groups.contains('hdfs-containers-group')"

위의 Deployment manifest는 4개의 fooJob 모듈이 그룹명 hdfs-containers-group에 매치되는 컨테이너에 배포된다는 의미이다. 배치 작업이 구동되거나 스케쥴될 때, 작업 모듈은 배치 작업을 구동하기 위해 작업 구동 요청 메시지를 뽑는다. 여러 개의 컨테이너의 작업 분할을 지원하기 위해 작업 정의는 작업이 어떻게 분할 될지를 정의할 필요가 있다. 파티셔닝 타입은 작업의 타입에 의존한다. 만약 JDBC를 통해 읽어드리는 작업은 로우의 수로 테이블의 데이타를 분할할 수 있고, 파일로부터 데이터를 읽어드리는 작업은 이용가능한 파일의 수로 분할할 수 있다.

FTP to HDFS나 FILE to JDBC 작업은 파티셔닝을 지원한다. 파티셔닝 지원을 작업에 추가하기 위해 singlestep-partition-support.xml을 작업 정의에 임포트 해야 한다. 이것은 구동 요청을 처리한 작업 모듈이 마스터로서 다른 작업 모듈과 통신하기 위한 인프라스트럭쳐를 제공한다. 또한 Partioner 인터페이스를 구현해야 한다.

Launching a job 

Sping XD에서는 배치 작업을 구동하기 위해 정규 이벤트 플로우 뿐만 아니라 트리거를 이용한다. 다음과 같은 방법으로 작업을 구동할 수 있다.

- Launch the Batch Job Ad-hoc

- Launch the Batch Job using a named Cron-Trigger

- Launch the Batch Job as sink.

Ad-hoc 

작업을 한 번 구동하기 위해서 job 명령어를 사용할 수 있다. 작업 모듈명이 helloSpringXD라고 되어 있다면 다음과 같다. 

xd:> job launch helloSpringXD

Launch the Batch using Cron-Trigger

배치작업을 cron 스케쥴러 기반으로 구동하기 위해서 trigger source를 활용해 다음과 같이 작성할 수 있다. 

xd:> stream create --name cronStream --definition "trigger --cron='0/5 * * * * *'  > queue:job:myCronJob" --deploy

배치작업은 소스(이 경우에는 트리거) 또는 프로세스 부터 파라미터를 받을 수 있다. 트리거는 payload를 정의하기 위해 --payload 표현식을 사용할 수 있다.

xd:> stream create --name cronStream --definition "trigger --cron='0/5 * * * * *'  --payload={\"param1\":\"Kenny\"} > queue:job:myCronJob" --deploy

다음 스케쥴에 중지를 하기 위해 스트림은 언디플로이 되어야 한다.

xd:> stream undeploy --name cronStream

Launch the Batch using a Fixed-Delay-Trigger

Fixed-delay-trigger는 주기적으로 작업을 구동하기 위해 사용된다. --fixedDeplay 파라미터를 사용하여 실행간의 대기시간(초)를 지정할 수 있다. 아래 예제는 myXDJob을 매 10초마다 payload에 싱글 어트리뷰트를 포함하여 전달할 때이다.

xd:> stream create --name fdStream --definition "trigger --payload={\"param1\": \"fixedDelayKenny\"} --fixedDelay=5 > queue:job:myXDJob" --deploy

다음 스케쥴에 중지를 하기 위해 스트림은 언디플로이 되어야 한다.

xd:> stream undeploy --name fdStream

Launch job as a part of event flow

배치작업은 항상 sink로써 사용된다. 이 말은 소스나 프로세서로 부터 메시지를 받을 수 있다는 것을 의미한다. 이번에는 유저가 만든 http source(http post를 받아서 http message의 payload를 stream의 다음 모듈로 전달)로 부터 http payload를 myHttpJob로 전달해보겠다.

stream create --name jobStream --definition "http > queue:job:myHttpJob" --deploy

스트림을 테스트하기 위해 아래와 같이 수행하자.

xd:> http post --target http://localhost:9000 --data "{\"param1\":\"fixedDelayKenny\"}"

Retrieve job notifications

Spring XD는 작업이 수행중일때 작업으로부터 notification를 확보하기 위한 기능을 제공한다. 배치 작업이 배포될떄,  다음의 리스너들을 pub/sub 채널과 함께 등록할 수 있다. 

Job Execution Listener, Chunk Listener, Item Listener, Step Execution Listener, Skip Listener

Stream은 모든 기본 배치 작업 리스너로 모든 이벤트 메시지를 받을 수 있고 해당 메시지를 로그로 전송한다.

중요 - The syntax for the tap that receives the aggregated events is: tap:job:<job-name>

xd> job create --name myHttpJob --definition "httpJob" --deploy

xd> stream create --name aggregatedEvents --definition "tap:job:myHttpJob > log" --deploy

xd> job launch myHttpJob

중요: The syntax for the tap that receives the job execution events is: tap:job:<job-name>.job

xd>stream create --name jobExecutionEvents --definition "tap:job:myHttpJob.job >log" --

deploy

중요: The syntax for the tap that receives the step execution events is: tap:job:<job-name>.step

stream create --name stepExecutionEvents --definition "tap:job:myHttpJob.step >log" --

deploy

중요: The syntax for the tap that receives the item events: tap:job:<job-name>.item,for skip events: tap:job:<job-name>.skip and for chunk events: tap:job:<job-name>.chunk

xd>stream create --name itemEvents --definition "tap:job:myHttpJob.item >log" --deploy

xd>stream create --name skipEvents --definition "tap:job:myHttpJob.skip >log" --deploy

xd>stream create --name chunkEvents --definition "tap:job:myHttpJob.chunk >log" --deploy

기본 리스너를 비활성화시키는 방법은 다음과 같다.

xd>job create --name myHttpJob --definition "httpJob --listeners=disable" --deploy

다음과 같이 특정 리스너를 명시할 수도 있다.

xd>job create --name myHttpJob --definition "httpJob --listeners=job,step" --deploy

Removing Batch Jobs

배치 작업은 다음과 같은 명령으로 삭제할 수 있다.

xd:> job destroy helloSpringXD

  

다음 배포를 위해 배치 작업 정의는 유지하고 언디플로이만 수행할 수도 있다.

xd:> job undeploy helloSpringXD