Airflow를 활용해 신용 평가 데이터를 DB에 적재하는 파이프라인을 구축하던 중 발생한 두 가지 주요 에러와 해결 과정을 정리한다.
1. TypeError: string indices must be integers, not 'str'
문제 상황
신용 평가 API 호출 결과를 파싱하여 DB 적재 함수(_load_users_credit)로 전달하는 과정에서 발생했다. data['user_id']와 같이 딕셔너리 키로 접근하려 했으나 에러가 발생하며 태스크가 실패했다.
원인 분석
에러 메시지는 data 변수가 딕셔너리가 아닌 문자열(String)임을 가리키고 있었다.
- 가정: API 응답이나 이전 태스크에서 넘어온 값이 JSON 객체(dict)일 것이라 판단했다.
- 팩트: 실제 데이터는 역직렬화(Deserialization)되지 않은 Raw String 상태였다. Python에서 문자열에 대괄호(
[])로 접근하면 인덱스(정수)를 기대하지만, 문자열 키를 넣었기 때문에 발생한 문법 오류였다.
해결 방법
데이터를 사용하는 시점에서 json.loads()를 통해 딕셔너리로 변환하거나, 데이터를 넘겨주는 시점에서 파이프라인의 데이터 직렬화 상태를 점검하여 객체 타입으로 전달되도록 수정했다.
2. MySQLdb.IntegrityError: Duplicate entry for key PRIMARY
문제 상황
데이터 타입 이슈 해결 후, 실제 DB에 데이터를 INSERT 하는 과정에서 발생했다. 특정 유저 아이디(C001)가 이미 테이블의 기본키(Primary Key)로 존재하여 중복 삽입이 불가능하다는 에러였다.
원인 분석
- 설계 결함: 해당 DAG는 주기적으로 신용 점수를 업데이트해야 한다. 하지만 초기 코드는 단순
INSERT쿼리로 작성되어 있어, 이미 데이터가 존재하는 유저의 경우 제약 조건 위반으로 처리가 중단되었다. - 테이블 구조:
customers테이블의 PK가user_id로 설정되어 있어, 동일 유저에 대한 다중 레코드 삽입이 차단된 상태였다.
3. ON DUPLICATE KEY UPDATE를 활용한 해결
단순히 에러를 회피하는 것이 아니라, 비즈니스 로직에 맞게 Upsert(Update + Insert) 방식을 채택하기로 했다.
수정된 SQL 로직
INSERT INTO customers (user_id, credit_score, grade)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE
credit_score = VALUES(credit_score),
grade = VALUES(grade);결과 및 기대 효과
1. 멱등성(Idempotency) 확보: 동일한 DAG를 여러 번 실행해도 에러 없이 최신 데이터로 유지된다.
2. 효율성: 데이터 존재 여부를 먼저 조회(SELECT)하고 조건문으로 분기할 필요 없이, DB 레벨에서 한 번의 쿼리로 삽입과 갱신을 동시에 처리한다.
4. 마치며
이번 이슈를 통해 두 가지 교훈을 얻었다.
첫째, 데이터의 타입을 맹신하지 말고 항상 로그를 통해 객체의 형상을 확인해야 한다.
둘째, DB 적재 로직을 설계할 때는 해당 테이블이 이력(History) 관리용인지, 현재 상태(Master) 관리용인지를 명확히 구분하여 적절한 쿼리 전략을 세워야 한다.
