- 올리브영 크롤링 데이터를 Bronze → Silver → Gold 로 가공하는 ETL 파이프라인
- 원시 제품 데이터에서 성분을 정규화·집계해 Apache Iceberg 테이블로 적재하고, 변경분을 Neo4j 그래프에 증분 반영한다
Bronze (S3 JSON)
└ 입력 선택 게이트(src/bronze_gate)가 카테고리마다 쓸 run을 고른 뒤 DuckDB 로 로드
(진행 중·비정상 run 제외, 부분 수집·상품 수 50% 미만은 이전 run으로 대체, 행마다 source_run_id 기록)
Silver (Iceberg · oliveyoung_db)
├ oliveyoung_silver_current 최신 스냅샷 (overwrite)
├ oliveyoung_silver_history 시계열 누적 (같은 batch_date는 교체)
└ oliveyoung_silver_error 처리 실패 DLQ (overwrite)
Gold (Iceberg · oliveyoung_db)
├ gold_product_ingredients 성분 × INCI 메타데이터 mart (overwrite)
├ gold_ingredient_frequency 카테고리별 Top 50 성분 (같은 batch_date는 교체)
└ gold_product_change_log CDC 변경 이력 NEW/REMOVED/CHANGED (같은 batch_date는 교체)
Neo4j (GraphDB)
└ change_log 미처리 배치를 읽어 Product/CONTAINS 증분 반영 (checkpoint 기반)
성분 매칭에 필요한 KCIA·CosIng 표준 데이터는 별도 레포(INCI 파이프라인)가 만든
inci_db Iceberg 테이블에서 읽어온다
- Bronze 로드 — S3 glob 으로 서브카테고리별 최신
run_idJSON 탐색 (DuckDB) - 정제 — 노이즈 제거, 성분 문자열 정규화, 번들 제품 감지, 오타 교정
- 매칭 — Aho-Corasick 오토마타로 KCIA 표준명 일괄 매칭 (O(n+m))
- 분기 — 정상 → Silver current/history, 실패 → Silver error(DLQ)
- Gold — Silver × INCI 조인으로 성분 mart, 카테고리별 빈도 집계
- CDC — Silver 스냅샷 비교로 NEW/REMOVED/CHANGED 추출 → change_log
- Neo4j — change_log 미처리분을 그래프에 증분 적용
config/settings.py S3 / Glue Catalog / DuckDB 설정 (oliveyoung_db, inci_db)
oliveyoung_common/ 공용 S3 경로 상수 (git submodule)
data/ 오타·불량키워드·커스텀 성분 사전 (JSON)
src/
bronze_to_silver/ 정제 + KCIA 매칭 (cleaner, ac_builder)
과거 Bronze run의 history 전용 백필
silver_to_gold/ CDC + Gold 집계
silver_to_neo4j_csv/ Neo4j 초기 적재용 CSV 익스포트
reference_pipeline/ 사전 JSON → Iceberg 동기화
silver_pipeline/ Silver 테이블 스키마·생성·write
gold_pipeline/ Gold 테이블·CDC·Neo4j CSV write
models/ 배치 메타(run_id·batch_date) 재노출 (oliveyoung_common.batch)
maintenance/ Iceberg 스냅샷 만료 등 유지보수
neo4j_incremental.py change_log → Neo4j 증분 반영
scripts/entrypoint.sh 컨테이너 실행 모드 분기
dags/ Airflow DAG (DockerOperator)
oliveyoung_common은 서브모듈이므로 클론 시git clone --recurse-submodules, 이미 받았다면git submodule update --init.
설정은 코드에 상수로 두고, AWS 인증은 EC2 IAM Role 로 처리한다(키 주입 불필요)
python silver_pipeline/create_silver.py
python silver_pipeline/create_category_master.py
python reference_pipeline/create_reference_tables.py
python reference_pipeline/sync_reference_data.py
python gold_pipeline/create_gold_tables.py all # dq_metrics 포함 Gold 전체 (개별: dq_metrics 등 인자 지정)
python gold_pipeline/create_gold_product_ingredients.pyEC2 IAM Role 을 그대로 쓰기 위해 --network host 로 실행한다.
인자는 entrypoint.sh 의 모드명이다
docker build -t oliveyoung-pipeline .
docker run --network host oliveyoung-pipeline sync_reference
docker run --network host oliveyoung-pipeline bronze_to_silver
docker run --network host oliveyoung-pipeline silver_to_gold
docker run --network host oliveyoung-pipeline neo4j_incremental| 모드 | 동작 |
|---|---|
sync_reference |
사전 JSON → Iceberg 동기화 |
bronze_to_silver |
Bronze 로드 → 정제 → Silver |
backfill |
지정한 과거 Bronze run → Silver history + 백필 DQ (기본 dry-run) |
silver_to_gold |
CDC + Gold 집계/mart |
neo4j_incremental |
change_log → Neo4j 증분 반영 |
silver_to_neo4j_csv |
Neo4j 초기 적재용 CSV → S3 |
create_reference_tables / create_gold_product_ingredients |
테이블 생성 |
DockerOperator 기반 DAG 세 개로 운영한다
oliveyoung_pipeline (schedule=None — 크롤링 DAG 가 트리거)
sync_reference → bronze_to_silver → silver_to_gold → neo4j_incremental
oliveyoung_backfill (수동 — 과거 Bronze run의 history 전용 재처리)
backfill_history_and_dq
oliveyoung_silver_to_neo4j_csv (수동 — 그래프 초기 벌크 적재용)
silver_to_neo4j_csv
정상 운영에서 Neo4j 는 neo4j_incremental 로 변경분만 반영하고,
CSV 익스포트는 그래프를 처음 채울 때만 쓴다
main push 시 GitHub Actions 가 OIDC 로 AWS 를 assume 해 다음을 자동 처리한다(SSH·상시 키 불필요).
| 워크플로 | 트리거 | 동작 |
|---|---|---|
build-and-push |
main push | ARM 러너에서 이미지 빌드 → ECR push (ECRRole) |
deploy |
main push | SSM send-command 로 EC2 갱신 (SSMRole) |
update-submodule |
repository_dispatch |
oliveyoung_common 갱신 시 서브모듈 포인터를 최신으로 커밋 |
oliveyoung_common 이 바뀌면 그쪽에서 dispatch 를 쏴 이 레포의 서브모듈 포인터가 자동으로 따라온다.
| 컬럼 | 타입 | 설명 |
|---|---|---|
category_id |
string | 카테고리 (파티션 키) |
product_id |
string | 제품 ID (UUID v5) |
product_brand / product_name |
string | 브랜드 / 정제된 제품명 |
product_ingredients |
list<string> | KCIA 표준명으로 정규화된 성분 |
rating / review_count |
float / int | 평점 / 리뷰 수 |
review_stats |
map | 리뷰 통계 |
crawled_at |
timestamptz | 크롤링 시각 |
batch_job / batch_date |
string / timestamptz | 배치 메타 |
gold_product_ingredients— 성분 × INCI 메타데이터(INCI/한글명/기능/규제/사용수). overwritegold_ingredient_frequency— 카테고리별·전체(TOTAL) Top 50 성분. 배치 누적, 같은 batch_date는 교체gold_product_change_log—change_type= NEW / REMOVED / CHANGED. 배치 누적, 같은 batch_date는 교체
처리 실패 레코드를 error_type과 함께 적재해 사후 분석에 쓴다(overwrite, 다시 읽어 재처리하는 코드는 없음)
INCOMPLETE_DATA_REJECTED · OPTION_BUNDLE_REJECTED · INVALID_METADATA_REJECTED
· HETEROGENEOUS_BUNDLE_REJECTED · DUPLICATE_PRODUCT_REJECTED · HIDDEN_BUNDLE_REJECTED
· UNMAPPED_RESIDUAL.
_REJECTED는 상품 제외, UNMAPPED_RESIDUAL은 매핑 안 된 잔여 성분 경고라 정상 적재와 겹칠 수 있다.
DQ에는 유형별 건수가 err_<error_type>로 기록되고, 유형이 비면 UNCLASSIFIED로 센다.
oliveyoung_backfill은 한 크롤 source_run_id의 Bronze JSON을 현재 사전·정제 규칙으로
다시 처리한다. 쓰기 범위는 oliveyoung_silver_history와
dq_metrics(stage=bronze_to_silver_backfill)뿐이며, silver_current·silver_error·Gold·CDC·Neo4j는
변경하지 않는다. 과거 시점 규칙을 재현하거나 그래프를 자동 보정하는 작업은 아니다.
Airflow UI에서 source_run_id를 넣고 dry-run을 먼저 실행해 manifest 무결성, 입력 건수,
동일 날짜의 정상/다른 백필 충돌을 검토한다. 적용은 별도의 새 DAG run에서 mode=apply와
동일한 confirm_source_run_id를 제출해야 한다. in_progress manifest와 날짜 충돌은 강제로
우회할 수 없고, 무결성 불일치는 검토 후 allow_incomplete=true로만 명시적으로 승인한다.
같은 키 재실행은 history와 백필 DQ의 해당 범위를 조건부 교체한 뒤 재조회 검증하며,
검증 성공 후 Discord 완료 리포트를 보낸다.
배포 전제, dry-run 확인 필드, 실패·재시도 절차는 백필 운영 절차를 따른다.
파이프라인 각 단계가 남기는 데이터 정합성 수치를 모으는 key/value(EAV) 테이블.
스키마 진화 없이 지표를 추가할 수 있고, 여러 파이프라인(crawl·bronze_to_silver·silver_to_gold 등)이 같은 테이블에 적재한다.
batch_date(YYYY-MM-DD) · run_id · stage · metric_name · metric_value · target_table · created_at.
- 스키마·writer는
oliveyoung_common/dq_metrics.py가 소유(순수함수write_dq_metrics), 생성은create_gold_tables.py dq_metrics. - 각 단계가
log_dq(Loki 로그) +write_dq_metrics(테이블)로 같은 수치를 이중 기록(테이블 적재는 비치명적). - bronze→silver는
silver_error와 함께err_<error_type>건수를 기록한다. 백필은 별도 stage인bronze_to_silver_backfill을 사용해 정상 실행과 섞이지 않는다. batch_date(단계 관통 논리 배치 날짜)로 crawl·bronze_to_silver·silver_to_gold를 한 배치로 묶어, 대시보드 그래프의 한 시점을 클릭하면 그 배치의 silver 행으로 드릴다운한다.run_id는 초단위 유니크 실행 식별.- 테이블 실제 위치는
GOLD_PATH(olive_young_gold/dq_metrics/). 조회는 dq_api(pyiceberg+DuckDB)가 읽어 Grafana(Infinity)에 노출. - 단계가 끝나면 그 배치의
dq_metrics요약을 Discord 완료 리포트로 보낸다(옵트인 —DISCORD_DQ_WEBHOOK_URL있을 때만, 미설정이면 미전송). 만성적이지만 비치명적인 지표는 알람 대신 이 리포트로 노출.
- current / history 분리 — 최신 조회용(
current, overwrite)과 시계열 추적용(history, 배치 누적)을 나눠 재읽기 없이 변화를 본다. - 재실행 멱등 — 실패한 태스크를 Clear·재실행하거나 같은
batch_date를 다시 돌려도 결과가 한 벌이다. history·frequency·change_log·dq_metrics는 같은batch_date의 이전 실행 행을 조건부 overwrite로 교체하고(백필 행은 보존), bronze→silver는 history → error → current 순서로 쓴 뒤 재조회 검증하고 DQ·리포트를 남긴다. current를 마지막에 쓰는 이유는 CDC가 current 최근 두 스냅샷을 비교하기 때문. Airflow 실행은BATCH_DATE가 비면 중단한다(교체 키 보호). 재실행이면 완료 리포트 제목에(재실행)이 붙는다. 단,neo4j_incremental까지 성공한 배치의 silver_to_gold는 다시 돌리지 않는다(그래프 재반영).- history 행의
batch_job은 실행 ID(bronze_to_silver_YYYYMMDD_HHMMSS)이고 DQrun_id와 같다(2026-10 이전 행은bronze_to_silver, 교체 범위에 포함). - 정제 결과 0건은 장애로 보고 쓰기 전에 중단한다. DQ 요약 기록이 실패하면 유형별(
err_*)은 쓰지 않는다. CSV 저장은 비치명. - 백필 apply는
oliveyoung_pipeline이 실행·대기 중이면 거부한다(정상 b2s와 s2g 사이에 백필 스냅샷이 끼면 CDC NEW가 오염). - 남는 위험: current를 쓴 뒤 검증이 실패해 재실행하면 같은 내용 current 스냅샷이 하나 더 생겨 CDC 최근 두 스냅샷 비교가 그 배치 변화를 놓칠 수 있다. 백필 중 크롤이 정상 실행을 트리거하는 반대 방향도 막지 않는다(백필은 수 분). 근본 해결은 CDC가 배치 경계 스냅샷을 고르는 것.
- history 행의
- CDC = Iceberg 스냅샷 비교 — 전체 스캔 없이 최신 2개 스냅샷의
product_id집합·추적 필드 차이로 변경분을 뽑는다. - Neo4j 증분 + checkpoint —
neo4j_sync_checkpoint에 마지막 처리 배치를 남겨 change_log 의 신규분만 그래프에 반영한다. - Aho-Corasick 매칭 — 수천 개 표준명을 한 번의 스캔으로 동시 탐색. 성분명 내부 쉼표는 마스킹으로 분리.
- 사전 분리 관리 — 오타·불량키워드·커스텀 성분 사전을 Git 의 JSON 으로 두고
sync_reference_data.py로 Iceberg 동기화. 코드 재배포 없이 수정. - 정합성 = 로그 + 테이블 이중 소스 — 각 단계가
log_dq(Loki)와write_dq_metrics(Icebergdq_metrics)로 같은 수치를 남긴다. 로그 기반 대시보드는 유지하고, 테이블 소스는 dq_api→Grafana로 별도 대시보드에 노출(장기 추세·드릴다운). 같은 수치를 단계 종료 시 Discord 완료 리포트로도 요약 발송한다. 지표 추가는 key/value라 스키마 진화 불필요.
- 스토리지: AWS S3 / 메타스토어: AWS Glue Catalog / 리전:
ap-northeast-2 - 런타임: Python 3.12, EC2 (IAM Role 인증)
- 주요 의존성:
pyiceberg,duckdb,pyahocorasick,pandas/pyarrow,boto3,neo4j