GenerateFlowFile 이 만들어 내는 FlowFile 에 1, 2, 3 … 으로 늘어나는 번호를 붙인다. GenerateFlowFile 자체는 상태를 갖지 않으므로 같은 내용만 반복해서 만든다. 번호를 붙이려면 상태를 기억하는 구성 요소가 필요하다.
스크립트를 쓰지 않는 방법이 있다. UpdateAttribute 는 자체적으로 상태를 저장할 수 있다.
프로세서 설정에서 다음을 지정한다.
| 항목 | 값 |
|---|---|
| Store State | Store state locally |
| Stateful Variables Initial Value | 0 |
동적 속성 counter |
${getStateValue('counter'):plus(1)} |
getStateValue 는 Store State 가 켜져 있을 때만 쓸 수 있는 함수이며, 직전 실행에서 저장한 값을 돌려준다. 동적 속성으로 계산한 값이 곧 다음 상태가 된다.
상태는 노드별 로컬 상태로 저장된다. 클러스터에서 노드마다 따로 증가하므로 전체에서 유일한 번호가 필요하면 그 프로세서를 Primary node only 로 두거나 아래 방법을 쓴다. 번호가 겹치지 않기만 하면 되는 경우에는 노드 식별자를 접두사로 붙여도 된다.
Concurrent Tasks 는 1 로 둔다. 2 이상이면 같은 값을 두 스레드가 동시에 읽어 번호가 중복된다.
여러 노드에서 하나의 번호 계열을 써야 하면 클러스터 범위 상태를 쓴다. ExecuteScript 또는 ExecuteGroovyScript 에 Groovy 로 작성한다.
import org.apache.nifi.components.state.Scope
def flowFile = session.get()
if (!flowFile) return
def stateManager = context.stateManager
def stateMap = stateManager.getState(Scope.CLUSTER)
def counter = ((stateMap.get('counter') ?: '0') as long) + 1
stateManager.setState(['counter': counter.toString()], Scope.CLUSTER)
flowFile = session.putAttribute(flowFile, 'seq', counter.toString())
session.transfer(flowFile, REL_SUCCESS)
내용 자체를 번호로 바꾸려면 속성 대신 본문을 쓴다.
flowFile = session.write(flowFile, { out ->
out.write(counter.toString().getBytes('UTF-8'))
} as org.apache.nifi.processor.io.OutputStreamCallback)
주의할 점이 있다. getState 와 setState 사이에 다른 스레드나 다른 노드가 끼어들 수 있다. 완전히 안전하게 하려면 replace(stateMap, newValues, Scope.CLUSTER) 를 쓰고 실패하면 다시 읽어 시도한다. 그래도 경합이 잦으면 처리량이 떨어지므로, 애초에 Primary node only + Concurrent Tasks 1 로 두고 단일 지점에서 번호를 붙이는 편이 단순하고 확실하다.
NiFi 2.x 에서는 ExecuteScript 의 Jython 엔진이 제거됐다. 파이썬으로 같은 일을 하려면 2.x 의 Python 확장 API 로 프로세서를 만들어야 한다. Groovy 는 그대로 쓸 수 있다.
시각 기반 값이면 상태가 필요 없다. 유일성만 필요하면 FlowFile 의 uuid 속성을 그대로 쓰거나 ${now():toNumber()} 를 쓴다.
데이터베이스 시퀀스를 쓰는 방법도 있다. ExecuteSQL 로 nextval 을 받아 오면 여러 NiFi 노드와 다른 시스템까지 하나의 계열을 공유할 수 있다. 정합성이 중요한 업무 키라면 이쪽이 맞다.