REMEMBER · TAKE-HOME ASSIGNMENT

제출 과제 — 내용과 아키텍처

근로복지공단 고용·산재보험 가입현황 2개년 CSV를 사업자등록번호 기준으로 동기화하는 Airflow 파이프라인. 2026-07-27 제출, 과제 통과.

과제가 요구한 것 (원문 요약)

배경. 근로복지공단이 관리하는 고용·산재보험 가입 사업장 내역 일부(2022년 초기 적재분, 2023년 업데이트분)로 데이터 정합성을 유지하며 동기화하는 파이프라인을 구축하라. 과제 기간은 5일.

Initial Load — 2022년 CSV를 마스터 테이블의 베이스라인으로 적재한다.
Sync — 2023년 CSV를 기준으로 Airflow를 통해 동기화한다.
- Insert: 사업자등록번호 기준으로 기존 DB에 없으면 신규 삽입.
- Update: 이미 있으면 2023년 파일의 최신 정보(사업장명·주소·업종명·상시근로자수·성립일자)로 갱신.
권장 스택: Python 3.11+, Airflow 2.x, DB는 SQLite/DuckDB 중 택 1. 관리용 컬럼·테이블은 자유롭게 추가 가능.
제출물 3종: ① DAG 소스 코드 ② Architecture & Key Definition(데이터 흐름도) ③ README에 설계 시 고려한 중점 사항.

원문: career/클로드이력서/리멤버앤컴퍼니-data-engineer/과제/과제-원문.pdf (2쪽)

무엇을 제출했나

단일 연도 파라미터 DAG 하나(employment_insurance_master_sync, @yearly·catchup=true·max_active_runs=1)가 2022 초기 적재와 2023 동기화를 같은 코드로 처리합니다. 검토자는 Docker만 있으면 ./up.sh 한 번으로 이미지 빌드 → 두 회차 실행 → 결과 조회까지 재현할 수 있게 했습니다.

Python 3.11Airflow 2.10.5DuckDBDocker Compose테스트 107개(컨테이너)

연도	소스 행수	신규	변경	변경 없음	소스에서 사라짐	마스터
2022 (Initial Load)	30,513	30,513	0	0	0	30,513
2023 (Sync)	23,261	2,174	4,000	17,087	9,426	32,687 (활성 23,261)

두 회차 모두 정합성 검사 8/8 통과, 2023 소스와 활성 마스터 직접 대조 불일치 0건. 컬럼 단위 변경 이력은 5,196행(주소 2,876 + 근로자수 1,972 + 이름 310 + 날짜 30 + 업종 8).

아키텍처

아래 3개 다이어그램은 제출물 ARCHITECTURE.md의 Mermaid 원본을 그대로 렌더링한 것입니다.

데이터 흐름도: 연도별 CSV가 resolve_source, land_raw, stage_and_validate, apply_and_finish 태스크를 거쳐 raw, staging, quarantine, master, change log, sync run 테이블로 흘러가는 구조 — **데이터 흐름.** 한 회차는 소스 확인 → 원문 적재(RAW) → 정규화·품질 판정(STG·격리) → 병합·이력·정합성 검사(MASTER) 순서로 흐릅니다. 파일이 없는 연도는 실패가 아니라 SKIPPED로 남깁니다. DuckDB 동시 쓰기를 피하려고 태스크를 직렬로 연결하고 `max_active_runs=1`로 제한했습니다.

실행 구조: 검토자가 up.sh를 실행하면 Docker Compose 안에서 Airflow와 PostgreSQL, DuckDB 볼륨이 동작하고 성공 후 DuckDB 파일이 호스트로 내보내진다 — **실행 구조.** 입력 CSV와 코드는 이미지에 굽고 호스트 경로를 바인드하지 않습니다(한글 경로·Drive 동기화 폴더에서 생기는 파일 접근 차이 회피). 실행 중 DuckDB는 이름 볼륨에 있고, 두 회차가 성공한 뒤에만 `warehouse/employment_insurance.duckdb`로 내보냅니다. PostgreSQL은 Airflow 메타데이터 전용입니다.

논리 데이터 모델: ops_sync_run 회차를 중심으로 raw_workplace, stg_workplace, ops_quarantine, master_change_log가 연결되고 master_workplace가 과제 결과물인 최종 마스터 테이블 — **논리 데이터 모델.** 회차(`ops_sync_run`)를 중심으로 각 테이블이 어느 회차에서 생겼는지 추적합니다. FK 표시는 조회용 논리 관계이고 실제 PK 제약은 `master_workplace.business_registration_no`와 `ops_sync_run.run_id` 두 개뿐입니다.

저장 계층과 재실행 동작

계층	테이블	책임	같은 연도 재실행 시
RAW	`raw_workplace`	CSV 원문과 출처 보존 — 규칙이 바뀌면 파일을 다시 받지 않고 여기서 재계산	해당 연도 교체
STAGING	`stg_workplace`	정규화·타입 변환·품질 판정(`has_valid_key`/`is_valid` 분리)	해당 연도 교체
MASTER	`master_workplace` ⭐	사업장당 1행 최신 상태 — 과제가 요구한 결과물	달라진 행만 갱신
MASTER	`master_change_log`	컬럼별 전·후 값 이력 (SCD2 대신 컬럼 단위 로그)	회차별 추가
OPS	`ops_quarantine`	격리(마스터 제외)와 확인 필요(반영 후 판정 대기) 기록	같은 run_id 교체
OPS	`ops_sync_run`	회차별 입력·정책·건수·검증 결과	run_id 기준 UPSERT

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

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

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

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

2. 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행.

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

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

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

4. 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)이 됐습니다.

DAG 한 회차의 태스크 호출 순서: resolve_source가 연도와 정책을 확정해 XCom으로 넘기고, land_raw가 스키마 검증 후 raw를 적재하고, stage_and_validate가 정규화와 품질 게이트를 거치고, apply_and_finish가 마스터 병합과 실행 기록 마감을 수행한다 — **태스크 호출 순서.** 커넥션을 태스크마다 열고 닫는 지점, 품질 게이트 실패 시 실행 기록을 FAILED로 닫고 즉시 중단하는 분기, 파일이 없을 때의 SKIP 분기가 함께 보입니다.

apply_master 트랜잭션 내부 순서: BEGIN 후 후보 임시 테이블 3개 생성과 건수 집계, 변경 이력 기록, 마스터 반영, 사라짐 표시, 정합성 8개 검사를 거쳐 COMMIT 또는 ROLLBACK으로 끝난다 — **`apply_master` 트랜잭션 내부.** 건수를 반영 전에 세고 이력을 갱신 전에 쓰는 두 순서가 이 단계의 핵심입니다. 정합성 검사도 커밋 전에 수행하므로, 실패한 회차는 마스터에 흔적을 남기지 않습니다.

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

`./up.sh` 실행 순서

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

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

up.sh 실행 순서: 사전 점검과 초기화, Docker Compose 빌드와 기동, health와 DAG 등록 폴링, ops_sync_run 상태 폴링, 로그 요약 출력과 inspect_db 실행, DuckDB 파일을 호스트로 복사하는 흐름 — **`./up.sh` 다섯 단계.** 준비 대기를 health · DAG 등록 · 파이프라인 완료 세 구간으로 나눈 이유는 실패 지점을 구분하려는 것입니다. 웹서버가 못 뜬 것, DAG 파싱이 깨진 것, 파이프라인이 느린 것은 대응이 다릅니다. DAG 등록이 안 되면 `dags list-import-errors`를 함께 출력합니다.

실패하면 trap이 컨테이너 상태와 최근 로그 120줄을 자동으로 출력합니다. 성공 후에는 서비스를 내리지 않고 남겨 두어 Airflow UI와 DuckDB를 직접 확인할 수 있게 하고, 종료는 ./down.sh로 분리했습니다.

설계 결정 7가지 (README 요약)

1. 데이터를 먼저 보고 정했다

구현 전에 두 CSV를 전수 조사. 근로자수가 2022년 1, 2023년 1.0 형식이라 문자열 비교면 공통 21,087건이 전부 변경으로 잡히고, 주소도 전남/전라남도류 표기 차이가 210건. 정규화 없이는 변경 건수가 5천 건 넘게 부풀었을 것.

2. 키는 고치지 않는다

사업자등록번호는 공백·하이픈만 제거. 9자리에 0을 채우는 보정은 실재하는 다른 사업장과 충돌해 되돌릴 수 없는 오염을 만들 수 있어, 형식이 어긋나면 격리하고 마스터에서 제외. 검증번호 불일치 9건은 두 연도에 같은 값으로 반복돼 오탈자로 단정하지 않고 WARN으로만 기록.

3. 2023년 빈 값은 명세대로 NULL 갱신, 보존은 옵션

주소 결측 2,249건 중 2,093건은 2022년에 값이 있어 추출 누락 가능성도 있으나, 과제가 "최신 파일 값으로 갱신"을 정의하므로 기본은 literal 적용. SYNC_PRESERVE_EXISTING_ON_NULL=true 경로를 함께 제공하고 두 정책 모두 검증 통과.

4. 2023년에 없는 사업장은 지우지 않는다

9,426건의 부재는 폐업/추출 누락/공개 범위 변경을 구분할 수 없고 과제 동작은 Insert·Update뿐. is_active=false, absent_since_year=2023으로 사실만 기록.

5. 테이블 6개 — 과제 결과물은 그중 하나

마스터만 만들면 Update가 규칙대로 됐는지 확인할 방법이 없다. RAW(재계산), 변경 로그(무엇이 바뀌었나), 실행 기록(어느 회차가 했나)을 함께 남겨 검증 가능성을 확보.

6. 절반만 반영된 상태를 남기지 않는다

마스터 반영·비활성 표시·정합성 검사 8종을 한 트랜잭션으로 묶고, 하나라도 실패하면 이력까지 롤백 후 회차를 FAILED로 마감. 같은 회차 재실행은 신규·변경 0건으로 수렴(멱등), 연도 역행은 가드가 차단.

7. 저장소는 DuckDB

연도별 CSV 전체 읽기·집합 조인·전수 검증이 중심인 배치라 행 단위 OLTP(SQLite)보다 컬럼 지향 실행과 read_csv·IS DISTINCT FROM을 가진 DuckDB가 맞았다. 대신 다중 프로세스 동시 쓰기에 맞지 않아 직렬 실행으로 제한 — 동시 writer가 필요해지면 중앙 DB로 옮긴다.

공백 정책 두 경로 — 실제 값 비교2026-08-11 로컬에서 두 정책을 각각 실행해 DB 두 개를 만들고 대조 (`scripts/run_pipeline.py --fresh --years 2022 2023`, 한쪽만 `--preserve-null`)

2023년 파일의 빈 값을 최신 상태로 볼지 추출 누락으로 볼지는 확정할 수 없어 정책으로 분리했습니다. 그 판단이 결과에서 실제로 얼마나 갈리는지 값으로 확인한 내용입니다. 소스와 코드는 같고 플래그 하나만 다릅니다.

결과 차이

정책	마스터 주소 결측	마스터 행수
기본 (덮어쓰기)	2,249	32,687
보존 (`--preserve-null`)	156	32,687

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

2,249가 156이 되는 이유

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

구분	건수
2022년에 주소가 있었다 — 정책이 갈리는 구간	2,093
2022년에도 주소가 없었다(또는 2023년 신규) — 정책 무관	156

보존 정책은 앞의 2,093건에서 기존 값을 지킵니다. 2,249 − 2,093 = 156. 156은 하한선입니다. 지킬 기존 값이 애초에 없는 케이스라 어떤 정책을 쓰든 NULL로 남습니다.

같은 사업장, 두 정책

사업자번호	사업장명	2022 원본 주소	2023 원본	기본 정책 마스터	보존 정책 마스터
`1010636012`	대성파워	경기 화성시 송산면 구레골길 30-24	`[빈 값]`	NULL	경기 화성시 송산면 구레골길 30-24
`1018105267`	(주)디비저축은행	서울 중구 남대문로 113	`[빈 값]`	NULL	서울 중구 남대문로 113
`1018186616`	(주)우광하이텍	서울 마포구 서강로9길 24	`[빈 값]`	NULL	서울 마포구 서강로9길 24

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

회차 카운터 (2023)

정책	신규	변경	변경없음	공백보존	비활성	마스터행
기본	2,174	4,000	17,087	0	9,426	32,687
보존	2,174	2,903	18,184	2,095	9,426	32,687

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

변경 이력에 남는 것

정책	컬럼	값이 있었는데 NULL로 지운 UPDATE
기본	`workplace_address`	2,093
기본	`industry_name`	1
기본	`insurance_established_date`	1
보존	—	0

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

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

알려진 한계 (README에 명시)

연 1회 전체 스냅샷·단일 DuckDB writer 전제. stage()가 한 해 분량을 메모리에서 정규화하므로 수백만 행에서는 청크 처리 필요.
주소 정규화는 시도명·공백 표기까지. 도로명·지번 통합은 외부 주소 데이터가 필요.
성립일자 역행 17건, 절단 의심 주소 4건은 판단 근거가 없어 값을 고치지 않고 ops_quarantine에 기록.

이 한계들은 면접 가이드의 과제 방어 질문 P0-10 ~ P1-27에서 답변으로 연결됩니다.