PostgreSQL 기반 실시간 데이터 동기화: 논리적 디코딩과 복제 권한 구성
Flink CDC를 사용하여 PostgreSQL의 데이터 변경 사항을 실시간으로 스트리밍하려면, 데이터베이스 측에서 적절한 복제 설정이 필요합니다. 특히 **논리적 디코딩(Logical Decoding)** 및 **복제 슬롯(Replication Slot)** 기능이 활성화되어야 하며, 접근 가능한 전용 사용자 계정도 구성되어야 합니다.
1. PostgreSQL 서버 설정 조정 (postgresql.conf)
Flink CDC가 작동하기 위해선 WAL(Write-Ahead Logging) 레벨이
logical로 설정되어야 하며, 관련 파라미터들을 조정해야 합니다. 설정 파일은 일반적으로 설치 경로의 데이터 디렉터리 내에 있으며, 정확한 위치는 다음 쿼리로 확인 가능합니다:
SHOW config_file;
다음은 수정이 필요한 주요 설정 항목들입니다:
wal_level = logical
max_replication_slots = 25
max_wal_senders = 25
wal_sender_timeout = 180s
shared_preload_libraries = 'pgoutput'
- wal_level = logical: 논리적 디코딩을 위해 필수입니다.
- max_replication_slots: Flink CDC는 각 테이블 또는 소스 그룹마다 하나의 슬롯을 사용할 수 있으므로, 동기화할 테이블 수에 따라 여유 있게 설정하세요.
- max_wal_senders: 활성화된 복제 연결 수 제한이며,
max_replication_slots와 일치시키는 것이 좋습니다.
- wal_sender_timeout: 네트워크 지연 시 연결이 끊기는 것을 방지하기 위해 기본값보다 늘리는 것이 안정성에 유리합니다.
- shared_preload_libraries: PostgreSQL 10 이상에서는 기본 제공되는
pgoutput 플러그인을 사용합니다. 별도 설치 없이 사용 가능합니다.
설정 후에는 PostgreSQL 인스턴스를 재시작하여 반영해야 합니다:
sudo systemctl restart postgresql
Docker 환경에서는 컨테이너를 재시작하세요:
docker restart your-postgres-container
2. 전용 사용자 생성 및 권한 부여
보안상의 이유로, Flink CDC 전용 사용자를 생성하고 최소한의 필요한 권한만 부여하는 것이 바람직합니다.
먼저 슈퍼유저로 PostgreSQL CLI에 접속합니다:
psql -U postgres -d your_database
다음 SQL 문을 실행하여 사용자와 권한을 설정합니다:
CREATE USER cdc_streamer WITH PASSWORD 'secure_password_123';
-- 복제 권한 부여
ALTER USER cdc_streamer REPLICATION;
-- 특정 DB 접속 허용
GRANT CONNECT ON DATABASE your_database TO cdc_streamer;
-- 스키마 사용 권한
GRANT USAGE ON SCHEMA public TO cdc_streamer;
-- 대상 스키마 내 모든 테이블 조회 권한
GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_streamer;
-- 향후 생성될 테이블에도 자동으로 권한 적용
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO cdc_streamer;
> 참고: 실제 운영 환경에서는
public 대신 비공개 스키마 이름을 사용하고, 테이블 범위를 좁히는 것이 좋습니다.
3. PUBLICATION을 통한 테이블 공개
Flink CDC는 PostgreSQL의
PUBLICATION 기능을 통해 변경 데이터를 추출합니다. 동기화할 테이블들을 포함하는 게시물을 생성해야 합니다.
예시:
-- 스키마 내 모든 테이블을 대상으로 게시물 생성
CREATE PUBLICATION db_changes_pub FOR ALL TABLES IN SCHEMA public;
-- 또는 특정 테이블만 지정
CREATE PUBLICATION user_updates_pub FOR TABLE users, user_logs;
생성된 게시물 확인:
SELECT * FROM pg_publication;
대상 테이블 목록 확인:
SELECT * FROM pg_publication_tables WHERE pubname = 'db_changes_pub';
4. REPLICA IDENTITY 설정 (업데이트/삭제 이벤트 처리용)
기본적으로 PostgreSQL은 UPDATE 및 DELETE 시 변경 전 로우의 전체 값을 기록하지 않습니다. Flink CDC가 변경 전 상태(before image)를 정확히 포착하려면 테이블의
REPLICA IDENTITY를
FULL로 설정해야 합니다.
적용 예시:
ALTER TABLE users REPLICA IDENTITY FULL;
ALTER TABLE orders REPLICA IDENTITY FULL;
현재 설정 확인 방법:
SELECT relname, relreplident
FROM pg_class
WHERE relname IN ('users', 'orders');
결과에서
relreplident 값이
f이면 FULL로 설정된 상태입니다.
5. Spring Boot 애플리케이션 설정
application.yml 또는
application.properties에 Flink CDC 소스 정보를 정의합니다.
YAML 형식 예시:
flink:
cdc:
source:
connector: postgres-cdc
hostname: localhost
port: 5432
database-name: your_database
schema-name: public
table-name: .* # 정규 표현식으로 전체 테이블 포함 가능
username: cdc_streamer
password: secure_password_123
decoding-plugin-name: pgoutput
slot-name: flink_slot_01
주요 옵션 설명:
- table-name: 단일 테이블명 또는 정규식 패턴 가능
- slot-name: 복제 슬롯 이름. 중복되지 않도록 고유하게 설정
- decoding-plugin-name: 반드시
pgoutput로 설정 (기본 제공)
6. Maven 의존성 추가
프로젝트 빌드 파일에 Flink CDC Postgres 커넥터를 포함해야 합니다.
Maven (
pom.xml) 예시:
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-connector-postgres-cdc</artifactId>
<version>2.4.0</version>
</dependency>
최신 안정 버전을 사용하는 것이 성능과 호환성 면에서 권장됩니다.
7. 설정 검증 절차
구성이 완료되면 다음 단계로 정상 동작을 확인합니다.
로그 모니터링:
tail -f /var/log/postgresql/postgresql-*.log
접속 거부, 슬롯 충돌, 디코딩 오류 등의 메시지가 없는지 확인합니다.
Flink 작업 실행 테스트:
애플리케이션을 시작하고, 대상 테이블에 INSERT/UPDATE/DELETE를 수행했을 때 변경 이벤트가 정상적으로 출력되는지 확인합니다.
복제 슬롯 상태 확인:
SELECT slot_name, active, restart_lsn, confirmed_flush_lsn
FROM pg_replication_slots
WHERE slot_name = 'flink_slot_01';
활성 상태(
active = true)이고 LSN 포인터가 정상 갱신되는지 확인하세요.