NiFi 기반 신고 데이터 수집·전송 파이프라인 설계

1. 전체 구조

수동 입력과 DB 저장분을 하나의 공통 흐름으로 합쳐 다른 서버로 전송하는 구성을 정리함

1.1 입력 경로

• 수동 입력 : 담당자가 직접 입력한 신고 데이터를 HTTP 요청으로 수신함. NiFi의 ListenHTTP 프로세서 사용
• DB 저장분 : 시스템에 체계적으로 쌓이는 신고 데이터. 원본 DB의 insert trigger로 별도 테이블에 적재한 뒤 NiFi가 조회하는 Trigger 방식 사용

1.2 공통 처리 흐름

[수동 입력]  ListenHTTP ──┐
                          ├─→ ScriptedTransformRecord(uuid 부여) ─→ InvokeHTTP(전송)
[DB 저장분]  Trigger 테이블 ─┘

• uuid 부여 : 두 경로 모두 ScriptedTransformRecord로 행마다 고유한 uuid를 붙임
• 전송 : uuid가 붙은 레코드를 InvokeHTTP로 대상 서버에 전달함

입력 경로는 둘로 나뉘지만 uuid 부여와 전송은 하나의 흐름으로 통일하는 것이 관리 포인트를 줄이는 핵심이다

---

2. 수동 입력 : ListenHTTP

외부에서 POST로 들어오는 데이터를 받아 NiFi 흐름으로 넘기는 진입점

• Base Path / Port : 외부에서 요청을 보낼 경로와 포트를 지정함
• 본문 형식 : JSON으로 통일하면 이후 JsonTreeReader로 바로 레코드 처리가 가능함
• 응답 : 기본적으로 200을 반환함. 인증이 필요하면 앞단 프록시나 Authorized DN 설정을 고려함

---

3. DB 저장분 : Trigger 테이블 방식

NiFi는 DB 트리거를 직접 받을 수 없으므로 트리거가 적재한 테이블을 조회하는 Outbox 구조로 구성함

3.1 Trigger 테이블 적재

NiFi가 DB 변경을 받는 방법은 크게 세 가지로 나뉨.

방식 비교

방식	설명	비고
Outbox 테이블	트리거가 변경분을 별도 테이블에 적재, NiFi가 QueryDatabaseTableRecord 등으로 조회	가장 단순하고 안정적
트리거에서 HTTP 호출	트리거가 직접 ListenHTTP로 전송	DB 부하와 장애 전파 위험으로 비권장
CDC	Debezium 등으로 변경 로그 수집	구성이 무겁지만 트리거 불필요

• 선택 : 원본 DB에 insert trigger를 두고 trigger 테이블에 적재하는 Outbox 방식 사용

3.2 상태 코드

trigger 테이블의 각 행은 상태 코드로 처리 단계를 관리함.

상태 코드 정의

status	의미
0	처리 전
1	처리 중 (선점)
2	완료
9	실패

3.3 선점 방식

가져올 때 status를 1로 바꾸면서 선점하는 방식임.

[1. 선점]     UPDATE: status '0' → '1' (+ batch_id 부여)
[2. 조회]     SELECT ... WHERE status='1' AND batch_id=?
[3. 변환/전송] ScriptedTransformRecord(uuid) → InvokeHTTP
[4-a. 성공]   UPDATE: status '1' → '2'
[4-b. 실패]   UPDATE: status '1' → '9'

• 조회와 선점 분리 금지 : SELECT 후 UPDATE를 따로 하면 사이에 다른 처리가 끼어들 수 있음. UPDATE를 먼저 하고 선점한 행만 다시 SELECT하는 순서가 안전함
• batch_id 컬럼 : UPDATE 시 NiFi가 만든 ${uuid} 같은 값을 함께 기록하면 내가 선점한 행만 정확히 조회 가능함. 없으면 status='1'인 행이 다른 실행분과 섞임
• DB별 단축 방법 : PostgreSQL은 UPDATE ... RETURNING, Oracle은 RETURNING INTO로 선점과 조회를 한 번에 처리함. MySQL은 지원하지 않아 2단계로 진행함
• 건수 제한 : ORDER BY PK LIMIT N으로 배치 크기를 제한해 한 번에 과도한 양을 잡지 않음

3.4 '1'로 굳은 행 복구

NiFi 재시작이나 장애로 status가 1인 채 멈춘 행은 아무도 다시 가져가지 않음.

• updated_at 컬럼 : 선점 시각을 기록함
• 복구 쿼리 주기 실행 : status='1' AND updated_at < NOW() - 10분인 행을 복귀시키는 UPDATE를 별도 스케줄로 실행함
• 복귀 대상 구분 : 재처리 중이던 행(retry_cnt > 0)은 9로, 최초 처리 중이던 행은 0으로 되돌림. 구분하지 않으면 재처리 건이 메인 흐름에 섞임

선점 방식은 batch_id와 복구 쿼리가 함께 있어야 완성된다

---

4. UUID 부여 : ScriptedTransformRecord

행마다 고유한 uuid를 붙이는 단계이며 Language는 Groovy 사용

record.setValue('uuid', UUID.randomUUID().toString())
return record

• Writer 스키마에 uuid 필드 필요 : Writer가 입력 스키마를 그대로 상속하면 uuid 필드가 출력에서 빠짐. uuid를 포함한 스키마를 명시해야 함
• 재시도 시 중복 : 랜덤 UUID는 재처리될 때마다 새 값이 붙어 같은 데이터가 다른 UUID로 중복 전송될 수 있음
• 결정적 UUID : 원본 PK가 있다면 UUID.nameUUIDFromBytes(pk.getBytes())처럼 PK 기반 UUID를 쓰면 몇 번을 재처리해도 같은 값이 나와 멱등성 확보 가능함

DB 유래 데이터는 재처리가 전제이므로 랜덤보다 PK 기반 결정적 UUID가 유리하다

---

5. 실패 처리와 재처리

실패는 9로 분리하고 재처리 플로우를 별도로 운영하면서 재처리 횟수를 3번까지 관리함

[메인 플로우]
0 → 1 (선점) → 전송 성공: 2
                전송 실패: 9 (retry_cnt는 0 유지)

[재처리 플로우 : 별도 스케줄]
9 AND retry_cnt < 3 → 1 (선점 + retry_cnt + 1) → 성공: 2 / 실패: 9

• retry_cnt 컬럼 : 기본값 0. 재처리 플로우가 선점할 때 retry_cnt = retry_cnt + 1을 함께 올리면 중간에 NiFi가 멈춰도 시도 횟수가 누락되지 않음
• 재처리 선점 조건 : WHERE status='9' AND retry_cnt < 3. 선점 시 status를 1로 바꾸고 batch_id를 새로 부여하는 방식은 메인 플로우와 동일함
• 최종 실패 : retry_cnt가 3인 상태에서 다시 실패하면 status=9, retry_cnt=3으로 남음. 재처리 조건에서 빠지므로 자동 제외되며, status='9' AND retry_cnt >= 3으로 확인과 알림 대상을 조회함
• 재처리 주기 : 실패 직후 바로 재시도하면 같은 장애에 연속으로 걸리므로 5~10분 간격으로 스케줄링함
• 시도 횟수 기준 : retry_cnt는 재처리 횟수만 세므로 최초 시도를 포함하면 총 4번 시도됨

5.1 at-least-once와 중복 처리

• 중복 발생 : 복구와 재처리 과정에서 같은 데이터가 수신 측에 두 번 이상 전달될 수 있음
• 수신 측 대응 : 결정적 UUID를 기준으로 수신 서버에서 중복을 걸러야 함

메인 플로우는 실패 시 9로 넘기기만 하고, 재처리 플로우가 선점 시점에 retry_cnt를 올리며 횟수 한도를 관리하는 구성이다

---

요약

구성 요소별 정리

구분	내용
수동 입력	ListenHTTP로 JSON 수신
DB 저장분	insert trigger → trigger 테이블 적재 → NiFi 조회 (Outbox 방식)
상태 코드	0 처리 전 / 1 처리 중 / 2 완료 / 9 실패
선점	UPDATE 먼저 후 batch_id로 선점 행만 SELECT
고착 복구	updated_at 기준 일정 시간 지난 '1' 행을 주기적으로 복귀
UUID	ScriptedTransformRecord로 부여, DB 유래는 PK 기반 결정적 UUID
재처리	별도 플로우, retry_cnt 3회 한도, 초과 시 9로 고정
중복 대응	at-least-once 전제, UUID로 수신 측에서 걸러냄

두 입력 경로를 공통 흐름으로 합치고, 상태 코드와 retry_cnt로 선점·실패·재처리를 관리하며, 결정적 UUID로 중복을 흡수하는 것이 이 설계의 골자이다