욱'S 노트

Introduction

이 문서는 Spring XD 분산 런타임(DIRT)의 내부에서 일어나는 사항 특히, 런타임 아키텍처가 고가용성(HA)를 가지고 클러스트 프로덕션 환경에서 failover를 수행하는지에 대한 내용이다. 설치와 분산 모드 실행에 대한 더 많은 정보는 이전에 자료를 참조하자. Spring XD의 핵심 런타임 컴포넌트와 Spring XD의 상태를 관리하고 실패로부터 자동 복구가 가능하게 하는 Zookeeper의 역할에 대해 살펴본다.

Configuring Spring XD for High Availability(HA)

프로덕션 상태에서 Spring XD 환경은 일반적으로 다수의 호스트간에 분산되어 클러스터된 환경일 것이다. Spring XD는 컨테이너 인스턴스를 추가할 떄마다 수평적으로 확장될 수 있다. 가장 단순한 케이스에서는 모든 컨테이너가 복제품이 되는 것이다.  각 인스턴스는 지정된 호스트에서 설정되고 모든 모듈은 사용가능한 모든 컨테이너에 라운드-로빈 방식으로 디플로이 되는 것이다. 하지만 이러한 단순한 가정은 실제 프로덕션 시나리오를 반영하지 못한다. 실제 상황에서는 리소스  사용을 최적화하기 위해서 컨트롤이 요구된다. 이러한 결과로 Spring XD는 모듈배포 와 특정 컨테이너 설정을 매칭하기 위해 유연한 알고리즘을 지원한다. 컨테이너 매칭 알고리즘은 나중에 살펴보겠다. 일단 가장 심픔한 케이스를 가정하자. 다수 컨테이너 수행은 수평적 확장뿐만 아니라 실패에 대한 복구도 가능하게 한다. 만약 컨테이너가 복구할수 없는 연결 실패로 사용할 수 없게 된다면 모듈은 자동적으로 이용가능한 인스턴스로 배포될 것이다.

Spring XD는 컨테이너와의 stream 배포 요청과 같은 상호작용을 처리하기 위한 하나의 Admin Server가 필요하다. 만약 백업이 없다면 Admin 서버는 single point of failure가 발생할 수 잇다. 그러므로 Admin Server는 프로덕션 환경에서는 두 대 이상으로 설정하는 것을 추천한다. 모든 Admin server는 REST로 요청을 처리 할 수 있지만, 실제로는 하나의 인스턴스 leader만이 요청을 수행하여 런타임 상태를 업데이트해야 한다. 만약 리더가 다운되면 다른 Admin server가 leader 역할을 수행하게 된다. Leader Election은 Curator 프레임워크로 부터 제공된 분산 시스템의 공통적인 기능이다.

HA Spring XD 설치를 위해선 외부 서버가 요구된다. - ZooKeeper, messaging middleware, data store. 외부 컴포넌트에 대한 상세한 설정방법은 컨설팅을 받아라.

ZooKeeper Overview

이전 섹션에서 만약 컨테이너가 다운되면 Spring XD는 해당 인스턴스에 디플로이된 모듈을 다른 이용가능한 컨테이너에 디플로이를 수행한다고 했다. 또한 Admin 리더 서버가 다운된다면 다른 Admin 서버가 리더역할을 수행한다고 하였다. ZooKeeper가 이러한 것을 가능하게 한다. 주키퍼는 분산 시스템 관리 및 코디네이션을 주요하게 사용되는 Apache 프로젝트이다. 이번 섹션에서는 Spring XD에서의 주키퍼의 역할을 이해하기 위해 기본적인 컨셉을 살펴보자.

주키퍼는 단순한 계층적 데이터 구조를 기반으로 한다. 개념적으로나 의미적으로나 파일 디렉토리 구조와 유사하다. 데이터는 노드에 저장된다. 노드는 path로 표현된다. 각 노드는 추가적인 데이터를 저장할 수 있는데 byte array로 직렬화된다. Spring XD는 모든 데이터를 JSON로써 직렬화된 Map이다.

주키퍼 노드는 ephemeral 이거나 persistent이다. 임시 노드는 세션이 유지될 동안만 존재한다. 퍼시스턴트 노드는 당연히 퍼시스턴트 된다. 임시 노드는 컨테이너 인스턴스를 등록하기 위해서 사용된다. Spring XD 컨테이너가 시작하면 내부적으로 생성한 컨테이너ID를 이용하여 /xd/containers/<container-id> 임시 노드를 생성한다. 연결이 종료되어 컨테이너 세션이  종료되면 예를들어 컨테이너에 프로세스가 종료되면 노드는 사라진다. 임시노드는 또한 호스트명, IP, 런타임 메트릭스, 유저 정의 컨테이너 속성등의 메타정보를 유지한다. 퍼시스턴트 노드는 일반 수행 및 복구를 위한 상태를 유지한다. 이것은 스트림 정의, 작업 정의, 디플로이 매니페스트, 모듈 배포 및 스트림과 작업의 배포 상태를 포함한다.

명백하게 주키퍼는 Spring XD 런타임은 크리터컬 부분이다. 당연히 HA를 지원해야 한다. 주키퍼는 앙상블이라고 불리는 분산 아키텍처를 지원한다. 이 문서의 범위를 넘어 더 자세한 내용을 살펴보면 주키퍼 서버 인스턴스는 적어도 3개 이상이여야 한다. 컨테이너와 어드민 노드는 주키퍼 앙상블의 클라이언트이고., 시작시 주키퍼로 반드시 접속되어야 한다. Spring XD 컴포넌트는 zk_client_connect 프로퍼티로 설정할 수 있으며 <host>:<port> 형태의 콤마로 구분된 리스트이다. 주키퍼 클라이언트는 성공할때까지 각 서버로 접속을 수행할 것이다. 만약 접속을 할 수 없다면 계속 시도를 할 것이다. 만약 접속에 실패하면 주키퍼 클라이언트는 다른 서버로 접속을 시도한다. 주키퍼 클러스터는 앙상블 사이의 데이터 복제본 일치를 보장한다.

주키퍼는 다음을 보장한다.

• Sequential Consistency - 클라이언트로부터 변경은 전송된 순서에 따라 적용된다.
• Atomicity - 변경은 성공 혹은 실패이다. 부분성공은 없다.
• Single System Image - 클라이언트는 접속한 서버에 상관없이 동일한 뷰를 본다.
• Reliability - 한번 업데이트가 적용되면 클라이언트에 의해 변경이 일어날때까지 퍼시스턴트된다.
• Timeliness - 클라이언트 뷰는 특정 시간의 최신 상태임을 보장한다.

주키퍼는 데이터는 메모리에 주로 유지되며 디스크 캐쉬에 의해 백업된다. 변경은 복구를 위해 디스크에 로깅 되고 인메모리 데이터베이스에 저장되기 전에 직렬화되어 저장된다. 게다가 노드에 대한 기본적인 CRUD를 수행하기 위해 주키퍼 클라이언트는 노드에 콜백을 등록할 수 있고, 노드나 노드의 자식으로부터 어떤 이벤트나 상태 변경에 응답할 수 있다. 그러한 노드 오퍼레이션과 콜백은 Spring XD 런타임을 관리하는 메커니즘이다.

The Admin Server Internals

하나 이상의 어드민 인스턴스가 수행중이라고 가정하자. 각 인스턴스는 시작할때 리더쉽을 요청한다. 만약 지정된 리더가 존재한다면 인스턴스는 xd/admin 노드를 watch할 것이며 리더가 사라지면 통보를 받을 것이다. Leader는 Curator에 의해 제공된 Leader Selector 방식에 의해 지정된 인스턴스이다.  Curator는 또한 Listener callback 인터페이스를 제공하는데 클라이언트는 노드에 등록된다. 어드민 서버는 최상위 레벨 노드를 생성한다.

• /xd/admin - 자식은 임시 노드로서 각 이용가능한 Admin 인스턴스이며 Leader Selector를 위해 이용된다.
• /xd/containers - 자식은 런타임 속성을 포함한 임시노드로서 호스트명, 프로세스 아이디, 아이피 주소 그리고 유저 정의 속성을 포함한다.
• /xd/streams - 각 스트림 정의 (DSL)을 포함한 퍼시스턴트 노드
• /xd/jobs - 각 작업 정의(DSL)을 포함한 퍼시스턴트 노드
• /xd/taps - 각 배포된 tap의 정의한 퍼시스턴트 노드
• /xd/deployments/streams - 스트림 디플로이 상태를 포함한 노드 (리프노드는 임시노드)
• /xd/deployments/jobs - 작업 디플로이 상태를 포함한 노드 (리프노드는 임시노드)
• /xd/deployments/modules/requested -  디플로이 조건을 포함한 모듈 디플로이 요청을 저장
• /xd/deployments/modules/allocated - 현재 디플로이된 모듈의 정보를 저장

다음과 같은 프롬프트를 볼 수 있다.

ls command를 이용하여 탐색해보자. 유니크한 컨테이너 ID는 차이가 있을 수 있다.

어드민과 컨테이너 인스턴스가 실행된 Spring XD의 초기 상태가 위와 같다. 아직 배포가 되지 않아 stream이나 작업 정의가 존재하지 않는다. /xd/deployments/modules/allocated는 /xd/containers의 ID와 일치하는 퍼시스턴트 자식을 가진다.  만약 분산환경에서 실행한다면 같은 앙상블에 속한 하나의 주키퍼 서버에 접속하게 되면 /xd/containers 및 /xd/admins 하위에 다수의 노드를 볼 수 있을 것이다. 외부의 앙상블이 Spring XD 클러스터의 상태를 저장하기 때문에 모든 디플로이먼트는 Spring XD 클러스터가 종료되어도 존재할 것이다.

xd-shell을 통해 스트림을 생성해보자.

ZK CLI로 돌아가보자.

get 커맨드를 이용하여 새로운 스트림 노드를 조회화면 JSON으로 표현된 stream definition을 조회할 수 있다. ephemeralOwner = 0x0은 임시노드가 아님을 나타낸다. 이제 스트림을 디플로이 해보자.

주키퍼를 확인해보자.

스트림의 status 노드는 디플로이먼트 상태가 디플로이로 나타난다. 

Spring XD는 스트림 디플로이 요청을 개별적인 모듈 디플로이 요청으로 분리한다. 그러므로, 각 스트림의 각 모듈이 컨테이너 인스턴스에 연관되어 있는 것을 알 수 있다. 컨테이너 인스턴스가 하나이기 때문에 같지만 분산환경에서는 스트림의 소스와 싱크는 각각의 컨테이너에 디플로이될 수 있다. 노드명의 형식은 <module_type>.<module_name>.<module_sequence_number>.<container_id> 이다.

디플로이된 모듈의 상세 정보를 저장한 메타데이터 그리고 상태노드는 임시 노드이다.  정보는 XD Shell 쿼리로 제공된다.

Module Deployments

이번 섹션은 Spring XD 런타임이 디플로이먼트를 내부적으로 어떻게 관리하는지 알아볼 것이다. 스트림 디플로이먼트 요청을 처리하기 위해 StreamDeploymentListener는 ContainerMatcher를 호출하여 각 모듈이 배포될 컨테이너 인스턴스를 선택하고 /xd/deployments/modules/requested/ 노드에 모듈의 디플로이먼트 프로퍼티를 기록한다. 만약 매치가 발견되면 StreamDeploymentListener는 /xd/deployments/modules/allocated/<container_id>에 모듈을 위한 노드를 생성한다. 컨테이너는 새로운 모듈 배포를 위해 container node를 모니터하는 DeploymentListener를 포함한다. 디플로이먼트가 성공하면 컨테이너는 새로운 모듈 노드 하위에 상태와 메타데이터에 해당하는 임시노드를 작성한다.

컨테이너가 분리되면 임시노드는 삭제되고 모듈은 언디플로이된다. ContainerListener는 사라진 노드에 응답하여 유효한 모듈을 또다른 인스턴스에 배포를 시도한다.

예를 들어 두개의 컨테이너 인스턴스가 존재하고, 단순한 스트림이 배포되어 있다.

이제 하나의 컨테이너 프로세스를 종료시키고 나머지 컨테이너에 모듈이 리디플로이 되는 것을 확인해보자.

나머지 컨테이너도 종료시키면 다음과 같은 경고 메시지를 볼 수 있다.

소개

Spring XD 분산환경(DIRT)는 다수의 노드간 작업 처리 분산을 지원한다.

Spring XD의 분산 런타임 환경은 다음과 같은 요소들로 구성된다.

• Admin - Stream과 job 디플로이를 관리하고 사용자를 위해 런타인 상태, 시스템 정보등을 REST 서비스로 제공한다.
• Container - Module과 배치 작업이 배포된 호스트
• Zookeeper -  XD 클러스터의 모든 런타임 정보를 제공. 실행 컨테이너 트래킹(모듈 및 작업 배포, 스트림 정의, 디플로이 매니페스트)
• Spring Batch Job Repository Database - 배치 작업을 위해 필요한 RDBMS.  HSQLDB가 제공되나 production에서는 적합하지 않다. JDBC를 지원하는 모든 데이터베이스면 가능하다.
• A Message Broker - 데이터 전송을 위해 사용. XD data tranport는 pluggable하다. 현재  Rabbit MQ와 Redis를 지원하고 Kafka는 stream만 지원한다. 현재 Kafka는 사용할 수 없다.
• Analytics Repository - XD에서는 집계와 측정을 위해 Redis를 사용한다. 

또한 XD는 CLI를 제공한다. 또한 XD-UI를 통해서도 XD Runtime과 통신할 수도 있다.

XD 배포판에서는 런타임 컴포넌트를 시작하기 위한 쉘스크립트를 제공한다. xd-admin, xd-container 혹은 xd-singlenode를 구동하던간에 명령뒤에 --help를 붙임으로써 도움을 얻을 수 있다.

Xd-admin 옵션

analytics - 분석 데이터를 저장하기 위한 데이터스토어. 디폴트는 redis이다.

help - 도움말을 출력한다.

httpPort - REST API Server의 http port. 기본값은 9393

mgmtPort - management server 포트. 기본값은 어드민 서버 포트

XDAdmin을 위한 http port를 지정하는 것을 권고한다. 만약 랜던포트를 사용할 경우 admin 서버의 톰캣이 시작되었을 때, 로그를 통해 확인할 수 있다.

Xd-Container 옵션

analytics - 분석 데이터를 저장하기 위한 데이터스토어. 디폴트는 redis이다.

groups - 컨테이너를 포함시키려는 그룹의 멤버쉽. comma 구분자 리스트

hadoopDistro - HDFS를 사용하기 위한 Hadoop 배포판. 세팅하지 않으면 HDFS를 사용할 수 없다.

help - 도움말을 출력한다.

mgmtPort - management server 포트. 기본값은 컨테이너 서버 포트

RDBMS 세팅

RDMS 세팅을 위해 xd/config/servers.yml을 살펴보자.

Datasource 설정을 변경함으로써 배치 작업 리파지토리를 변경할 수 있다.

다음은 Spring Batch Job Repository에 대한 설정이다. 주의할 점은 spring.batch.initializer.enabled는 기본값이 true이다. 이것은 Spring Batch schema가 생성되어있지 않을 경우 자동으로 생성시켜주는 옵션이다. 이미 테이블이 생성되어 있다면 이것은 무시된다. 

• sessonTimeout - 세션 타임아웃
• connectionTimeout - Connection 타임아웃
• intialRetryWait - 실패 연결 후 재시도 대기시간
• retryMaxAttempts - 실패 연결 후 최대 재시도 횟수

다음 redis를 구동하자.

RabbitMQ 사용

만약 실행중인 RabbitMQ 인스턴스가 있다면 Spring XD에서 사용할 수 있다. Spring XD에서는 기본적으로 localhost 포트 5672로 접근을 수행한다. 기본적인 계정 인증은 guest/guest를 가정한다. 변경하고 싶다면 servers.yml을 수행할 수 있다. 

분산모드 시작하기

Spring XD는 두가지의 서버로 구성된다.

XDAdmin - 컨테이너로 모듈 배포를 관리

XDContainer - 모듈을 수행

아래와 같은 커맨드로 각각 시작할 수 있다.

Spring XD는 모듈의 출력을 다음 모듈의 입력 데이터로 전송한다. 일반적으로 컨테이너 노드간에서는 리모트 전송이 요구된다. 또한 Admin 서버 또한 데이터 버스를 job launch channel을 이용해서 메시지를 전송해 작업을 실행하기 위해서 이용한다. Admin과 모든 컨테이너에의해 똑같은 전송이 공유되기 떄문에 전송 설정은 servers.yml에 있다. 기본 전송은 redis이다. 전송 방법을 변경하기 위해서는 rabiit이라고 전송방법을 명시하거나 다른 지원하는 전송방법을 변경하면 된다. 전송방법에 변경은 클러스터의 모든 XD node에 복제되어야 한다.

만약 다수의 XD 인스턴스가 하나의 RabbitMQ 서버를 공유하고 있다면 각 시스템이 같은 이름의 스트림을 포함하고 있다면 이슈가 될 수 있다. 각 시스템에 대한 다른 RabbitMQ virtual 호스트를 사용하는 것을 추천한다. servers.yml에  spring.rabbitmq.virtual.host 프로퍼티를 업데이트 함으로써 XD의 정확한 virual host를 지정할 수 있다.

기본적으로 XD 컨테이너는 분석 데이터를 redis에 저장한다. --analytic 옵션을 통해 다른 스토어를 지정할 수 있다.

또한 추가적인 설정 옵션은 다음과 같다. Spring XD 설치 위치를 지정하자. 

XD Admin의 http port는 다음과 같이 명시한다.

XD 컨테이너 노드는 기본적으로 server port 0 부터 시작한다. 이것은 사용가능한 http port를 스캔한다는 것을 의미한다. 만약 XD 컨테이너의 HTTP endpoint를 제거한다면 server.port=-1 로 세팅하면 된다. 단 이러한 경우는 PassS 환경에서 HTTP source 지원이 제대로 동작하지 않을 수 있다.  일반적으로 XD는 특정한 port로 바인드하도록 요구되기 때문이다. XDAdmin, XDContainer 프로세스는 server.port로 바인드 된다.

하둡 사용하기

Spring XD에서는 다음과 같은 Hadoop 배포판을 지원한다.

xd/bin>$ ./xd-shell --hadoopDistro <distribution>
xd/bin>$ ./xd-admin
xd/bin>$ ./xd-container --hadoopDistro <distribution>

분산모드에서 쉘 사용하기

만약 XD-Shell이 admin server와 다른 서버에 배포되었다고 가정하자. 아래와 같이 admin server를 세팅할 수 있다.

소개

Spring XD에서 현재 지원하는 4가지 타입은 stream을 위한 source, sink, processor 및 배치처리를 위한 job이다. 이번엔 custom source module을 개발해보자. 

Stream의 첫번째 모듈은 항상 source이다. Source 모듈은 Spring Integration으로 구성되며 외부리소스로부터 inbound channel adapter를 통해 데이터를 피딩받아 output channel로 메시지를 생성하는 역할을 담당한다.

Spring Integration은 다양한 전송 및 데이터 저장소와의 연계를 위해 다양한 adapter들을 제공한다.(JMS, File, Http, Web Services, Mail 등등) 일반적으로 source 모듈은 기제공된 inbound channel adapter를 이용해서 생성하게 된다. 

Create the module Application Context file

소개

Spring XD는 stream을 정의하여 데이터를 획득하는 것을 지원한다. Stream은 module로 구성된다. Module은 캡슐화된 작업 단위를 재활용가능한 컴포넌트로 만든 것이다. Spring XD의 job 또한 모듈로서 구현되어야 한다.

모듈은 타입에 따라 분류될 수 있다. 타입은 일반적으로 모듈의 역할과 기능을 나타낸다. 현재 Spring XD 모듈의 타입은 source, sink, processor 그리고 job이다. 타입은 스트림에서 모듈이 어떻게 조합될지 혹은 배치 작업으로 디플로이 될 떄 사용될 지 여부를 결정한다.

• Source는 외부 리소스로 또는 이벤트로 부터 트리거되어 output을 제공한다. stream의 첫번째 모듈은 반드시 source여야 한다.
• Processor는 입력 메시지를 사용하여 어떤 작업을 수행한 후 새로운 메시지를 생성한다. 그래서 입력과 출력이 모두 요구된다.
• Sink는 입력 메시지를 소비하여 외부 리소스에 데이터를 출력한다. 그리고 stream을 종료시킨다.
• Job은 Spring XD에서 사용가능한 Spring Batch 작업이다.

Spring XD에서는 stream으로 조합할 수 있는 빌트인 모듈을 제공한다. File, HDFS, Spark, Kafka, http, syslog, GemFire 등등. 또한 사용자는 복잡한 빅데이터 어플리케이션을 이러한 모듈들을 자바코드 없이 조합해 만들 수 있다. Spring XD의 모듈을 생성하기 위해서는 Spring, Spring Integration 또는 Spring Batch에 대한 지식은 필수적이다. 다음 내용들은 이러한 주제들에 대해 이미 알고 있다고 가정한다.

Uber-jar로 모듈을 패키징 할때 컴파일 및 테스트를 위해 필요한 Spring XD 라이브러리를 제공하는 빌드 지원 툴은 각각의 지원툴은 Spring Boot Maven Plugin 또는 Spring Boot Gradle Plugin이다.

모듈은 Spring XD Container로 부터 제공되니 않는 디펜던시를 포함해야 한다. 이러한 사항들은 module이 디플로이 될 때 module class loader에 의해 로딩된다. 라이브러리 디렉토리에 포함되는 않는 jar는 ClassDefNotFoundException을 발생시킬 것이다. 게다가 모듈 Spring XD 클래스패스에 다른 버젼의 라이브러리가 존재한다면 confliect나 class loading 이슈가 발생할 수도 있을 것이다. 그러나 걱정할 필요는 없다.

• Spring Boot 패키징 레이아웃은 uber-jar에 제공된 디펜던시가 포함되지 않는 것을 보장해준다. Spring XD 모듈 빌드는 제공된 디펜던시는 spring-xd-dirt로 정의하고 해당 클래스들을 모듈 개발시 제공한다.
• 만약 Spring XD runtime 디펜던시를 포함한 uber-jar라면 해당 디펜던시를 제거해준다.

빌드 지원 툴은 spring-xd-dirt와 spring-xd-test를 정의된 디펜던시를 지원한다. spring-xd-test는 모듈 개발에 유용하다.

• 자바로 정의된 Module Options Metadata
• In-Container 모듈 테스팅 - 임베디드 싱글 노드 컨테이너를 구동, 모듈 배포 및 결과 검증

Testing a Module Project

자세한 테스트 예제는 각각 타입별 모듈 섹션에서 제공한다. 모듈을 빌드하기 위해서는 다음과 같은 커맨드를 수행한다.

Registering a Module

Spring XD Module Registry에 모듈을 등록해보자. 모듈은 stream이나 job으로 배포되기 전에 등록되어야 한다. 모듈이 패키징 되었다면 쉘의 업로드 커맨드를 통해서 등록을 할 수 있다.

모듈정의는 모듈을 유니크하게 식별하기 위해 다음과 같은 속성을 요구한다.

name - 컴포넌트명, 모듈의 목적을 나타내는 한 단어

type - 모듈 타입, source, sink, processor 그리고 job

모듈은 Spring XD에 modules 디렉토리에 위치한다. 그리고 Module Registry는 타입과 일치하는 서브 디렉토리에 따라 구성된다.

Spring XD는 ModuleRegistry라는 전략 인터페이스를 제공한다. 현재 Spring XD ResourceModuleRegistry를 구현하여 제공한다. 

Custom 모듈은 독립 모듈과 분리해서 위치한다. Custom 모듈은 위치를 변경할 수도 있다. severs.yml의 xd.customModule.home의 위치를 수정하면 된다. 디폴트 위치는 ${xd.home}/custom-modules이다. 그러나 custom module이 production에서 사용된다면 네트워크 파일시스템으로 세팅해라. 두가지 이점이 있다. 첫번째, custom 모듈은 XD Admin을 포함한 모든 노드의 XD 컨테이너에서 접근가능해야 된다. 이러면 모든 컨테이너의 해당 모듈이 디플로이 될 수 있을 것이다. 두번째 Spring XD의 설치본에 등록되어 있다면 Spring XD 업그레이드시 사라질 것이다.

Module Class Loading

모듈은 독립된 클래스 로더를 사용하여 모듈의 라이브러리 디렉토리로부터 클래스를 로딩한다. 만약 해당 클래스를 찾을수 없다면 Spring XD에서 일반적으로 사용되는 parent 클래스 로더의 클래스를 로딩할 것이다. 그러나 두가지 사항에 유의하자.

• Spring XD 클래스패스에서 제공된 jar 파일을 modules lib에 넣는것은 피하자. ClassCastException이나 class loading 이슈가 발생할 수 있다.
• 메시지로의 payload 타입에 대한 참조를 가진 클래스는 xd/lib에 설치되어야 한다. 그래야 Producer 및 Consumer 모듈에서 접근이 가능하다.

스프링 프로퍼티로 query, language, consumerKey 그리고 consumerSecret이 제공되고 있다. Spring XD는 이러한 프로퍼티들을 각 모듈 인스턴스에 옵션으로서 제공된다. 이러한 모듈의 노출된 옵션은 TwitterSearchOptionsMedataa.java에 정의되어 있다.

예를 들면 아래와 같이 옵션에 따른 두가지 다른 stream을 생성할 수 있다.

추가적으로 옵션은 모듈에서 Spring Bean 찹조가 될 수 있다. 각 모듈 인스턴스는 빈의 다른 구현을 인젝션할 수 잇을 것이다. 이러한 기능을 통해 같은 모듈 정의를 다른 설정과 함께 자신의 어플리케이션 컨텍스트를 생성할 수 있게 된다. 이러한 유용한 기능의 결과로 input, output과 같은 단순한 프로퍼티 이름을 빈의 표준 아이디로 사용할 수 있게 되는 것이다.

위의 예에서 디폴트값을 사용한 것을 확인 할 수 있다. 때때로 디폴트는 stream명, module명 또는 runtime context로 대체될 수 있다. 예를들면 file source는 디렉토리를 필요로 한다. 적당한 방법은 공통의 input 패스를 디폴트로 정의하는 것이다. 그리고 Stream 정의에서 dir옵션을 통해 디렉토리를 변경하는 것이다. 이러한 사항들은 module info 커맨드를 통해 확인할 수 있다.

또한 Spring XD에서는 모든 모듈에 아래와 같은 공통적인 PlaceHolder를 제공한다.

모듈은 재활용가능한 Spring Integration 혹은 Spring Batch application context이다. 이들은 module options을 통해 동적으로 설정될 수 있다.

모듈 옵션은 stream이나 작업 정의로 부처 설정된 어떤값이다. 되도록이면 사용가능한 옵션을 metadata로서 제공해야 한다. 옵션값을 할당하는 우선순위는 다음과 같다.

1. stream정의에서 제공된 값

2. platform-wide 디폴트 값

3. 모듈의 메타 데이터에서 제공된 디폴트값

Platform-wide 디폴트는 다음과 같은 기준으로 할당된다.

1. <moduletype>.<modulename>.<optionname>로 명명된 system property

2. <moduletype>.<modulename>.<optionname>로 명명된 환경변수

3. <root>/<moduletype>/<modulename>/<modulename>.properties 프로퍼티 파일의 <optionname>이 key인 프로퍼티

4. <root>/ <module-config>.yml의 <moduletype>.<modulename>.<optionname>

Composing Modules

Stream은 모듈의 순서를 정의한 것이다. 때때로 stream는 공통 처리 chain을 공유하고 싶을때가 있다. Composite 모듈은 똑같은 정의 중복을 피하는 방법으로 사용될 수 있다. 그리고 더욱 중요한 것은 성능을 증가하는 방식으로 사용될 수 있다. Stream에 표현된 각 모듈은 디플로이의 단위가 된다. Singlenode 런타임에서는 모듈간 통신을 In-memory 채널을 이용하지만 분산환경에서는 메시징 미들웨어를 통한다. 많은 경우 stream는 미들웨어 전송 및 오브젝트 마샬링을 피하기 위해 함께 배포되는 것이 성능이 더 나을 수 있다. 

Composite 모듈을 생성하기 위한 간단한 커맨드를 보면 다음과 같다.

잘 생성이 되었는지 확인해보자.

Composite 모듈은 Sink로 나타난다. 이유는 이론적으로 sink로 동작하기 때문이다. 그리고 (c) prefix가 붙은 걸 확인할 수 있다.

다음은 두 개의 processor를 composite 해보자. 이런 경우에는 processor로 분류된다.

다음은 source와 processor를 결합하자. 이것은 source로 분류된다.

Composite 모듈의 logical 타입에 근거하여 stream에서 사용할 수 있다. 다음은 사용예이다.

더 이상 composite 모듈을 사용하고 싶지 않다면, delete 커맨드로 삭제할 수 있다. 하지만 이 경우 모듈이 stream에서 사용중에 있다면 먼저 stream을 destroy한 다음 delete를 수행하여야 한다.

모듈을 생성하거나 삭제할 때 타입별로 똑같은 이름이 존재한다면 타입 필터를 사용할 수 있다. (<type>:<module_name>) 

소개

Spring XD에서 가장 기본적인 stream은 source로부터 이벤트 드리븐 데이타를 획득하여 다수의 sink로 전달하는 것이다. Stream 프로세스는 XD Containter상에서 수행되며, Stream 정의는 XD Admin server를 통해 Container로 전달되면 배포가 완료된다. 

Source, Sink 그리고 Processor는 모듈의 기정의된 설정이다. 모듈 정의는 xd-root/xd/modules 하위 디렉토리에서 확인 할 수 있다. 모듈 정의는 일반적인 스프링 정의 파일이다. 정의에서는 EIP 패턴을 지원하기 위해 Spring integration의 Input/Output adapter와 Transformer와 같은 기존의 스프링 클래스들을 사용한다.

Stream을 정의하기 위해서는 고수준의 DSL을 사용한다. 

DSL은 UNIX의 파이프/필터 문법이랑 유사하다. 포트와 파일명을 위한 기본값이 오버라이드 되어 다음과 같이 사용된다.

간단한 스트림 만들기

XD Admin server는 stream 정의는 생명주기를 관리하기 위한 모든 RESTful API를 제공한다. 그러나 가장 쉬운 방법은 XD Shell을 사용하는 방식이다. Shell을 시작하는 간단한 방법은 Getting started를 참조하자.

새로운 스트림은 스트림 정의를 포스팅함으로서 생성된다. 정의는 단순한 DSL로 부터 생성된다. 예를 들어 다음과 같은 shell 명령을 수행해보자.

ticktock으로 명명된 DSL 표현식 time | log 에 기반한 정의를 생성한다. 파이프 심볼은 source와 sink를 연결하기 위해 사용한다.

그리고 스트림을 배포하기 위해서는 다음과 같은 커맨드를 사용한다.

스트림 서버는 module 디렉토리에서 time | log 정의를 발견하고 stream을 설정한다. 단순한 예제이지만 time 소스는 매초당 현재 시간을 log 싱크로 전송하고 싱크는 로깅 프레임워크를 사용하여 출력한다.

만약 스트림의 모듈의 다수 인스턴스를 생성하고 싶다면 프로퍼티를 포함하여 디플로이 커맨드를 할 수 있다.

또한 SpEL 표현식을 포함하여 모듈에 조건을 프로퍼티에 포함 시킬 수도 있다. 모듈의 인스턴스들은 표현식이 참으로 판단된 Containter에만 배포될 것이다.

스트림 디플로이/언디플로이

종종 스트림을 중지하기를 원하지만 미래에 사용하기 위해 정의를 유지하고 싶을때가 있다. 이러한 경우 undeploy 명령을 통해 스트림을 중지시키고 deploy 명령을 수행하여 다음에 재시작하면 된다.

예제

Time 소스보다 좀더 복잡한 예제를 살펴보자. Http는 지원되는 또다른 소스 타입이다. HTTP POSTS를 통해서 데이터를 획득한다. Http 소스는 어드민 서버(기본 8080)랑 다른 포트(기본 9000)로 데이터를 취득한다는 것을 주의하자.

다음과 같은 커맨드를 수행해보자.

그리고 커맨드를 이용해 post 데이터를 전송해보자.

log 싱크에서는 결과를 출력으로 보낼 것이다.

스트림 처리

다음은 단순한 스트림 처리의 예이다. HTTP 포스트 데이터를 대문자로 변경하는 것이다.

DSL 문법

위의 예에서 소스와 싱크를 연결하기 위해서는 파이프 심볼 |을 사용하였다. 또한 소스와 싱크 설정에 파라미터를 전달할 수 있다. 만약 http 소스에서 포트를 8000으로 바꾸고 싶다면 다음과 같이 정의할 수 있다.

스프링 설정 파일에 대한 조금만 알고 있어도 모듈 정의에 해당 프로퍼티들이 노출되어 있다는 것을 알 수 있을 것이다.

복잡한 기능들

위에 제공된 예제에서는 각 스트림은 단순한 모듈 정의로 생성되었다. 하지만 모듈들은 메시징 미들웨어를 통한 데이터 전송 및 중복을 피하기위해 그룹핑 될 수 있다. Composing Modules 섹션에서 더 많은 특징에 대해 알 수 있다.

단순한 선형 처리를 대신해 방향성 있는 그래프 처리가 요구 될 수도 있다. 두 가지 기능이 여기에 연관된다. 첫번째는 채널로부터 복수의 플로우를 결합하기 위해서 named channel을 사용하는 것이다. 이러한 채널의 특징은 queue-based 또는 topic-based 채널이며, 접두어를 사용한다.(queue:myqueue, topic:mytopic) Named Channel 섹션에서 더 확인해보자. 두번째는 실행시 어떤 정보에 의해서 출력 채널을 결정해야할 경우가 있다. 이러한 경우에 대한 사항은 Dynamic Router 섹션에서 확인해보길 바란다.

소개

앞에서 현재 XD에서는 4가지 타입의 모듈(source, sink, processor, job)을 지원한다고 했다. 이제 job 모듈을 개발해보자.

간단한 작업 개발하기

사실상 Spring XD에서 job 모듈을 개발한다는 것은 Spring Batch를 기반으로 한다. 일반적인 배치작업을 개발하기를 원한다면 Spring Batch를 참조하기 바란다.

먼저 배치작업을 하나 정의해보자. 그냥 간단한 tasklet하나를 다음과 같이 지정하였다.

설정 파일의 이름은 job-hello.xml으로 하자. 잘 살펴보면 일반 스프링 배치와 차이가 있다. 그것은 별도의 jobRepository의 설정이 필요없다는 것이다. 기본적으로 Spring XD에서 jobRepository를 관리하므로 여기서는 설정을 할 필요가 없다.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
       xmlns:batch="http://www.springframework.org/schema/batch"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd">
    
    <batch:job id="helloJob">
        <batch:step id="step01">
            <batch:tasklet ref="helloTasklet"/>
        </batch:step>
    </batch:job>

    <bean id="helloTasklet" class="net.daum.datahub.test.HelloTasklet"/>
</beans>

Tasklet를 다음과 같이 작성하자.

public class HelloTasklet implements Tasklet {
    @Override
    public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception {

        System.out.println("Hello XD First Bacth !!!");

        return RepeatStatus.FINISHED;
    }
}

단순하게 메시지만 콘솔로 출력하는 예제이므로 이미 개발은 끝났다.

테스트하기

일단 Spring XD에서 작업을 구동해보기 위해 배포를 해야 한다. 일단 실행 클래스를 jar 파일로 만들자.

$XD_HOME/xd/modules/job 밑에 job-hello란 디렉토리를 만들고 다음과 같은 구조로 배포하자.

Xd-shell 콘솔로 가서 모듈을 조회해보자. 해당 모듈 디렉토리에 빈설정 파일을 자동으로 인식하여 등록이 된다.

job-hello가 등록된 것을 확인할 수 있다.

다음과 같이 배포를 하고 실행을 시키면 된다.

XD쪽 로그를 보면 정상적으로 콘솔에 출력이 발생한다.

결론

개발은 Spring Batch나 Spring Hadoop을 아는 사람이면 쉽게 할 수 있을듯 하다. 오히려 특정한 배포 구조를 맞추는 것이 오히려 관건일듯도 하다.

1.1.0-SNAPSHOT에서는 maven-plugin 등의 feature가 추가되고 있는 듯 하다. Spring Hadoop의 작업으로 테스트를 수행해보고 싶었으나, 레퍼런스 매뉴얼에도 걍 Spring Hadoop을 참고하라고만 나온다. 쩝~

소개

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이다.

Deployment manifest support for job

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

- 디플로이될 작업모듈 수

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

예를들면

위의 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라고 되어 있다면 다음과 같다. 

Launch the Batch using Cron-Trigger

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

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

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

Launch the Batch using a Fixed-Delay-Trigger

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

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

Launch job as a part of event flow

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

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>

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

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

중요: 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

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

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

Removing Batch Jobs

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

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

소개

Spring XD는 데이터 획득, 실시간 분석, 배치 처리 및 데이터 추출을 위한 단일화된 분산 확장 서비스이다. Spring XD의 아키텍처는 백명이상의 개발자가 수년에 걸쳐 진행한 Spring Batch, Integration and Data 프로젝트에 기반한 아키텍처이다. 이러한 프로젝트 기반하에 Spring XD는 데이터 처리를 시작하기 위해 즉시 사용할 수 있는 서버와 설정 DSL을 제공한다. 

Spring XD는 두가지 모드의 동작을 가지고 있다. Single 모드는 하나의 프로세스가 모든 processing 및 administration을 담당한다. 이 모드는 어플리케이션의 쉽고 단순하게 개발하고 테스팅할 수 있도록 도와준다. 분산 모드는 클러스트에 작업을 분산할 수 있다. Administrative 서버는 user command와 runtime event에 반응하여 클러스터 공유된 런타임의 상태 및 작업 수행을 관리한다.

Runtime Architecture

Spring XD의 가장 중요한 컴포넌트는 XD Admin과 XD Container 서버이다. HTTP를 통해 Admin server에 DSL를 사용해서 요청된 작업 처리를 요청 할 수 있다. 그러면 Admin 서버는 수행 작업을 수행 모듈로 매핑한다. 모듈은 Spring ApplicationContext로 구현된 실행의 단위이다. 분산 런타임은 multiple XD 컨테이너 서버 중 실행할 모듈을 제공한다. 하나의 XD 컨테이너는 여러개의 모듈을 실행할 수 있다. Single 모드를 사용한다면 모든 모듈은 하나의 XD Container와 XD Admin Server가 하나의 프로세스에서 실행 될 것이다. 

DIRT Runtime

분산환경 줄여서 DIRT는 여러개의 XD 컨테이너 인스턴스에 작업을 분산 할 수 있다. XD Admin 서버는 수행 작업을 개별적인 모듈 정의로 쪼갠다. 그리고 각 모듈을 ZooKeeper를 이용해 컨테이너 인스턴스에 할당한다. 각 컨테이너는 모듈 정의를 기다렸다가 모듈이 할당되거나 디플로이되면 Spring ApplicationContext를 수행한다

모듈들은 설정된 메시징 미들웨어(Rabbit, Redis) 사용하여 데이터를 공유한다. 모듈간의 메시징을 감소하기 위해서 여러개의 모듈을 하나의 모듈처럼 동적할 수 있게 결합시키기도 한다.

Support for other distributed runtimes

1.0 릴리이즈부터 XD Admin과 XD 컨테이너 인스턴스를 수행하는 것을 Spring XD 자체적으로 가능하다. 또 다른 대안으로 Spring XD를 Hadoop의 YARN에서 수행가능하며 미래에는 Pivotal Cloud Foundary를 지원할 예정이다. 

Single Node Runtime

싱글 노드 런타임은 같은 프로세스에서 Admin, Container, Zookeeper 및 HSQLDB를 실행시킨다. 싱글 노드 런타임은 테스팅과 개발을 목적으로 하지만 작은 프로덕션의 경우에는 적당할 수 있다. 

Admin Server Architecture

Admin Server는 embeded servlet 컨테이너를 이용해 REST endpoint를 노출한다. REST를 이용해 stream의 생성, 디플로이, 언디플로이, 제거 및 작업, 런타임 상태 조회, 분석과 가은 기능을 제공한다. Admin Server는 Spring MVC와 Spring HATEOAS의 구현체이며 REST의 표현은 HATEOAS의 정책을 따른다. Admin Server와 Container 서버의 모니터 및 런타임 상태 업데이트는 ZooKeeper를 이용한다.

Container Server Architecture

Spring XD 데이터 처리에서 주요한 컴포넌트는 다음과 같다. 

Streams, Jobs, Taps

Stream은 어떻게 event-driven 데이터가 수집되고 처리되고 저장되고 전달되는지에 대한 정의이다. 예를 들어 stream은 syslog data를 수집해서 필터링하여 HDFS에 저장할 수 있다.

Job은 조잡하고 시간이 많이 걸리는 배치 프로세싱 단계를 잘 편성한 것이다. 예를들어 작업은 HDFS 연산을 수행하고 이어 여러개의 MapReduce 작업 처리를 수행하도록 정의할 수 있다.

Tap은 Stream이나 Job이 처리에 간섭하지 않고 데이터를 처리하는데 사용된다. 전화에서 사용하는 wiretap처럼 Stream에 위치한 Tap은 데이터를 consuming할 수 있다. Tap은 원래의 Stream처리에 영향을 주지 말아야 한다.

Streams

Spring XD는 event stream을 처리하는 프로그래밍 모델은 Spring Integration 프로젝트가 구현체이기도 한 EIP 패턴이다. 프로그래밍 모델은 쉽게 테스트할 수 있도록 설계되었다. Stream은 다음과 같은 타입의 모듈들로 구성된다. Input Source, Processing Steps, Output sink.

Input source는 외부 소스로 부터 메시지를 생성한다. XD는 다양한 소스(예 : syslog, tcp, http)를 지원한다. 모듈의 출력은 메시지인데 메시지는 payload와 key-value 형태의 메시지 헤더로 구성된다. 메시지는 소스로부터 message channel을 통해 흐르고, 선택적으로 Processing Step을 거쳐 Output sink로 전달되다. Ouput sink는 메시지를 외부 리소스로 전달한다. 

단순한 stream 선형처리는 UNIX의 파이프엔 필터 모델에 비유할 수 있다. 필터는 어떤 이벤트를 생산하고 처리하고 소비하는 컴포넌트이다. 이것은 stream(source, processing steps and sink)의 모듈과 일치한다. 파이프는 필터간에 데이터가 전송되는 길이다. 이는 stream의 데이터를 전송하는 Message Channel과 일치한다. HTTP로 데이터를 읽어서 그것을 파일로 작성하는 간단한 stream을 UNIX 파이프/필터 문법을 사용하여 정의하면 다음과 같다. 

파이프는 HTTP source로 부처 File sink로 데이터를 전달하는 메시지 채널을 표현한다. 메시지 채널의 구현은 메모리, Redis 혹은 RabbitMQ일 수 있다. 추상화된 메시지 채널은 pluggable하게 데이터 전송 구현체를 교체할 수 있도록 XD 아키텍처에서 지원한다. 미래 버젼에서는 JMS 같은 전송을 지원할 예정이다.

UNIX 파이프/필터 문법은 선형 처리를 위한 Spring XD DSL의 기반이다.  비선형 처리는 named channel을 사용해서 부분적으로 지원한다. 하나의 스트림을 여러개의 스트림으로 효과적으로 분리하기 위해 라우터 싱크와 결합할 수 있다. 미럐에는 추가적은 비선형 프로세싱 지원이 있을 것이다.

Stream에 기반한 프로그래밍 모델은 버젼 4.0부터 Spring Integration으로부터 Spring Core에 포함되었다. 메시지 핸들러 클래스 주요 컨셉은 입력된 데이터를 처리하기 위한 코딩 표준을 단순하게 하자는 것이다. 예를 들어 http source로 부터 전달된 HTTP POST 바디 데이터는 다음과 같이 처리 할 수 있다.

메시지로부터 수신된 payload는 메소드 퍼리에 string으로 전달된다. Payload의 내용은 http request의 body이다. 해당 처리의 리턴값을 다음 단계의 Message로서 전달 될 것이다. 이러한 프로그래밍 표준은 매우 쉽게 당신의 processor 컴포넌트를 독립적으로 개발하고 테스트 할 수 있게 한다. Spring XD에서는 SpEL과 Groovy를 이용한 필터와 트랜스포머를 개발하는데 코드가 필요없도록 한다. 예를 들어 트랜스포머와 같은 처리 스텝을 stream을 추가하는 단순한 방법은 다음과 같다.

Streams Deployment

Container 서버는 ZooKeeper를 통해 모듈 디플로이 이벤트를 받을 수 있도록 받을 수 있도록 구동된다. 컨테이너 노드가 모듈 디플로이 이벤트를 처리할 때, 모듈의 입력과 출력 채널은 Stream 처리를 수행하기 위해 데이터 버스에 연결된다.  

싱글 노드 설정에서는 데이터버스는 인메모리 다이렉트 채널이고, 분산 설정에서는 데이터버스는 설정된 전송 미들웨어를 통해 통신한다. Redis, Rabbit 모두 Spring XD 분산을 지원한다.

예제 http | file에서 Admin는 각 모듈을 분리된 컨테이너 인스턴스에 할당할 수 있다. 모듈의 정의는 Module Registry에 저장된다. 모듈은 Spring XML 정의을 포함하며, 모듈의 동작을 담당하는 클래스 및 디펜던시 jar들로 구성된다. 모듈정의는 DSL로 전달된 파라미터로 모듈 정의를 커스터마이징할 수 있는 변수 placeholder를 포함한다. 예를 들어 http listening port를 옵션으로 전달하면 모듈 정의 placeholder값을 대체시킨다.

Module Registry는 파일 시스템 <xd-install-directory>/modules를 기반으로 한다. 컨터이너에 의해 모듈 디플로이먼트가 처리 될때, 모듈정의는 registry로 부터 로드되고 Container 프로세스내에 모듈을 실행하기 위한 새로운 Spring ApplicationContext를 생성한다. 

Jobs

배치 작업의 생성과 실행은 기능적으로 Spring Batch와 Spring for Apache Hadoop 프로젝트를 기반으로 한다.

Taps

Taps는 Stream이나 Job에 의한 데이터처리를 간섭하지 않고 소비하는 것을 제공한다. Taps는 Stream 데이터의 메트릭스 수집이나 분석을 수행하는데 추천한다.

소개 

Spring XD는 stream을 정의하기 위한 DSL을 제공한다. 배치 작업의 스텝 처럼 정교한 stream을 정의할 수 있도록 진화해오고 있다.

Pipes and Filters

단순한 선형의 stream은 일련의 순서를 가진 모듈들의 집합이다. 기본적으로 Input Source, Processing Steps(optional) 그리고 Ouput Sink 이다. 가장 단순한 예제는 Http로 부터 읽어서 File 로 출력하는 것이다. DSL로 표현하ㅕㄴ

스트림은 어떤 처리를 포함할 수 있다.

스트림에서 모듈간의 연결은 |를 이용해서 표현한다.

Module Parameters

각 모듈은 파라미터를 가질 수 있다. 모듈에 의해 제공할 수 있는 파라미터는 모듈의 구현체에 의해 정의된다. 예를 들어 http source 모듈은 port를 노출한다. 데이터 획득 포드를 세팅하기 위해,

만약 내용이 space나 | character를 포함한다면 quote 파라미터를 사용하면 된다.

만약 파라미터의 값이 '이 포함되어있다면 두번 single quote를 사용하면 된다.

Named Channel

Source나 sink 대신에 named channel을 사용하는 것이 가능하다. 일반적으로 stream에 있는 모듈은 인터넷 채널을 사용하여 접속을 한다. 하지만 명시적인 named channel을 사용함으로써 더 정교한 플로우를 만드는 것이 가능하다. Unix 처럼 sourcing/sinking 데이터의 from/to을 > 로 표현한다. Named channel은 channel type을 사용하여 명시한다. 채널 타입은 다음과 같다.

다른 입력 채널로부터 전달된 데이터를 공유하기 위해 named channel을 사용한 예제이다.

taps 라는 특수한 named channel은 publishing data를 mutiple sink할 수 있다.

mystream이 받아서 파일과 로그로 데이터를 쓴다.

Labels

Label은 alias 또는 group modules를 의미한다. Label은 :를 사진 단순한 이름이다. 다른 스트림들이 쉽게 특정한 모듈의 정의를 이용하기 쉽게 해준다.

Label은 동일한 이름의 모듈이 하나의 스트림에서 사용될 때 헷갈리지 않도록 해주는데 유용하다.

소개

Spring XD는 설정할 수 있는 두 가지 주요 부문이 있다. 서버와 모듈이다. 서버(xd-singlenode, xd-admin, xd-container)는 Spring Boot 어플리케이션이고, 설정하는 방법은 Spring Boot Reference를 참조하기 바란다. YAML 기반의 설정 파일 servers.yml을 수정하는 것이 가장 단순한 케이스이다. servers.yml 설정 파일의 값들은 XD jar에 포함되어 있는 디폴트 application.yml 파일의 값들을 오버라이트 한다.

서버 설정

모듈 설정

모듈은 이름과 타입에 기초한 내포된 디렉토리 구조에 프로퍼티 파일을 위치함으로써 설정할 수 있다. 내포된 디렉토리 구조는 루트는 기본적으로 $XD_HOME/config/modules이다. 이 위치는 OS 환경변수 XD_MODULE_CONFIG_LOCATION에 의해 변경될 수 있다. 만약 XD_MODULE_CONFIG_LOCATION이 명시적으로 세팅되었다면 패스의 끝에 file path separator("/")를 붙일 필요가 있다.

예를 들어 twittersearch 모듈을 설정하기를 원한다면 다음과 같은 경로에 파일을 만들어야 한다.

그리고 파일의 내용은 consumerKey와 consumerScret 같은 프로퍼티명을 포함한다. 

모듈 프로퍼티 파일을 다양한 방법이 있다.

1. stream이나 job DSL 정의에서 명시 될수 있다.

2. 자바 시스템 프로퍼티

3. OS 환경 변수

4. XD_MODULE_CONFIG_LOCATION/<type>/<name>/<name>.properties

5. 모듈 메타데이터의 명시된 디폴트 값

XD_MODULE_CONFIG_LOCATION/<type>/<name>/<name>.properties의 값은 다른 리소스 위치에 정의된 키를 참조하기 위한 프로퍼티 플레이스홀더일 수 있다. 디폴트로 제공되는 리소스는 XD_MODULE_CONFIG_LOCATION/modules.yml이다. 서버 실행 스크립트를 실행하기 전에 XD_MODULE_CONFIG_NAME OS 환경변수를 세팅함으로써 변경할 수 있다.

modules.yml 파일은 다른 모듈에서 공유할 수 키의 값을 명시할 수 있다. 예를들어 같은 트위터 개발자 인증을 공통으로 사용하기 위해 modules.yml에 다음과 같이 정의한다.

그리고 XD_MODULE_CONFIG_LOCATION\source\twitterstream\twitterstream.properties에서는 아래와 같이 사용한다.