리멤버 과제 코드 리뷰 대비 — 태스크 동작·실행 순서·공백 정책 실측

작성 2026-08-11. 1차 면접 2026-08-12(수) 직전 정리.

이 문서는 웹 페이지(human/career/remember-interview/assignment.html)에 넣은 내용과 같습니다. GitHub 접속이 막혀 배포가 대기 중이라, 지금 읽을 수 있게 마크다운으로 함께 둡니다.

근거는 제출본 코드(career/클로드이력서/리멤버앤컴퍼니-data-engineer/과제/작업본/)와, 같은 코드를 두 정책으로 각각 실행해 만든 DuckDB 두 개입니다.

1. 태스크 4개는 어떻게 동작하나

네 태스크가 맡는 일은 각각 처리 대상 결정, 원문 확보, 값 판정, 마스터 반영입니다. 앞 태스크가 만든 dict가 XCom으로 넘어가면서 키가 누적되므로, 마지막 태스크는 소스 정보부터 스테이징 집계까지 한 번에 들고 있습니다. DB 커넥션은 태스크마다 열고 닫습니다.

resolve_source — 무엇을 처리할지 결정한다

DB를 전혀 건드리지 않는 유일한 태스크입니다. Airflow가 주입한 data_interval_start에서 연도를 꺼내는데, 회차의 데이터 구간에서 뽑기 때문에 실행 시각과 무관합니다. 몇 달 뒤에 재실행해도 그 회차가 처리하는 연도는 그대로입니다. 연도를 파일명 템플릿에 끼워 경로를 만들고 존재 여부를 확인해서, 없으면 AirflowSkipException으로 건너뜁니다. 오늘 기준으로 2024·2025 회차도 스케줄되기 때문에 이 분기가 없으면 히스토리가 전부 실패로 물듭니다. 마지막으로 SyncPolicy()를 여기서 한 번만 만들어 dict로 직렬화합니다. 정책 필드가 환경변수를 읽는 구조라 태스크마다 새로 만들면 한 실행 안에서 규칙이 갈릴 수 있어, 첫 태스크에서 확정해 흘려보냅니다.

land_raw — 원문을 손대지 않고 확보한다

커넥션을 열고 ddl.sql을 적용합니다. 전부 IF NOT EXISTS라 매번 그냥 다시 돌립니다. 이어서 ops_sync_run에 회차를 RUNNING으로 기록하면서 소스 파일의 SHA-256과 크기, 정책 JSON을 같이 넣습니다. 같은 run_id면 UPSERT로 카운터까지 초기화해서, 재시도가 이전 시도의 숫자와 섞이지 않게 합니다.

적재는 검증부터 합니다. 마스터에 반영된 최신 연도와 비교해 역행이면 파일을 열기도 전에 끊고, DESCRIBE로 CSV 헤더를 컬럼 순서까지 대조합니다. 그다음 임시 테이블에 먼저 받습니다. 전부 문자열로 읽고(all_varchar), 깨진 행을 건너뛰지 않고 (ignore_errors=false), row_number()로 원본 행 번호를 붙입니다. 0건이면 여기서 실패시킵니다. 0건을 통과시키면 뒤에서 마스터 전건이 사라짐 처리되기 때문입니다. 검증을 모두 통과한 다음에야 트랜잭션 안에서 해당 연도를 지우고 새로 넣습니다. 순서가 이래야 실패했을 때 기존 스냅샷이 남습니다. 2022년 30,513행, 2023년 23,261행.

stage_and_validate — 비교 가능한 값으로 만들고 판정한다

raw를 원본 행 순서대로 읽어 행별로 정규화합니다. 이 단계를 SQL 대신 파이썬 순수 함수로 둔 이유는 규칙을 단위 테스트로 직접 검증할 수 있게 하려는 것입니다. 공백처럼 보이는 특수문자와 제어문자를 정리하고, 사업자등록번호는 구분 기호만 떼고 자릿수 보정은 하지 않으며, 1과 1.0이 섞인 근로자수는 소수부가 0일 때만 정수로 받습니다. 판정 결과는 두 플래그로 갈립니다. has_valid_key는 "소스에 존재했다"의 근거, is_valid는 "마스터에 반영해도 된다"의 근거입니다. 하나로 묶으면 값 하나가 깨진 행이 사라진 것으로 처리됩니다.

그다음 같은 사업자등록번호가 여러 번 나온 경우를 마지막 행 채택으로 정리하고 규칙별 건수를 집계합니다. 트랜잭션 안에서 해당 연도 스테이징을 교체하고 격리 행을 다시 쓰는데, 이때 병합 단계가 남긴 DATE_REGRESSION 경고는 지우지 않습니다. 마지막이 품질 게이트입니다. DB를 보지 않고 건수만으로 불량률을 계산해 임계치(기본 1%)를 넘으면 실행 기록을 FAILED로 닫고 AirflowFailException으로 즉시 실패시킵니다. 재시도해도 같은 결과라 재시도가 무의미하기 때문입니다. 두 회차 모두 ERROR는 0건이었습니다.

apply_and_finish — 한 트랜잭션에서 반영하고 검증한다

여기서만 마스터가 바뀝니다. 트랜잭션을 열고 먼저 "무엇을 바꿀지"를 임시 테이블 세 개로 확정합니다. 스테이징과 마스터를 LEFT JOIN해 정책을 적용한 최종값과 변경 전 값을 나란히 담고(merge_candidate), 그중 실제로 값이 달라진 것만 골라내고(merge_changed), 등장한 키를 모읍니다(seen_keys). 그리고 반영 전에 건수를 셉니다. 반영 후에는 신규와 변경을 구분할 수 없고 재등장 건수는 0이 됩니다.

다음 순서도 강제입니다. 변경 이력을 먼저 쓰고 마스터를 갱신합니다. 마스터를 먼저 바꾸면 이전 값을 읽을 수 없습니다. 이력은 컬럼 단위로 남기고 변경 사유("결측이던 값이 채워짐", "소스 값이 비어 있어 NULL로 갱신")까지 적습니다. 반영은 신규 삽입, 등장 키의 생존 상태 갱신, 값이 바뀐 행의 속성 갱신 세 단계입니다. 이어서 이번 소스에 없던 활성 행을 찾아 지우지 않고 비활성으로 표시합니다. 같은 트랜잭션 안에서 정합성 8개를 검사하고, 하나라도 깨지면 마스터와 변경 이력이 함께 롤백됩니다. 통과하면 커밋한 뒤에 로그를 찍고, 카운터 13개와 8개 체크 결과를 JSON으로 ops_sync_run에 마감 기록합니다. 2023년은 신규 2,174 / 변경 4,000 / 무변경 17,087 / 비활성 9,426으로 마스터가 32,687행(활성 23,261)이 됐습니다.

네 태스크가 병렬 구간 없이 한 줄로 이어지는 이유는 데이터 의존성이면서 DuckDB 단일 writer 제약입니다. 그래서 커넥션도 태스크 안에서만 잡고 DAG에 max_active_runs=1을 걸었습니다.

태스크 호출 순서

sequenceDiagram
    autonumber
    participant sch as Airflow scheduler
    participant t1 as resolve_source
    participant t2 as land_raw
    participant t3 as stage_and_validate
    participant t4 as apply_and_finish
    participant duck as DuckDB

    sch->>t1: data_interval_start 주입
    t1->>t1: 연도 계산과 경로 조립
    alt 파일 없음
        t1--x sch: AirflowSkipException 실패가 아닌 SKIP
    else 파일 있음
        t1->>t1: SyncPolicy 생성 후 직렬화
        t1-->>t2: XCom source_year, source_path, policy
    end
    t2->>duck: connect 후 init_schema
    t2->>duck: merge.start_run RUNNING 기록
    t2->>duck: merge.land_raw 역행·헤더 검증 후 연도 교체
    t2-->>t3: XCom + source_row_count
    t3->>duck: merge.stage 정규화와 격리
    t3->>t3: quality.evaluate_gate 불량률 판정
    alt 게이트 미통과
        t3->>duck: merge.finish_run FAILED
        t3--x sch: AirflowFailException 재시도 없음
    end
    t3-->>t4: XCom + staged_valid_count
    t4->>duck: merge.apply_master 한 트랜잭션
    t4->>duck: merge.finish_run SUCCESS + 카운터 13개
    t4->>duck: report 요약 로그 출력

apply_master 트랜잭션 내부

sequenceDiagram
    autonumber
    participant sync as master_sync.apply_master
    participant cand as master_candidates
    participant hist as master_history
    participant mut as master_mutations
    participant rec as reconciliation
    participant duck as DuckDB

    sync->>duck: BEGIN TRANSACTION
    sync->>duck: 연도 역행 재확인, 병합 전 행수
    sync->>cand: build_merge_candidates
    cand->>duck: TEMP merge_candidate, merge_changed, seen_keys
    sync->>cand: collect_candidate_metrics
    Note right of cand: 반영 전에 센다.<br/>반영 후에는 신규와 변경을 구분할 수 없고<br/>재등장은 0이 된다
    sync->>hist: 성립일자 역행 경고, 변경 이력 기록
    Note right of hist: 갱신 전에 쓴다.<br/>갱신 후에는 old_value 를 읽을 수 없다
    sync->>mut: apply_candidate_rows 신규·생존·속성 갱신
    sync->>duck: 사라진 활성 행 비활성 표시
    sync->>rec: reconcile 8개 검사
    alt 실패 체크 있음
        sync--x duck: ROLLBACK
    else 전부 통과
        sync->>duck: COMMIT
    end

2. ./up.sh 실행 순서

검토자 입장에서 명령 하나로 끝나야 하므로, 스크립트가 다섯 단계를 순서대로 진행하고 각 단계마다 대기 조건을 따로 둡니다. 이전 산출물 초기화, 이미지 빌드와 기동, 준비 대기, 파이프라인 완료 대기, 결과 확인과 DB 반출입니다.

여기서 설명할 지점이 하나 있습니다. 스크립트는 DAG를 직접 trigger하지 않습니다. DAGS_ARE_PAUSED_AT_CREATION=false와 catchup=True 조합으로 스케줄러가 2022·2023 회차를 스스로 만들고, 스크립트는 파이프라인이 직접 쓴 ops_sync_run의 상태를 2초 간격으로 폴링합니다. Airflow API 대신 결과 테이블을 보기 때문에, 태스크는 성공했는데 데이터가 들어가지 않은 상황을 구분할 수 있습니다. 폴링 쿼리는 읽기 전용 접속이 실패할 수 있어 (단일 writer가 쓰는 중) duckdb.Error를 잡아 그 경우를 대기로 처리합니다.

준비 대기를 health, DAG 등록, 파이프라인 완료 세 구간으로 나눈 이유는 실패 지점을 구분하려는 것입니다. 웹서버가 못 뜬 것, DAG 파싱이 깨진 것, 파이프라인이 느린 것은 대응이 다릅니다. DAG 등록이 안 되면 dags list-import-errors를 함께 출력합니다. 실패하면 trap이 컨테이너 상태와 최근 로그 120줄을 자동으로 출력합니다. 성공 후에는 서비스를 내리지 않고 남겨 두어 Airflow UI와 DuckDB를 직접 확인할 수 있게 하고, 종료는 ./down.sh로 분리했습니다.

sequenceDiagram
    autonumber
    actor 사용자
    participant up as up.sh 호스트
    participant compose as docker compose
    participant af as airflow 컨테이너
    participant duck as DuckDB 볼륨

    사용자->>up: ./up.sh
    up->>up: docker, compose v2, daemon 점검
    up->>compose: 라벨 조회로 타 디렉터리 중복 실행 차단
    Note over up: 1/5 초기화
    up->>compose: down -v --remove-orphans, warehouse 정리
    Note over up: 2/5 빌드와 기동
    up->>compose: up -d --build
    compose->>af: 이미지 빌드 후 airflow standalone
    Note over up: 3/5 준비 대기
    loop 2초 간격 최대 4분
        up->>af: curl localhost:8080/health
    end
    loop 2초 간격 최대 3분
        up->>af: airflow dags list 에서 DAG 확인
    end
    Note over af: scheduler 가 catchup 으로<br/>2022·2023 회차를 스스로 생성
    Note over up: 4/5 완료 대기
    loop 2초 간격 최대 6분
        up->>af: heredoc 파이썬으로 상태 조회
        af->>duck: read_only 접속, ops_sync_run status
        af-->>up: waiting 또는 success 또는 failed
    end
    Note over up: 5/5 결과 확인
    up->>af: 로그 grep 요약, dags list-runs
    up->>af: python scripts/inspect_db.py
    up->>compose: DuckDB 파일을 호스트 warehouse 로 복사
    up-->>사용자: DB 경로, Airflow URL, 확인 명령 안내

3. 공백 정책 두 경로 — 실제 값

2023년 파일의 빈 값을 최신 상태로 볼지 추출 누락으로 볼지는 확정할 수 없어 정책으로 분리했습니다. 그 판단이 결과에서 얼마나 갈리는지 값으로 확인했습니다. 2026-08-11 로컬에서 두 정책을 각각 실행해 DB 두 개를 만들고 대조했습니다 (scripts/run_pipeline.py --fresh --years 2022 2023, 한쪽만 --preserve-null). 소스와 코드는 같고 플래그 하나만 다릅니다.

결과 차이

행수는 같습니다. 정책은 어떤 사업장이 마스터에 있는지에는 영향을 주지 않고 값만 바꿉니다.

2,249가 156이 되는 이유

2023년 소스에서 주소가 비어 온 사업장이 정확히 2,249건입니다. 이걸 "2022년에 주소가 있었는지"로 쪼개면 이렇게 갈립니다.

보존 정책은 앞의 2,093건에서 기존 값을 지킵니다. 2,249 − 2,093 = 156. 156은 하한선입니다. 지킬 기존 값이 애초에 없는 케이스라 어떤 정책을 쓰든 NULL로 남습니다. 면접에서 두 숫자를 말할 때 이 문장을 붙이면 외운 값이 아니라 이해한 값으로 들립니다.

같은 사업장, 두 정책

저축은행 본점이 남대문로에서 사라졌을 가능성은 낮습니다. 그런데 2023년 파일의 주소 칸은 비어 있습니다. 실제로 비워진 것인지 이번 추출에서 빠진 것인지 판단할 근거가 소스에 없다는 것이 이 문제의 전부입니다.

회차 카운터 (2023)

여기 숫자 하나가 설명 지점입니다. 공백 보존이 2,095건 일어났는데 변경 건수는 1,097건만 줄었습니다. 보존은 컬럼 단위이고 분류는 행 단위이기 때문입니다. 주소만 비어 온 행은 보존하면 바뀔 값이 없어져 변경없음으로 넘어가지만(17,087 + 1,097 = 18,184), 주소가 비면서 근로자수까지 바뀐 행은 여전히 변경으로 남습니다. 2,095는 주소 2,093 + 업종 1 + 성립일자 1이고, 아래 표의 기본 정책 수치와 같은 값입니다.

변경 이력에 남는 것

기본 정책에서는 2,095건 전부가 change_reason = '소스 값이 비어 있어 NULL로 갱신(보존 정책 off)'로 남습니다. 무엇을 지웠는지가 기록에 있어서, 나중에 해석이 뒤집히면 이력만 보고 되짚을 수 있습니다. 보존 정책에서 이 값이 0이어야 한다는 것이 정합성 검사 NO_SILENT_NULL_OVERWRITE이고, 기본 정책에서는 NULL 갱신이 의도된 동작이라 이 검사를 0으로 두고 넘어갑니다.

두 정책 모두 8개 검증 통과

기본(덮어쓰기): 2022년 SUCCESS 8-8 / 2023년 SUCCESS 8-8
보존(preserve): 2022년 SUCCESS 8-8 / 2023년 SUCCESS 8-8

검증이 "마스터 값은 소스 값과 같아야 한다"로 굳어 있으면 보존 정책은 무조건 실패합니다. _policy_mismatch_count가 preserve 플래그를 SQL CASE에 넣어 정책대로 기대값을 다시 계산하고 마스터 실제값과 대조하기 때문에 두 경로가 모두 통과합니다.

30초로 말한다면

2023년 파일에서 주소 칸이 빈 사업장이 2,249건입니다. 그중 2,093건은 2022년에 주소가 있던 곳이고, 저축은행 본점처럼 실제로 이전했다고 보기 어려운 곳도 섞여 있습니다. 명세가 최신 파일 값으로 갱신하라고 했으니 기본은 NULL로 덮어쓰고, 그러면 마스터 주소 결측이 2,249건이 됩니다. 보존 정책을 켜면 2,093건을 지켜서 156건으로 줄어듭니다. 156은 2022년에도 주소가 없던 곳이라 어느 정책이든 NULL로 남는 하한선입니다. 어느 쪽이 맞는지는 원천에 확인할 문제라 코드에서는 플래그로 분리했고, 검증이 정책을 알고 기대값을 계산하기 때문에 두 정책 모두 8개 검증을 통과합니다.

공백을 NULL로 바꾼 이유를 물으면 층을 먼저 나눠야 합니다. 정규화 단계에서 공백만 든 셀을 결측으로 통일한 것과, 병합 정책에서 2023년 빈 값으로 기존 값을 덮어쓴 것은 다른 결정입니다. 앞은 비교를 가능하게 하려는 것이고(전각 공백 U+3000이 사업장명에서만 30건, 빈 문자열을 남기면 표기 차이만으로 UPDATE로 잡힘), 뒤는 위의 명세 해석 문제입니다.

4. 모듈을 왜 이렇게 나눴나

변경 이유가 서로 다른 코드를 한 파일에 두지 않으려고 나눴습니다. 병합이 한 덩어리였을 때는 정책을 한 줄 바꾸는 작업이 이력 기록과 검증 쿼리까지 같이 열게 만들었습니다. 지금은 후보 생성(master_candidates)은 변경 감지 규칙이나 정책이 바뀔 때, 이력(master_history)은 남길 항목이 바뀔 때, 변이(master_mutations)는 마스터 컬럼이 바뀔 때, 검증(reconciliation)은 검증 항목이 늘 때 각각 열립니다.

나눈 뒤 얻은 것이 하나 더 있습니다. 이 파이프라인은 순서가 설계입니다. 건수는 반영 전에 세야 하고, 이력은 갱신 전에 써야 하고, 검증까지 한 트랜잭션이어야 합니다. master_sync는 그 순서만 조립하고 각 단계 구현을 갖고 있지 않아서, 181줄 한 파일을 읽으면 순서가 그대로 보입니다.

트랜잭션 경계도 구조로 고정했습니다. atomic을 여는 곳은 apply_master 하나뿐이고 나머지 모듈은 BEGIN·COMMIT을 모릅니다. 커밋 지점이 여러 곳에 생기는 사고를 파일 경계로 막았습니다. 테스트 표면도 함께 갈라집니다. 정규화·품질은 DB 없이 순수 함수로, 후보 생성은 임시 테이블 결과로, 검증은 숫자만으로 확인합니다. 한 덩어리면 이 전부를 통합 테스트를 통해서만 볼 수 있습니다.

merge.py는 로직 없는 공개 파사드입니다(39줄, 재수출과 __all__뿐). 내부를 어떻게 나눠도 DAG·CLI·테스트가 부르는 진입점(merge.stage, merge.apply_master 등)은 그대로 두려는 경계이고, __all__로 "여기까지가 쓰라고 만든 표면"을 코드로 표시했습니다.

비용도 있습니다. 파일이 늘어 호출 흐름을 따라가려면 한 번 더 점프해야 합니다. 그래서 순서를 master_sync 한 곳에 모아 그 비용을 줄였습니다. 처음부터 이 구조는 아니었고 구현 중에는 한 모듈이었습니다. 트랜잭션 경계와 이력 순서가 코드에서 드러나지 않는다고 판단해 제출 전에 책임별로 나눴습니다. 현재 include/pipeline/ 19개 모듈에서 가장 큰 파일이 quality.py 209줄이고, 병합 관련 4개는 59~159줄입니다.

누가 지적했는지 물으면 숨길 필요는 없습니다. 제출 전 재점검에서 찾았고, 검토 도구도 썼지만 지적 내용을 요구사항과 실행 결과로 확인하고 제출을 결정한 책임은 본인에게 있다는 것이 사실에 맞는 답입니다.

참고 위치

배포 상태: 커밋 2개(addad17, b34bf8a)가 로컬에 있고 푸시 대기 중입니다. 이 머신의 현재 네트워크에서 GitHub이 차단되어(22번·443번 모두 타임아웃, IP 직접 접속도 동일) 푸시가 되지 않습니다. 네트워크가 풀리면 git push origin main 한 번으로 Cloudflare 배포까지 이어집니다.