CUBRID -> Postgres 마이그레이션 후 NIFI 변경점

NiFi로 돌리던 트리거 기반 파이프라인 DB를 CUBRID에서 PostgreSQL로 바꾸게 돼서, 작업 전에 손봐야 할 부분을 쭉 적어둠.

---

NiFi 연결 설정

DBCPConnectionPool에서 바꿀 것.

[Connection URL] - jdbc:postgresql://<호스트>:5432/<DB이름>

[Driver Class Name] - org.postgresql.Driver

[Driver Location] - PostgreSQL JDBC jar 경로. 클러스터라 노드마다 같은 경로에 넣어줘야 함

[Validation Query] - SELECT 1

[Max Total Connections] - 노드 수 × 풀 개수 × 풀당 연결 수가 PostgreSQL max_connections(기본 100)를 넘으면 안 됨

---

NiFi 프로세서

[ExecuteSQL / ExecuteSQLRecord] - Set Auto Commit을 false로, Fetch Size도 지정 (1000 정도). PostgreSQL 드라이버는 auto commit이 켜져 있으면 Fetch Size를 줘도 결과를 한 번에 다 메모리에 올림. 선점 건수 많을 때 NiFi 메모리 터질 수 있음

[PutDatabaseRecord] - Database Type을 PostgreSQL로, Schema Name은 소문자로 (NCSC_INTEG → ncsc_integ)

[PutSQL] - sql.args.N.type을 실제 컬럼 타입이랑 맞추기. 문자열/숫자 안 맞으면 바로 에러

[프로세서에 직접 적어둔 SQL] - 선점 UPDATE, 원본 JOIN 조회, 실패 처리(상태 9, 재처리 +1), 완료 처리, 재처리 대상 조회. 아래 SQL 문법 기준으로 하나씩 봐야 함

[ScriptedTransformRecord] - record.getValue("msg_incl_url")에서 컬럼 이름 대소문자가 달라지면 에러 없이 null이 나오고 암호화가 그냥 건너뛰어짐. 실패 큐에서 실제 값 찍어보기

[NUMERIC, timestamp 컬럼] - 레코드로 읽을 때 타입 변환 문제 생길 수 있음. Use Avro Logical Types, Default Decimal Precision/Scale 설정하고 실제로 흘려보기

[에러 메시지 비교 로직] - executesql.error.message를 문자열로 비교하는 곳이 있으면 문구가 달라지니 수정

[선점 쿼리] - UPDATE ... RETURNING이나 SELECT ... FOR UPDATE SKIP LOCKED로 한 문장에 처리 가능. 여러 노드가 동시에 가져가도 겹치지 않아서 Primary Node 제약을 풀 수도 있음

PutSFTP, PublishKafka, UpdateAttribute, AttributesToJSON, RouteOnAttribute, JsonRecordSetWriter는 DB랑 상관없어서 그대로 둠.

---

SQL 문법

[현재 시각] - SYSDATE, SYS_DATETIME, SYSDATETIME → now(), current_timestamp

[오늘 날짜] - TRUNC(SYSDATE) → current_date

[날짜 계산] - 날짜에서 숫자 빼던 거 → now() - interval '5 minutes'

[SUBSTRING_INDEX] - PostgreSQL엔 없음. defang SQL에서 쓰고 있어서 split_part나 regexp_replace로 바꿔야 함. REPLACE, LENGTH, TRIM, SUBSTRING은 그대로 써도 됨

[컬럼 변경] - ALTER TABLE t MODIFY col ... → ALTER TABLE t ALTER COLUMN col TYPE ...

[시스템 카탈로그] - db_attribute, db_class, db_trigger → information_schema.columns, tables, triggers

[타입 비교] - job_prcs_stts_cd가 VARCHAR면 = 1은 에러, = '1'로 써야 함

[식별자] - [이름] 대괄호는 문법 에러. 따옴표 없이 쓰면 전부 소문자로 바뀌고, 큰따옴표로 감싸면 대소문자를 구분함. "TB_DCLR_RCPT"랑 tb_dclr_rcpt는 다른 테이블. 그냥 테이블을 소문자로 만들고 쿼리에서 따옴표 안 쓰는 걸로 통일. 기존 쿼리에 NCSC_INTEG."tb_dclr_rcpt" 같은 거 남아 있는지 확인

[시간대] - now()는 세션 TimeZone 기준. 로그 처리일시나 재시도 간격 계산이 CUBRID 때랑 같은 시각으로 찍히는지 봐야 함

---

테이블 정의

[AUTO_INCREMENT] - GENERATED ALWAYS AS IDENTITY 또는 BIGSERIAL

[DATETIME] - TIMESTAMP

[DEFAULT SYS_DATETIME] - DEFAULT now()

[STRING] - TEXT

[스키마] - CUBRID는 사용자가 곧 스키마인데 PostgreSQL은 CREATE SCHEMA로 따로 만들고 GRANT 주고, 필요하면 search_path까지

[시퀀스] - 기존 데이터를 PK 값째로 옮겼으면 setval로 시퀀스를 최댓값 뒤로 맞춰야 함. 안 하면 다음 INSERT에서 PK 중복

---

트리거

이건 바꾼다기보다 새로 짜야 함. CUBRID는 트리거 안에 SQL을 바로 적는데, PostgreSQL은 함수를 먼저 만들고 트리거가 그 함수를 부르는 구조.

CUBRID

CREATE TRIGGER trig_src_insert
AFTER INSERT ON src_tbl
EXECUTE INSERT INTO trg_tbl (col_a, stts_cd, reg_dt)
        VALUES (obj.col_a, '0', SYS_DATETIME);

PostgreSQL

CREATE OR REPLACE FUNCTION fn_src_insert() RETURNS trigger AS $$
BEGIN
    INSERT INTO trg_tbl (col_a, stts_cd, reg_dt)
    VALUES (NEW.col_a, '0', now());
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trig_src_insert
AFTER INSERT ON src_tbl
FOR EACH ROW EXECUTE FUNCTION fn_src_insert();

(이름은 설명용으로 지어낸 것)

[문장 수] - CUBRID는 한 문장만 가능해서 여러 개면 프로시저를 CALL 해야 했는데, PostgreSQL은 함수 안에 IF, 변수 다 됨

[행 값 참조] - obj → NEW 또는 OLD. AFTER INSERT의 obj는 NEW, DELETE의 obj는 OLD라서 트리거마다 보고 바꿔야 함

[조건] - WHEN (NEW.col_a IS NOT NULL)으로 트리거에 걸거나 함수 안에서 IF

[행 단위 / 문장 단위] - FOR EACH ROW / FOR EACH STATEMENT로 직접 적어야 함

[이벤트 여러 개] - 이벤트별로 트리거를 따로 만들 필요 없이 AFTER INSERT OR UPDATE OR DELETE로 묶고 TG_OP로 구분 가능

[반환값] - BEFORE 트리거는 RETURN NEW 해야 행이 저장됨(NULL이면 취소). AFTER는 반환값 무시라 RETURN NULL

[트리거 실패] - 원래 DML도 같이 실패하는 건 둘 다 같음

[버전] - EXECUTE FUNCTION은 11 이상. 그 아래면 EXECUTE PROCEDURE

---

Python 적재 스크립트

[드라이버] - pycubrid → psycopg2 (또는 psycopg 3)

[자리표시자] - ? → %s

[rollback] - PostgreSQL은 트랜잭션 안에서 에러 한 번 나면 rollback 하기 전까지 그 연결의 쿼리가 전부 실패함. except 잡고 계속 도는 반복문이면 conn.rollback() 꼭 넣기

[중복 체크] - existing_keys로 거르던 걸 INSERT ... ON CONFLICT (mng_no, rcpt_no) DO NOTHING으로

[대량 적재] - executemany 대신 execute_values나 COPY

---

운영 도구

[CLI] - csql → psql

[자동 커밋 끄기] - ;autocommit off → \set AUTOCOMMIT off

[대량 적재] - loaddb → COPY

[백업/복원] - backupdb / restoredb → pg_dump / pg_restore. 특정 시점 복원이 필요하면 WAL 아카이브를 미리 설정해둬야 함

[DBeaver] - 연결 새로 만들기. Manual Commit 전환은 똑같이 Database > Transaction Mode

---

전환 순서

데이터가 계속 들어오는 상태에서 바꿔야 해서 순서가 중요함.

1. 원본 테이블에 새 데이터 안 들어오게 막기
2. 트리거 테이블에 남은 상태 0 / 처리 중 / 재처리 대기 건이 다 빠질 때까지 대기
3. NiFi 플로우 정지
4. 데이터 이관, 트리거 재작성, NiFi 연결 변경
5. 테이블 하나로 트리거 → 선점 → 변환 → 전송까지 끝까지 테스트
6. 나머지 테이블 하나씩 전환

처리 중이던 건이 CUBRID 쪽에 남은 채로 넘어가면 그 건은 빠지거나 두 번 나감. 그래서 2번에서 트리거 테이블이 완전히 빌 때까지 기다리는 게 핵심.

[문자셋] - CUBRID DB가 UTF-8이 아니었다면(EUC-KR 등) 한글 이관할 때 변환이 필요함. PostgreSQL 쪽은 UTF8로 생성