PostgreSQL

PostgreSQL: PG ↔ Oracle 이관 시 데이터 정합성 체크 방법

dewstream 2025. 1. 4. 09:00

※ Data integrity check methods during migration from PostgreSQL ↔ Oracle.
 
안녕하세요. 듀스트림입니다.
 
최근 받은 요청 중 하나가 스테이징 DB는 PostgreSQL, 실 운영 DB는 Oracle인데 스테이징 DB를 거쳐서 실 운영 DB에 데이터가 들어왔을 때 그 데이터의 정합성 체크 방법을 알려달란 요청이 있었습니다.
 
요구사항은 총 두 개였습니다.
1. 초기 이관 시 정합성 체크
2. 추가 이관 시 추가 이관한 데이터만 정합성 체크(timestamp 컬럼 존재)
 
이제 어떻게 해야되는지 한 번 살펴볼까요?


정합성 체크의 주 목적은 데이터의 일관성과 정확성, 완전성 체크입니다.

 
※ 메타데이터 체크는 선행되어야 합니다.

 

PostgreSQL: PG to Oracle 이관 시 메타데이터 정합성 체크 SQL

※ Metadata integrity check methods during migration from PostgreSQL to Oracle. PostgreSQL 체크용 쿼리-- 1.스키마별 디스크 사용량SELECT schema_name, pg_size_pretty(sum(table_size)::bigint) as disk_spaceFROM (SELECT pg_catalog.pg_namespace.

dewstream.tistory.com


1. COUNT(*)를 이용한 건 수로 체크하는 방법

두 테이블의 건 수가 같은지를 확인하는 데 목적이 있습니다.
 
▸ PostgreSQL

-- 전체 데이터 체크
SELECT COUNT(*) FROM table_name;

-- 추가 이관 데이터만 체크
SELECT COUNT(*) 
FROM table_name 
WHERE created_at > '2025-01-01 00:00:00';

 
▸ Oracle

-- 전체 데이터 체크
SELECT COUNT(*) FROM table_name;

-- 추가 이관 데이터만 체크
SELECT COUNT(*) 
FROM table_name 
WHERE created_at > TO_DATE('2025-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS');

2. HASH 기반 검증 방법

행의 순서가 다르거나, 정렬 기준이 명확하지 않은 경우 동일한 데이터라도 다른 해시값이 생성될 수 있습니다.

 

▸ PostgreSQL

-- 전체 데이터 체크
SELECT MD5(string_agg(column1::text || column2::text, '' ORDER BY id)) AS hash_value 
FROM table_name;

-- 추가 이관 데이터만 체크
SELECT MD5(string_agg(column1::text || column2::text, '' ORDER BY id)) AS hash_value 
FROM table_name 
WHERE created_at > '2025-01-01 00:00:00';

-- 타겟으로 체크
SELECT id, MD5(column1::text || column2::text) AS hash_value
FROM table_name
WHERE id = 'target_id';

 
▸ Oracle

-- 전체 데이터 체크
SELECT STANDARD_HASH(LISTAGG(column1 || column2, '') WITHIN GROUP (ORDER BY id), 'MD5') AS hash_value
FROM table_name;

-- 추가 이관 데이터만 체크
SELECT STANDARD_HASH(LISTAGG(column1 || column2, '') WITHIN GROUP (ORDER BY id), 'MD5') AS hash_value
FROM table_name
WHERE created_at > TO_DATE('2025-01-01 00:00:00', 'YYYY-MM-DD HH24:MI:SS');

-- 타겟으로 체크
SELECT id, STANDARD_HASH(column1 || column2 || column3, 'MD5') AS row_hash
FROM table_name
WHERE id = 'target_id';

3. Python을 활용한 자동화

▸ 예시: HASH 값 비교 자동화

import pandas as pd
from sqlalchemy import create_engine

# PostgreSQL 접속 정보
pg_conn_info = {
    "host": "your_postgres_host",
    "port": "5432",
    "dbname": "your_database_name",
    "user": "your_username",
    "password": "your_password"
}

# Oracle 접속 정보
oracle_conn_info = {
    "user": "your_username",
    "password": "your_password",
    "dsn": "your_oracle_dsn"
}

# SQLAlchemy 연결 생성
pg_engine = create_engine(f"postgresql+psycopg2://{pg_conn_info['user']}:{pg_conn_info['password']}@{pg_conn_info['host']}:{pg_conn_info['port']}/{pg_conn_info['dbname']}")
oracle_engine = create_engine(f"oracle+cx_oracle://{oracle_conn_info['user']}:{oracle_conn_info['password']}@{oracle_conn_info['dsn']}")

# 사용자 입력
last_checked_time = input("Enter the last checked time (YYYY-MM-DD HH:MM:SS): ")

# PostgreSQL 해시 생성
pg_query = f"""
SELECT MD5(string_agg(column1::text || column2::text, '' ORDER BY id)) AS hash_value 
FROM table_name 
WHERE created_at > '{last_checked_time}'
"""
pg_hash = pd.read_sql(pg_query, pg_engine).iloc[0, 0]

# Oracle 해시 생성
oracle_query = f"""
SELECT STANDARD_HASH(
    LISTAGG(column1 || column2, '') WITHIN GROUP (ORDER BY id), 
    'MD5'
) AS hash_value 
FROM table_name 
WHERE created_at > TO_DATE('{last_checked_time}', 'YYYY-MM-DD HH24:MI:SS')
"""
oracle_hash = pd.read_sql(oracle_query, oracle_engine).iloc[0, 0]

# 해시 비교
if pg_hash == oracle_hash:
    print("Data is consistent")
else:
    print("Data mismatch detected")

오늘은 여기까지입니다.
감사합니다 :)