ETL 단계와 실패 경계

미션 ZIP에서 제공하는 기존 normalize를 사용합니다. 다음 코드는 따라하기를 독립 실행하기 위해 함수 구현을 함께 싣습니다. 저장 단위에서는 cleaning.py에서 import합니다.

"""결합 표를 정제합니다. 입력은 보존하고 결측값은 None으로 다룹니다."""
import argparse
import csv
import json
import math
import re
from datetime import datetime
from pathlib import Path

HEADER = ['region','date','total_vehicles','valid_observations','weather_region','rain_mm','station_rows','valid_stations','join_status']

def integer(text, nullable=False):
    text = text.strip()
    if nullable and text == '':
        return None
    if not re.fullmatch(r'[0-9]+', text):
        raise ValueError('INTEGER')
    return int(text)

def day(text):
    text = text.strip()
    for fmt, pattern in [('%Y-%m-%d', r'[0-9]{4}-[0-9]{2}-[0-9]{2}'), ('%Y/%m/%d', r'[0-9]{4}/[0-9]{2}/[0-9]{2}')]:
        if re.fullmatch(pattern, text):
            try:
                return datetime.strptime(text, fmt).date().isoformat()
            except ValueError:
                break
    raise ValueError('DATE')

def normalize(row):
    result = dict(row)
    result['region'] = row['region'].strip()
    if not result['region']:
        raise ValueError('KEY')
    result['date'] = day(row['date'])
    for key in ['total_vehicles','valid_observations','station_rows','valid_stations']:
        try:
            result[key] = integer(row[key], key != 'valid_observations')
        except ValueError:
            raise ValueError('INTEGER:' + key) from None
    rain = row['rain_mm'].strip()
    try:
        rain = None if rain == '' else float(rain)
    except ValueError:
        raise ValueError('RAIN') from None
    if rain is not None and (not math.isfinite(rain) or rain < 0):
        raise ValueError('RAIN')
    result['rain_mm'] = rain
    if (result['total_vehicles'] is None) != (result['valid_observations'] == 0):
        raise ValueError('CONTRACT:traffic')
    result['weather_region'] = row['weather_region'].strip()
    result['join_status'] = row['join_status'].strip()
    if result['join_status'] not in {'matched','mapping_missing','weather_missing','rain_missing'}:
        raise ValueError('STATUS')
    return result


raw = {'region':'A','date':'2026/09/01','total_vehicles':'0','valid_observations':'1','weather_region':'WA','rain_mm':'','station_rows':'1','valid_stations':'0','join_status':'rain_missing'}
row = normalize(raw)
print(row['date'],row['total_vehicles'],row['rain_mm'])

예외가 발생한 뒤 저장 함수가 호출되지 않는지 확인합니다.

calls=[]
def collect():
    calls.append('collect')
    return 'snapshot.csv'
def transform(path):
    calls.append('transform')
    raise ValueError('HEADER: missing total_vehicles')
def load(rows):
    calls.append('load')
try:
    rows=transform(collect())
    load(rows)
except ValueError as error:
    print(error)
print('called', ','.join(calls))

중복 키 실패 후 이전 논리적 행이 남습니다. 실패한 새 행을 저장 성공으로 보고하지 않습니다.

import sqlite3
from contextlib import closing
with closing(sqlite3.connect(':memory:')) as db:
    db.execute('CREATE TABLE observations(region TEXT,date TEXT,total_vehicles INTEGER,PRIMARY KEY(region,date))')
    with db:
        db.execute('INSERT INTO observations VALUES(?,?,?)',('A','2026-09-01',100))
    try:
        with db:
            db.execute('DELETE FROM observations')
            db.executemany('INSERT INTO observations VALUES(?,?,?)',[('A','2026-09-02',0),('A','2026-09-02',80)])
    except sqlite3.IntegrityError:
        print('rollback')
    print(db.execute('SELECT region,date,total_vehicles FROM observations ORDER BY region,date').fetchall())

starter.zip을 푼 루트에서 아래 명령을 실행합니다. 기존 60개와 새 20개, 총 80개 기본 테스트 통과입니다. HTTP 통합 9개는 기본 건너뜀입니다. starter는 실제 저장을 요구하는 검사가 실패합니다. 실패의 actual과 expected를 읽고 TODO를 채운 뒤 다시 검사합니다. HTTP 통합은 이 작성 샌드박스의 포트 권한 때문에 실행하지 못했으며 기본 검사와 구분합니다.

python3 -m unittest discover -s tests -v

solution 또는 TODO를 완성한 미션 ZIP 루트에서 python3 -m etl.run을 실행합니다. 기본 설정은 네트워크 없이 joined.csv를 읽으며 다음은 같은 ZIP을 임시 폴더에 복사해 실제 실행한 결과입니다. 수집 바이트 수와 저장 행 수를 확인하고 etl.sqlite의 observations를 조회해 결측과 영점을 대조합니다.

m06 solution을 포함한 ZIP의 etl/run.py에서 load 호출을 연결합니다. pipeline.json에 지정한 경로를 사용합니다. 정상 6행, 영점과 NULL, 같은 입력 반복, 열 누락·날짜 오류·키 충돌 시 저장 차단, 기존 DB와 원본 보존 검사를 통과시킵니다. 기존 코드와 보고서는 유지합니다. 실제 HTTP 통합은 BOOTCAMP_HTTP=1로 별도 실행합니다.