Kudu 디스크가 작아 NAS·Kafka 원천의 bronze 테이블 34개를 8일만 보관하기로 하고, 매일 새벽 3시(KST)에 오래된 행을 지우는 NiFi PG 를 만든 구성이다. 함께 NiFi 노드가 GMT 로 도는 환경에서 "전날" 재수행 일자를 넘기는 방법을 정리한다.
Kudu Table Retention PG 아래에 테이블마다 bronze.<테이블명> PG 를 두고 각각 프로세서 4개로 구성한다.
KST 03:00:00 수행 (GenerateFlowFile, CRON, Primary node only)
↓
유지 기간 (UpdateAttribute : retention_days = 8)
↓
Delete 쿼리 생성 (ReplaceText)
↓
Delete 쿼리 실행 (PutSQL → Impala DBCPConnectionPool)
Delete 쿼리는 ReplaceText 에서 만든다.
DELETE FROM bronze.nas_pt_login
WHERE login_ts < '${now():toNumber():minus(${retention_days:toNumber():multiply(86400000)}):format('yyyy-MM-dd','GMT')}'
PutKudu 로 할 수 없다(PK 단건 삭제만 가능). Impala JDBC 커넥션 풀을 쓰는 PutSQL 로 실행한다. Kudu master 주소는 필요 없다.PutSQL 의 SQL Statement 속성이 비어 있으면 FlowFile 콘텐츠를 SQL 로 실행한다.UpdateAttribute 는 같은 프로세서 안의 다른 속성을 참조할 수 없다(모든 속성이 입력 FlowFile 기준으로 동시에 평가된다). retention_days 와 cutoff_date 를 한 프로세서에 두면 동작하지 않으므로 값 하나만 두고 SQL 안에서 계산한다.timestamp)은 백틱으로 감싼다.PutSQL 의 failure/retry 는 Retry 3회 · Penalize · 30초로 둔다.NAS 로그 파일은 자정에 이름.log.yyyymmdd 로 롤오버되므로 전날 파일을 한 번 더 읽어야 한다. NiFi 와 서버가 모두 GMT 라면 EL 로 now() 를 계산하는 방식은 실행 시각에 따라 날짜가 어긋난다. 날짜 문자열은 항상 KST 로 못박아 셸에서 만들어 주입한다.
REPRODUCE_DATE=$(TZ=Asia/Seoul date -d "yesterday" +%Y%m%d)
주입 경로가 NiFi 변수/파라미터면 REST API 로, ListenHTTP 트리거면 curl 로 값을 실어 보낸다. 재수행용 GenerateFlowFile 은 Primary node 로 두되 그 뒤 SplitText / ExtractText 는 all nodes 로 둘 수 없다는 경고가 나오므로 앞단만 primary 로 둔다.
유지 기간 프로세서의 값 하나만 바꾼다. 값이 두 군데 흩어지지 않게 한다.