List
데이터 엔지니어링 작업에서 단순히 데이터만 옮기는 작업을 넘어, 사용하는 라이브러리나 툴의 내부 동작 방식을 정확히 이해하는 것은 매우 중요합니다. 이번 글에서는 DB 트랜잭션 격리 수준(Isolation Level)이 어떤 문제를 일으킬 수 있는지 자세히 살펴보고, 이를 토대로 데이터 엔지니어링 시 고려해야 할 사항까지 정리해 보겠습니다.
문제 상황: Kafka Consumer를 통한 DB 처리 이슈
예전에 진행한 데이터 엔지니어링 작업 중, Kafka Consumer를 활용하여 실시간으로 DB 데이터를 처리해야 하는 요구사항이 있었습니다.
로컬 환경에서는 잘 작동하던 코드가 Kubernetes 환경에서는 Consumer가 최초 DB 연결 시점 이후에 추가된 데이터를 전혀 인지하지 못하는 현상이 발생했습니다.
Consumer 로직은 다음과 같이 구성되었습니다.
# 모든 DB connection을 미리 선언하여 서비스 컨테이너에 주입
rcc_path_di = RccPathContainer(config=container_config).rcc_path_di()
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
elif msg.error():
raise KafkaException(msg.error())
else:
# 메시지와 미리 선언된 DB connection으로 DB SELECT, INSERT 처리
handle_kafka_message(msg, rcc_path_di)
Python
복사
로컬 환경에서는 이 방식으로 문제없이 작동했으나, 실제 Kubernetes pod 환경에서는 pod가 생성된 시점 이후에 새롭게 추가된 데이터를 처리하지 못하는 이슈가 발생했습니다.
원인 분석: DB 트랜잭션의 격리 수준(Isolation Level)
문제의 원인은 바로 DB 트랜잭션의 격리 수준(Isolation Level)에 있었습니다.
격리 수준은 동시에 실행되는 여러 트랜잭션 사이의 데이터 가시성을 결정하는 중요한 개념입니다. 대표적으로 다음 네 가지 수준이 존재합니다.
1. READ UNCOMMITTED
•
커밋되지 않은 데이터도 읽을 수 있는 수준
•
흔히 Dirty Read가 발생합니다.
예시 상황
트랜잭션 A | 트랜잭션 B |
a를 10 → 20으로 변경 (커밋하지 않음) | |
a를 조회 → 20으로 보임 | |
a 변경사항 롤백 | 이미 20을 읽어갔으므로 처리에 혼란 발생 |
2. READ COMMITTED
•
다른 트랜잭션에서 커밋된 데이터만 읽습니다. (대부분 DB 기본 설정)
예시 상황
트랜잭션 A | 트랜잭션 B |
a를 10 → 20으로 변경 (커밋하지 않음) | |
a를 조회 → 10으로 보임 | |
a 변경사항 커밋 | 이후 a를 조회하면 20으로 보임 |
3. REPEATABLE READ
•
트랜잭션이 시작된 시점 이후 다른 트랜잭션이 데이터를 변경하더라도 최초 SELECT 결과가 계속 유지됩니다.
예시 상황
트랜잭션 A | 트랜잭션 B |
트랜잭션 시작 → a 조회: 10 | |
a를 10 → 20으로 변경하고 커밋 | |
다시 a를 조회해도 여전히 10으로 보임 |
4. SERIALIZABLE
•
트랜잭션을 순차적으로만 처리하여 완벽한 격리를 제공합니다.
•
동시성이 제한되어 성능에 부정적 영향을 미칠 수 있습니다.
◦
하나의 트랜잭션이 끝나기 전에 다른 트랜잭션이 같은 데이터에 접근하려고 하면, 접근 자체가 허용되지 않고 대기 상태로 들어감
트랜잭션 A | 트랜잭션 B |
트랜잭션 시작, a 조회: 10 | |
트랜잭션 시작, a 조회 시도 (대기 상태) | |
a를 10 → 20으로 변경하고 커밋 | (아직 대기 상태) |
트랜잭션 종료 | 트랜잭션 A 종료 후 a 조회: 20 |
실제 서비스의 격리 수준 확인
서비스에서 실제 사용 중인 격리 수준을 확인해봤습니다.
-- MySQL
SHOW VARIABLES LIKE 'transaction_isolation';
-- PostgreSQL
SHOW TRANSACTION ISOLATION LEVEL;
SQL
복사
•
DEV 환경: REPEATABLE READ
•
STAGE, PROD, DATA-MART 환경: READ COMMITTED
즉, 개발 환경(DEV)의 REPEATABLE READ 설정으로 인해 최초 DB 커넥션 시점 이후 추가된 데이터가 보이지 않았던 것입니다.
pymysql과 psycopg2의 트랜잭션 처리 방식
더 중요한 점은 사용하는 라이브러리의 내부 작동 방식입니다.
•
pymysql, psycopg2의 경우, DB connection을 생성할 때 별도로 설정하지 않으면 기본적으로 autocommit=False입니다.
•
autocommit=False 설정하에서는 별도의 트랜잭션 명령어 (BEGIN)를 명시하지 않아도, 처음 커서를 생성하고 execute를 호출하는 순간부터 자동으로 트랜잭션이 시작됩니다.
•
한 번 시작된 트랜잭션은 명시적으로 커밋(commit)하거나 롤백(rollback)하지 않으면 지속됩니다. 따라서 이후의 모든 SELECT는 이 트랜잭션의 시작 시점을 기준으로 데이터를 바라보게 됩니다.
다시 말해, 최초 SELECT 이후 모든 데이터 조회는 해당 트랜잭션의 최초 시작 시점을 기준으로만 이루어지므로, 이후에 추가된 데이터를 조회하지 못하게 되는 것입니다.
예를 들어, 다음과 같은 코드가 있다면 문제가 발생할 수 있습니다.
# autocommit=False 상태 (기본값)
conn = pymysql.connect(host="db_host", user="user", password="pass", db="db_name")
cursor = conn.cursor()
# 최초 SELECT 시 트랜잭션이 시작됨
cursor.execute("SELECT * FROM users")
# 이후 추가된 데이터는 이 트랜잭션 내에서 볼 수 없음
cursor.execute("SELECT * FROM users WHERE id = NEW_ID")
# 위 쿼리에서 새로 추가된 데이터가 보이지 않음
Python
복사
아래는 psycopg 개발 문서에 있는 내용입니다. psycopg는 기본적으로 트랜잭션 상태에서 동작하며 이를 관리하지 않으면 문제가 발생하기 때문에, 단순히 실행하는 목적이 아니라 consumer와 같이 장기 실행 프로그램에서는 명시적 커밋(commit) 또는 롤백(rollback)을 호출하거나 autocommit 모드를 사용할 필요가 있습니다.
문제의 핵심 정리
이번 문제의 핵심은 두 가지입니다.
•
DB의 격리 수준 (Isolation Level) 설정을 확인하지 않고, 기본적인 세팅이라고 가정하고 작업을 진행함
•
pymysql, psycopg2의 기본 설정(autocommit=False) 으로 인해 트랜잭션이 장기 유지되는 사실을 모르고 해당 라이브러리를 사용
실제로 STAGE, PROD 환경은 READ COMMITTED여서 이 문제가 덜 드러났을 가능성이 큽니다. 그러나 개발 환경인 DEV는 REPEATABLE READ이므로 트랜잭션이 시작된 시점 이후의 데이터 변경을 확인할 수 없는 상황이었습니다.
해결 방법 및 고려사항
이 문제를 해결하는 가장 좋은 방법은 트랜잭션을 쿼리별로 처리하는 방식입니다. 즉, 각 SELECT 또는 INSERT 작업이 개별 트랜잭션을 갖도록 설정하는 것이죠.
간단히 해결하려면, DB connection을 생성할 때 다음과 같이 설정하면 됩니다.
# pymysql의 경우
conn = pymysql.connect(..., autocommit=True)
# psycopg2의 경우
conn = psycopg2.connect(...)
conn.autocommit = True
Python
복사
•
이렇게 설정하면 각 쿼리 실행 후 자동으로 트랜잭션이 커밋됩니다.
•
결과적으로 실시간으로 추가된 데이터도 정상적으로 인식하고 처리할 수 있습니다.
또한, 단순 connection을 생성하는 것 뿐만 아니라 실패시 트랜잭션 관리도 중요합니다.
DB 트랜잭션을 처리하다 보면 다음과 같은 에러가 자주 발생할 수 있습니다.
InFailedSqlTransaction: current transaction is aborted, commands ignored until end of transaction block
Plain Text
복사
이 에러 메시지는 이전에 수행한 DB 작업이 실패했으므로, 현재 세션(트랜잭션)이 오류 상태에 있다는 뜻입니다. 이 상태에서는 추가적인 DB 명령어를 실행해도 모두 실패하게 됩니다.
예를 들어, 다음과 같은 코드에서 문제가 발생할 수 있습니다.
import psycopg2
conn = psycopg2.connect(...)
cursor = conn.cursor()
try:
cursor.execute("INSERT INTO table_a (id, name) VALUES (1, 'A')")
cursor.execute("INSERT INTO table_b (id, age) VALUES (1, 'invalid_age')") # 에러 발생
conn.commit()
except Exception as e:
print(f"에러 발생: {e}")
# 이 상태에서 rollback을 하지 않고 다음 쿼리를 실행하면 계속 실패함
Python
복사
위 코드에서 두 번째 쿼리가 실패하면 그 이후의 모든 DB 작업이 위의 에러 메시지를 출력하면서 실패합니다. 즉, 명시적으로 rollback을 호출해줘야만 이 상태에서 벗어날 수 있습니다.
올바른 처리는 다음과 같습니다.
import psycopg2
conn = psycopg2.connect(...)
cursor = conn.cursor()
try:
cursor.execute("INSERT INTO table_a (id, name) VALUES (1, 'A')")
cursor.execute("INSERT INTO table_b (id, age) VALUES (1, 'invalid_age')") # 에러 발생
conn.commit()
except Exception as e:
print(f"에러 발생: {e}")
conn.rollback() # 반드시 rollback 호출
Python
복사
이렇게 rollback을 명시적으로 수행하면, 에러 상태에서 빠져나와 다음 DB 작업을 문제없이 진행할 수 있게 됩니다.
데이터 엔지니어링에서 라이브러리와 툴의 이해가 중요한 이유
데이터 엔지니어링은 단순히 데이터를 이관하거나 옮기는 작업을 넘어 사용하는 라이브러리나 툴의 내부 작동 방식에 대한 깊은 이해가 필수적입니다. 이번 사례에서도, 라이브러리의 기본 설정이 어떤 트랜잭션 방식을 사용하는지, DB의 격리 수준이 어떻게 동작하는지 정확히 이해하지 않으면 데이터를 정상적으로 처리하기 어려웠습니다.
특히 데이터베이스와 연결하여 작업하는 경우에는 DB Connection의 트랜잭션 특성과 격리 수준을 미리 명확히 파악하고 대응하는 것이 매우 중요합니다. 이는 운영 환경에서 발생할 수 있는 문제를 미리 예방하고, 요구사항에 정확히 부합하는 견고한 데이터 처리 파이프라인을 구성하는 데 필수적인 요소입니다.
마무리
지금까지 Kafka Consumer를 통해 실시간으로 DB 데이터를 처리하면서 발생한 트랜잭션 격리 수준 이슈와 라이브러리의 내부 동작 방식을 정확히 이해해야 하는 이유에 대해 살펴봤습니다. 단순한 문제 해결을 넘어서, 데이터 엔지니어링을 하는 과정에서 라이브러리와 툴의 내부 메커니즘을 깊이 이해하는 중요성에 대해 다시 한번 생각해볼 기회가 되었으면 좋겠습니다.