FTP 에서 받은 파일을 하루 종일 PutHDFS 로 적재한다. 적재는 들어올 때마다 하되, 뒤에 붙은 ExecuteStreamCommand 는 하루에 한 번, 23:00 에만 돌아야 한다. 적재 건수는 날마다 다르고, 어떤 날은 파일이 아예 들어오지 않을 수도 있다.
ControlRate 는 통과량을 제한하는 프로세서다. 지정한 창(Time Duration) 동안 지정한 양(FlowFile 수 · 바이트 · 속성 값의 합) 만 내보내고 나머지는 큐에 남긴다. 버리지 않는다.
그래서 "하루에 한 건만 통과시키고 나머지는 없애자" 는 용도로는 맞지 않는다. 하루에 한 건만 내보내면 나머지 수천 건이 큐에 그대로 쌓여 back pressure 에 걸리고, 다음 날에는 그 잔여분이 먼저 나간다.
또한 ControlRate 에는 초과분을 따로 보내는 관계가 없다. 관계는 success 와 failure 뿐이며, 초과한 FlowFile 은 관계로 나가지 않고 입력 큐에 머문다. 화면에서 rate.exceeded 같은 관계를 찾을 수 없는 것은 그 때문이다.
필요한 것은 "그날 첫 번째 한 건만 남기고 나머지는 버린다" 는 동작이므로 DetectDuplicate 를 쓴다. 캐시에 같은 키가 이미 있으면 duplicate 관계로 보내고, 그 관계를 auto-terminate 하면 사라진다.
흐름은 다음과 같이 둔다.
ListFTP → FetchFTP → PutHDFS → DetectDuplicate → (non-duplicate) → Wait 또는 큐 대기 → ExecuteStreamCommand
→ (duplicate) → auto-terminate
DetectDuplicate 설정은 다음과 같다.
| 속성 | 값 |
|---|---|
| Cache Entry Identifier | ${now():format('yyyyMMdd', 'Asia/Seoul')} |
| FlowFile Description | 기본값 |
| Age Off Duration | 48 hours |
| Distributed Cache Service | DistributedMapCacheClientService |
| Cache The Entry Identifier | true |
캐시 키를 날짜 하나로만 두면 그날 첫 건만 통과한다. 파일 종류별로 한 건씩 필요하면 키에 구분자를 넣는다.
${listftp.remote.source}-${now():format('yyyyMMdd', 'Asia/Seoul')}
Age Off Duration 을 두지 않으면 캐시 항목이 계속 쌓인다. 하루 주기라면 48 시간 정도로 둔다.
DistributedMapCacheServer 컨트롤러 서비스를 함께 만들어야 한다. 클러스터에서는 서버가 한 노드에만 뜨고 나머지 노드가 그 노드를 바라보므로, 클라이언트 서비스의 Server Hostname 을 고정된 노드로 지정한다. 그 노드가 내려가면 중복 판정이 동작하지 않는다는 점은 감안해야 한다.
통과한 한 건을 23:00 까지 붙잡아 두는 방법은 두 가지다.
가장 단순한 방법은 ExecuteStreamCommand 자체를 CRON driven 으로 두는 것 이다. 스케줄을 0 0 23 * * ? 로 두면 그 시각에 스케줄되고, 큐에 통과된 한 건이 있으면 그것을 처리한다. 파일이 하나도 들어오지 않은 날은 큐가 비어 있으므로 아무 일도 하지 않는다. 이 동작이 요구사항과 정확히 맞는다.
중간 큐에 남은 건이 다음 날로 넘어가는 것을 막으려면 그 연결에 FlowFile Expiration 을 걸어 둔다. 예를 들어 23 hours 로 두면 처리되지 못한 건은 만료돼 사라진다.
두 번째 방법은 Wait · Notify 를 써서 별도의 트리거 신호를 기다리게 하는 것인데, 트리거를 만들 GenerateFlowFile 이 또 필요하므로 위 방법보다 복잡하다.
23:00 에 도는 작업이 "오늘 날짜" 를 쓰려면 서버 타임존이 실제 업무 기준과 같아야 한다. 서버가 UTC 이면 한국시간 23:00 은 UTC 14:00 이라 날짜가 같지만, 자정을 넘기는 작업에서는 날짜가 하루 어긋난다.
Expression Language 의 format 은 두 번째 인자로 타임존을 받는다. 이것이 가장 확실하다.
${now():format('yyyyMMdd', 'Asia/Seoul')}
시간을 더하는 함수 이름은 plus 이며 숫자에만 적용된다. 날짜에 시간을 더하려면 밀리초로 바꾼 뒤 더하고 다시 포맷한다.
${now():toNumber():plus(3600000):format('yyyyMMdd')}
${now():plus(1h)} 같은 표기는 없다. 사람이 읽는 기간 표기를 그대로 받는 EL 함수는 존재하지 않으므로 밀리초로 계산한다.