MySQL → Oracle 대용량 데이터 이관 – 병렬 처리와 예외 처리 전략 (Python 기반)
MySQL → Oracle 대용량 데이터 이관 – 병렬 처리와 예외 처리 전략 (Python 기반)
대용량 데이터를 다른 DBMS로 이관하는 작업은 단순한 insert 반복이 아니다.
특히 MySQL에서 Oracle로 수백만 건의 데이터를 이관하면서, 성능, 데이터 정합성, 중복 처리, 재처리 가능성까지 고려해야 한다.
이 글은 실무에서 경험한 Python 기반 병렬 이관 전략을 공유한다.
이관 요구사항 요약
- 출발지: MySQL
- 도착지: Oracle
- 데이터량: 약 300만 건
- 요구조건:
- 1분 이내 처리
- 중복 무시
- 장애 발생 시 특정 구간만 재처리 가능
- Oracle의 제약조건(UNIQUE KEY) 충돌 방지
기술 스택
- Python 3.11+
pymysql: MySQL 데이터 조회용cx_Oracle: Oracle 데이터 입력용concurrent.futures: 병렬 처리logging: 로깅 및 예외 추적
병렬 처리 전략 설계
1. 범위 기반 분할
전체 ID 범위를 균등하게 나눈 후 Thread 단위로 병렬 처리했다.
예:
Thread 1 → ID 1 ~ 1,000,000
Thread 2 → ID 1,000,001 ~ 2,000,000
Thread 3 → ID 2,000,001 ~ 3,000,000- 한 구간이 실패해도 전체 작업에 영향 없이 해당 범위만 재실행 가능
- 파라미터로 구간만 바꾸면 되므로 재사용성도 높음
2. 단건 INSERT + 예외 필터
- 대량의 executemany() 대신 1건 단위 insert + 예외 처리 방식을 채택
- Oracle에서 중복 제약(ORA-00001)을 걸러내기 위해 try-catch로 감쌈
이관 스크립트 (Python 예시)
import pymysql
import cx_Oracle
from concurrent.futures import ThreadPoolExecutor
import logging
# 설정 정보
mysql_config = {...}
oracle_config = {...}
logging.basicConfig(level=logging.INFO)
def fetch_data(start_id, end_id):
with pymysql.connect(**mysql_config) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT USER_CODE, TOKEN, APP_ID, OS_TYPE, CREATED_AT
FROM source_table
WHERE ID BETWEEN %s AND %s
""", (start_id, end_id))
return cur.fetchall()
def insert_data_to_oracle(data):
with cx_Oracle.connect(**oracle_config) as conn:
with conn.cursor() as cur:
for row in data:
try:
cur.execute("""
INSERT INTO target_table (
USER_CODE, TOKEN, APP_ID, OS_TYPE, CREATED_AT
) VALUES (
:1, :2, :3, :4, TO_TIMESTAMP(:5, 'YYYY-MM-DD HH24:MI:SS.FF')
)
""", row)
except cx_Oracle.IntegrityError as e:
if 'ORA-00001' in str(e):
logging.debug("중복 건 무시: %s", row)
continue
else:
raise
conn.commit()
def migrate_range(start_id, end_id):
logging.info("Start range: %s - %s", start_id, end_id)
data = fetch_data(start_id, end_id)
logging.info("Fetched %d rows", len(data))
insert_data_to_oracle(data)
logging.info("Finished range: %s - %s", start_id, end_id)
# 병렬 처리 실행
ranges = [
(1, 1000000),
(1000001, 2000000),
(2000001, 3000000)
]
with ThreadPoolExecutor(max_workers=3) as executor:
for start_id, end_id in ranges:
executor.submit(migrate_range, start_id, end_id)안정성과 회복력 확보 전략
✅ 중복 예외 무시 (ORA-00001)
- Oracle의 UNIQUE 제약조건 위반은 ORA-00001로 발생
- 이 예외만 필터링해 무시함으로써 중복 insert 오류를 무해화
✅ 재처리 가능성 확보
- 구간을 분리하여 병렬 처리하면,
- 실패한 쓰레드만 범위 기준으로 재실행 가능
- 전체 작업을 다시 시작할 필요 없음
- 예외 발생 시 로그로 기록하여 정확한 구간 파악 가능
✅ 로깅 및 추적
INFO Start range: 1 - 1000000
INFO Fetched 999992 rows
INFO Finished range: 1 - 1000000- 로깅 레벨: INFO (진행 상황), DEBUG (중복 건)
- 구간별 처리 현황 로그를 통해 디버깅 및 검증 가능
성능 최적화 팁
- INSERT 최적화 : 단건 insert이지만 커넥션과 커서 유지 + 중복 무시로 안정성 확보
- 날짜 포맷 : TO_TIMESTAMP()를 SQL 내부에서 사용하여 성능 유지
- Batch 크기 조정 : 경우에 따라 fetch 단위를 줄이거나 구간 수를 늘려 조절 가능
- Thread 수 : CPU 코어 수 + DB 커넥션 수를 고려해 병렬 수 설정
결과 요약
- 전체 이관 시간: 약 55초
- 중복 에러: 모두 무시 처리 성공
- 안정성: 각 구간 독립 처리 → 재처리 가능
- 이관 도중 장애 발생 시에도 빠르게 회복 가능
마무리
Python 기반의 병렬 이관은 구현이 단순하면서도 유연하게 확장 가능하다.
특히 Oracle의 제약조건을 고려한 예외 중심 설계, 구간 분할 전략, 로깅 기반 추적은
실무에서 대용량 데이터를 안정적으로 처리하는 데 큰 도움이 되었다.
요약 글은 티스토리 MySQL에서 Oracle로, 대용량 데이터를 안전하게 이관한 방법 (Python 사용기) 에서 확인할 수 있다.
Server Dev