README.md

FTP Batch Pipeline

과제 선정 배경

EUV 설비 내부 오염이 확산되면 particle이 reticle 또는 reticle stage(RS)에 유입될 수 있습니다. 이로 인해 reticle을 scanner에 loading하기 전 수행하는 RBI(Reticle Backside Inspection) 검사에서 fail이 발생하고, reticle loading이 진행되지 못하는 문제가 발생합니다.

따라서 reticle backside 오염과 관련된 검사 텍스트 및 이미지를 안정적으로 수집하고, 발생 시점 기준으로 연결해 분석 가능한 형태로 보관할 수 있는 시스템이 필요합니다.

과제 목표

RUBI 텍스트와 RUIP 이미지를 수집 및 매칭하여 reticle backside 오염 분석 데이터를 구축합니다. 특히 RBI fail 원인 particle의 발생 시점과 관련 이미지를 추적할 수 있도록 텍스트 검사 결과와 이미지 데이터를 1:1로 연결하고, 변환된 PNG 결과물을 분석 서버 경로에 업로드하는 배치 파이프라인을 구현합니다.

기술적 선택지 및 선택 사유

본 시스템은 reticle 오염 원인 분석 알고리즘 자체가 아니라, 분석에 필요한 검사 텍스트와 이미지 evidence를 정합성 있게 연결하는 데이터 파이프라인입니다. 구현 시 선택할 수 있었던 주요 기술적 분기점과 현재 선택은 다음과 같습니다.

분기점	선택 가능 방식	현재 선택	선택 사유
처리 기준	이미지 기준, 텍스트 기준, 텍스트+이미지 통합 처리	COMBINED 텍스트 기준 통합 처리	RBI 검사 결과 텍스트가 기준 데이터이고, 이미지는 분석 근거 자료로 연결되는 구조가 적합합니다.
매칭 방식	파일명 기준, 시간 기준, reticle ID/검사 ID 기준, 수동 매칭	파일명 prefix와 timestamp 기반 자동 매칭	현재 입력 데이터에서 공통으로 확보 가능한 식별자가 파일명 prefix와 생성 시각입니다.
매칭 시간 범위	동일 시각만 허용, 1분 이내, 5분 이내, 10분 이상, 가장 가까운 파일 무조건 매칭	5분 이내 최근접 이미지 1개	너무 좁으면 정상 데이터가 누락되고, 너무 넓으면 잘못된 이미지가 연결될 수 있어 5분을 기준으로 둡니다.
매칭 실패 처리	전체 skip, 텍스트만 저장, 미매칭 상태 저장, 에러 후 재시도	COMBINED에서는 텍스트만 저장	텍스트 검사 결과는 분석의 기준 데이터이므로 이미지가 없어도 보존합니다.
중복 처리	매번 변환/업로드, 기존 결과 skip, 해시 비교, DB 상태값 기준 재처리	서버 PNG 존재 시 skip, 로컬 PNG 존재 시 재사용	반복 실행 시 불필요한 이미지 변환과 업로드 비용을 줄입니다.
완료 기준	DB 저장 성공, PNG 변환 성공, 서버 업로드 성공, 원본 삭제 성공	서버 업로드 성공 후 DB 저장 및 원본 삭제	분석 서버에 결과 이미지가 없는 상태에서 완료 처리되는 문제를 줄입니다.
원본 파일 처리	원본 유지, 성공 후 삭제, archive 이동, 실패 파일 별도 보관	처리 성공 후 FTP 원본 삭제	중복 처리를 줄이고 입력 경로를 정리합니다.
날짜 범위	입력일만 처리, 전날+입력일 처리, 전후 N일 처리, 마지막 처리 시점 이후 처리	전날+입력일 2일 처리	자정 전후에 생성된 텍스트와 이미지의 매칭 누락을 줄입니다.

개요

이 프로젝트는 FTP 서버의 RUBI 텍스트와 RUIP 이미지를 수집해 처리하는 reticle 오염 분석용 배치입니다.

• RUBI: 텍스트 파일을 파싱해 SQLite에 벌크 인서트합니다.
• RUPI: 이미지와 텍스트를 매칭해 PNG로 변환하고 FTP rbi/ruip/.../*.png로 업로드합니다.
• COMBINED: 텍스트를 기준으로 이미지까지 같이 처리하는 통합 모드입니다.

현재 운영 기본값은 COMBINED 기준입니다. 입력일 하나를 받으면 내부적으로 전날 + 입력일 두 날짜를 함께 처리합니다.

ER Dose Error 배치

ER_DOSE_RAW 배치는 mbeat.er_data_raw의 dose warning 로그를 파싱해 prism_common.er_dose_raw_parsed에 적재하며, 수집 완료 후 prism_common.de_trend_die_yield_daily 일별 DIE Yield 서머리 테이블을 갱신합니다. parsed 테이블에는 eq_name, code, code_occur_time, title, contents와 contents에서 실제로 필요한 exposure_handle, action_handle, lot_id, lot_name, lot_seq, wafer_seq, de_err, n_slit, use_yn을 저장합니다. 조회 대상 code는 DW-3411, DW-3425, DW-343A, DW-343B, LO-0050, LO-0051, LO-0052, LO-0061, LO-8166, LO-8167, KE-9103, KE-9104이며, 코드 값은 DB에 저장된 원본 형식 그대로 비교합니다.

배치는 code_occur_time 기간 조건으로 조회한 후보를 한 번에 메모리로 올리지 않고, chunk 단위로 읽어서 파싱 후 바로 COPY 적재합니다. 파티션 적재는 공통 copy_insert_to_partition_table을 사용하며, 적재 전에 DataFrame 컬럼을 테이블의 물리 컬럼 순서와 동일하게 정렬한 뒤 COPY ... FROM STDIN WITH CSV HEADER를 실행합니다. 현재 기본 chunk 크기는 ER_DOSE_RAW 및 ER_DOSE_EUV 배치 모두 30000이며 실행 시 조정할 수 있습니다. 조회는 SQLAlchemy 서버사이드 커서(stream_results=True, max_row_buffer=chunk_size) 기반 스트리밍으로 수행되지만, 실제 메모리 사용량은 chunk 크기와 raw contents 크기에 영향을 받으므로 운영 환경에 맞게 조정해야 합니다. 청크 단위로 처리되더라도 설비(eq_name)별로 이전에 파싱한 lot_seq와 wafer_seq를 기억하여 지속 적용합니다. ER_DOSE_RAW은 적재 worker 1개를 사용해 이전 청크의 COPY와 다음 청크의 조회·파싱을 겹쳐 실행하며, 대기 중인 DataFrame은 최대 1개로 제한합니다. 이 변경의 운영 실측 결과는 DB 스트리밍 및 RAW 성능 개선 문서에 기록합니다.

ER_DOSE_RAW와 ER_DOSE_EUV의 processor 기본 실행은 최근 2일 lookback 모드입니다. 실행일 기준 오늘 포함 최근 2일을 날짜별로 검사하고, 먼저 원천 raw와 parsed의 전체 건수를 비교합니다. 두 배치 모두 전체 건수가 다르고 parsed 건수가 0보다 클 때만 원천에서 (eq_name, code, code_occur_time)이 같은 행을 중복 제거한 건수를 한 번 더 계산합니다. 이 건수가 parsed 건수와 같으면 중복으로 인한 차이이므로 스킵하고, 여전히 다르거나 parsed 건수가 0이면 해당 날짜 parsed 파티션을 TRUNCATE한 뒤 원천 raw를 처음부터 다시 파싱해 적재합니다. ER_DOSE_EUV_TARGET_DATE를 명시하면 해당 날짜 1일만 같은 방식으로 검사하고, ER_DOSE_START_TIME/ER_DOSE_END_TIME를 명시하면 count 비교 없이 지정한 시간 범위를 처리합니다. EUV source count는 root cause 파싱 대상인 contents만 세어 parsed count와 비교합니다.

DW 로그에서 exposure_handle이 같은 설비의 이전 값보다 1000 이상 커지면 테스트샷성 row로 보고 저장은 하되 use_yn='N'으로 표시합니다. 일반 분석에서는 use_yn='Y' 조건을 사용하면 되고, row 자체는 저장되므로 raw count와 parsed count 비교가 계속 어긋나는 문제를 피할 수 있습니다.

ER_DOSE_EUV는 mbeat.er_data_raw_euv 기반 root cause 결과용 실행입니다. 대상 결과는 prism_common.er_dose_euv_parsed에 저장하며, er_line, belong, type은 저장하지 않습니다. contents에서 dose_error_detected_in_file, exposure_id, time, root_cause와 각종 EUV metric 컬럼을 파싱해 적재합니다. 컬럼명은 소문자 snake_case 기준으로 공백, ., -, <, =를 _로 치환하며, 파생 컬럼은 root_cause_code만 저장합니다. 파싱 및 적재가 완료되면 prism_common.de_trend_root_cause_daily 서머리 테이블에 일별/설비별/원인별 발생 빈도(frequency)를 자동 UPSERT 합니다.

상세 스키마와 파싱 규칙은 ER_DOSE_ERROR.md를 기준으로 관리합니다.

디렉토리 구조

ftp_batch/
├── app/
│   └── batch_runner.py
├── common/
│   ├── date_utils.py
│   └── path_utils.py
├── config/
│   └── local_test_settings.py
├── infra/
│   ├── db_manager.py
│   └── ftp_scanner.py
├── matching/
│   └── image_text_matcher.py
└── processors/
    ├── rubi_processor.py
    └── rupi_processor.py

batch_main/
└── main.py
airflow_modules/
└── ftp_batch_jobs.py
dags/
└── ftp_batch_hourly_dag.py

main.py
test.py
init_db.py
local_ftp_server.py
IMAGE_TEXT_MATCHING.md

처리 흐름

RUBI

1. /RUBI/<YYYYMMDD> 스캔
2. 파일별 다운로드
3. 텍스트 파싱 후 DataFrame 생성
4. SQLite 벌크 인서트
5. commit 성공 시 FTP 원본 삭제

RUPI

1. /RUIP/<YYYYMMDD> 이미지 스캔
2. /RUBI/<YYYYMMDD> 텍스트 후보 조회
3. 이미지 기준으로 가까운 텍스트 매칭
4. 로컬 PNG 준비
5. 마지막에 업로드 큐를 순차 업로드

COMBINED

1. 전날 + 입력일의 /RUBI와 /RUIP를 모두 먼저 다운로드
2. 텍스트 파일을 시간순으로 파싱
3. 텍스트보다 늦지 않고 5분 이내인 가장 가까운 이미지 1개를 1:1 매칭
4. 매칭 이미지가 있으면 PNG 준비 후 업로드 큐에 등록
5. 매칭 없는 텍스트는 텍스트만 DB 저장
6. 마지막 업로드 단계에서 성공한 항목만 DB finalize 후 텍스트/이미지 FTP 원본 삭제

주요 구성 요소

BatchRunner

위치: /Users/parkjunho/PycharmProjects/PythonStudy/ftp_batch/app/batch_runner.py

• RUBI, RUPI, COMBINED 실행 분기
• 전날 + 입력일 두 날짜 처리
• COMBINED에서 로컬 PNG 3일 보관 cleanup
• 업로드 큐 제어

FTPScanner

위치: /Users/parkjunho/PycharmProjects/PythonStudy/ftp_batch/infra/ftp_scanner.py

• 날짜 폴더 스캔
• 다운로드/업로드
• 원격 존재 여부와 크기 검증
• FTP 원본 삭제

현재 scan()은 날짜 폴더 바로 아래 파일만 조회하고 재귀 스캔은 하지 않습니다.

DBManager

위치: /Users/parkjunho/PycharmProjects/PythonStudy/ftp_batch/infra/db_manager.py

• select(query, params=None, connection=None)
• bulk_insert_df(table_name, df, connection=None)
• execute(query, params=None, connection=None)
• transaction()

조회는 pandas.read_sql_query, 인서트는 DataFrame -> sqlite3.executemany 기반 벌크 인서트입니다.

RubiProcessor

위치: /Users/parkjunho/PycharmProjects/PythonStudy/ftp_batch/processors/rubi_processor.py

• 텍스트 디코딩
• 라인 파싱
• rubi_ingest용 DataFrame 생성

RupiProcessor

위치: /Users/parkjunho/PycharmProjects/PythonStudy/ftp_batch/processors/rupi_processor.py

• TIF -> PNG 변환
• rupi_ingest upsert/finalize
• 로컬 PNG 경로 생성

image_text_matcher

위치: /Users/parkjunho/PycharmProjects/PythonStudy/ftp_batch/matching/image_text_matcher.py

• 파일명에서 prefix, timestamp 추출
• 텍스트 기준 최근접 이미지 선택
• 이미지 기준 최근접 텍스트 선택
• RUIP -> rbi/ruip/.../*.png 경로 변환

실행 방법

공통 main 실행

운영용 공통 진입점은 main.py 입니다. 실제 분기 로직은 batch_main/main.py에 두고, 루트 main.py는 얇은 실행 래퍼로 유지합니다.

BATCH_TARGET 값에 따라 RBI 배치와 ER Dose 배치를 분기합니다.

먼저 프로젝트 루트로 이동합니다.

cd /Users/parkjunho/PycharmProjects/PythonStudy

RBI 실행:

BATCH_TARGET=RBI \
RBI_INPUT_DATE=2026-05-31 \
RBI_PARSER=COMBINED \
.venv/bin/python main.py

ER_DOSE_RAW 실행:

ER_DOSE_DB_DSN='postgresql://user:password@host:5432/dbname' \
.venv/bin/python main.py --date 2026-05-31 --parser ER_DOSE_RAW

또는 er_dose.properties 에 DB 연결을 넣고 실행할 수 있습니다.

ER_DOSE_DB_DSN=postgresql://user:password@host:5432/dbname

ER_DOSE_EUV도 ER_DOSE_RAW와 별도 실행으로 동작합니다.

필수 환경변수:

• BATCH_TARGET: RBI, ER_DOSE_RAW, ER_DOSE_EUV
• RBI_INPUT_DATE: RBI 기준 날짜, YYYY-MM-DD
• ER_DOSE_DB_DSN 또는 DATABASE_URL: PostgreSQL DSN
• 또는 프로젝트 루트 er_dose.properties 파일의 ER_DOSE_DB_DSN

선택 환경변수:

• RBI_PARSER: RUBI, RUPI, COMBINED, 기본값 COMBINED
• ER_DOSE_RAW_TARGET_DATE: ER Dose raw 대상 날짜, YYYY-MM-DD
• ER_DOSE_EUV_TARGET_DATE: ER Dose EUV 대상 날짜, YYYY-MM-DD
• ER_DOSE_CHUNK_SIZE: ER Dose raw/euv fetch chunk 크기
• ER_DOSE_LOOKBACK_DAYS: ER_DOSE_RAW 기본 lookback 일수, 기본값 2
• INPUT_DATE: RBI_INPUT_DATE 대체값
• ER_DOSE_TARGET_DATE: raw 레거시 대상 날짜 이름
• TARGET_DATE: ER_DOSE_TARGET_DATE 레거시 대체값
• START_TIME: ER_DOSE_START_TIME 대체값
• END_TIME: ER_DOSE_END_TIME 대체값
• ER_DOSE_START_TIME, ER_DOSE_END_TIME: 기존 시간 범위 직접 지정 방식도 계속 지원. 두 값은 반드시 함께 지정해야 함

로컬 패키지 설치:

mkdir -p .vendor
pip3 install --target .vendor SQLAlchemy psycopg2-binary

sitecustomize.py 가 .vendor 를 자동으로 sys.path 에 추가하므로 별도 PYTHONPATH 설정 없이 실행할 수 있습니다.

실행 규칙:

• main.py --date <YYYY-MM-DD> --parser RUBI|RUPI|COMBINED이면 FTP 기반 RBI 배치를 실행합니다.
• main.py --date <YYYY-MM-DD> --parser ER_DOSE_RAW이면 해당 하루의 ER RAW 로그 파싱 배치를 실행합니다.
• main.py --date <YYYY-MM-DD> --parser ER_DOSE_EUV이면 EUV root cause 파싱 배치를 실행합니다.
• 내부적으로 ER_DOSE_RAW는 ER_DOSE_RAW_TARGET_DATE, ER_DOSE_EUV는 ER_DOSE_EUV_TARGET_DATE를 사용합니다.
• raw/euv processor 모두 대상 날짜 기준으로 하루 범위를 계산합니다.
• BATCH_TARGET=ER_DOSE_RAW가 현재 raw 배치의 기본 이름입니다. 레거시 ER_DOSE도 계속 지원합니다.
• BATCH_TARGET=ER_DOSE_RAW 또는 BATCH_TARGET=ER_DOSE_EUV를 환경변수만으로 실행하고 날짜 인자를 주지 않으면 최근 2일 lookback + 날짜별 count 비교 기반 재적재 전략이 적용됩니다. 두 배치 모두 전체 count가 다를 때만 고유 이벤트 count를 추가로 비교합니다.
• 인자 없이 main.py를 실행하면 기존과 동일하게 환경변수 기반 실행입니다.

DB 초기화

/Users/parkjunho/PycharmProjects/PythonStudy/.venv/bin/python \
  /Users/parkjunho/PycharmProjects/PythonStudy/init_db.py

로컬 FTP 서버

/Users/parkjunho/PycharmProjects/PythonStudy/.venv/bin/python \
  /Users/parkjunho/PycharmProjects/PythonStudy/local_ftp_server.py

배치 실행

RUBI

/Users/parkjunho/PycharmProjects/PythonStudy/.venv/bin/python \
  /Users/parkjunho/PycharmProjects/PythonStudy/test.py \
  --input-date 2026-04-11 \
  --parser RUBI

RUPI

/Users/parkjunho/PycharmProjects/PythonStudy/.venv/bin/python \
  /Users/parkjunho/PycharmProjects/PythonStudy/test.py \
  --input-date 2026-04-11 \
  --parser RUPI

COMBINED

/Users/parkjunho/PycharmProjects/PythonStudy/.venv/bin/python \
  /Users/parkjunho/PycharmProjects/PythonStudy/test.py \
  --input-date 2026-04-11 \
  --parser COMBINED

ER_DOSE_RAW

ER_DOSE_DB_DSN='postgresql://user:password@host:5432/dbname' \
/Users/parkjunho/PycharmProjects/PythonStudy/.venv/bin/python \
  /Users/parkjunho/PycharmProjects/PythonStudy/main.py \
  --date 2026-04-11 \
  --parser ER_DOSE_RAW

ER_DOSE_EUV

ER_DOSE_DB_DSN='postgresql://user:password@host:5432/dbname' \
/Users/parkjunho/PycharmProjects/PythonStudy/.venv/bin/python \
  /Users/parkjunho/PycharmProjects/PythonStudy/main.py \
  --date 2026-04-11 \
  --parser ER_DOSE_EUV

직접 ER Dose 실행 스크립트를 사용할 수도 있습니다.

ER_DOSE_RAW를 날짜 없이 실행하면 최근 2일 lookback 모드로 동작합니다.

python3 -m er_dose.run_er_dose_batch \
  --parser ER_DOSE_RAW \
  --dsn 'postgresql://user:password@host:5432/dbname'

python3 -m er_dose.run_er_dose_batch \
  --date 2026-04-11 \
  --parser ER_DOSE_RAW \
  --dsn 'postgresql://user:password@host:5432/dbname'

python3 -m er_dose.run_er_dose_batch \
  --date 2026-04-11 \
  --parser ER_DOSE_EUV \
  --dsn 'postgresql://user:password@host:5432/dbname'

기존 시간 범위 직접 지정 방식도 계속 지원합니다.

python3 -m er_dose.run_er_dose_batch \
  --start-time 2026-04-11T00:00:00 \
  --end-time 2026-04-12T00:00:00 \
  --dsn 'postgresql://user:password@host:5432/dbname'

Airflow

• DAG: /Users/parkjunho/PycharmProjects/PythonStudy/dags/ftp_batch_hourly_dag.py
• 래퍼: /Users/parkjunho/PycharmProjects/PythonStudy/airflow_modules/ftp_batch_jobs.py
• 스케줄: 매시간 정각
• 기본 실행 모드: COMBINED

설정

위치: /Users/parkjunho/PycharmProjects/PythonStudy/ftp_batch/config/local_test_settings.py

주요 값:

• CLIENT_FTP_ROOT_PATH = "/RUIP"
• TEXT_FTP_ROOT_PATH = "/RUBI"
• SERVER_FTP_ROOT_PATH = "/rbi"
• LOCAL_WORK_DIR
• LOCAL_DB_PATH
• RUPI_SCALE_PERCENT

데이터 정합성 / 로컬 보관 정책

• RUBI는 DB commit 성공 전에는 FTP 원본을 삭제하지 않습니다.
• COMBINED에서 매칭된 항목은 업로드 성공 후 DB commit, 그 다음 FTP 원본 삭제 순서로 처리합니다.
• 로컬 raw txt/tif는 scratch 파일이라 처리 후 즉시 삭제합니다.
• 로컬 png는 재업로드 캐시로 3일 유지하고, 배치 시작 시 3일 지난 파일만 정리합니다.

현재 한계

• commit 성공 후 FTP delete 실패 재시도 큐는 아직 없습니다.
• DB는 SQLite 기준입니다.
• scan()은 날짜 폴더 바로 아래 파일만 읽고 재귀 스캔은 지원하지 않습니다.

이번 범위에 넣지 않은 것

• .env 기반 설정 로딩
• ON CONFLICT DO NOTHING 기반 중복 처리 정책 변경
• FTP 재귀 디렉토리 삭제 유틸
• 00:00 ~ 01:59 전날만 처리하는 시간 분기
• PostgreSQL 전환

자세한 매칭 규칙은 /Users/parkjunho/PycharmProjects/PythonStudy/IMAGE_TEXT_MATCHING.md 를 참고하면 됩니다.