※ 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")
오늘은 여기까지입니다.
감사합니다 :)
'PostgreSQL' 카테고리의 다른 글
PostgreSQL: 락 체크 쿼리 (0) | 2025.01.07 |
---|---|
PostgreSQL: PG ↔ Oracle 이관 시 메타데이터 정합성 체크 쿼리 (1) | 2025.01.05 |
PostgreSQL: Patroni로 고가용성(HA) 구성 방법 (0) | 2024.12.31 |
PostgreSQL: timeout 관련 파라미터 (1) | 2024.12.28 |
PostgreSQL: repmgr로 고가용성(HA) 구성 방법 (0) | 2024.12.19 |