Transform — 청소, cast, derive · Data Engineering Quest

로직이 사는 곳

Extract 는 byte 를 끌어오고, load 는 써. 그리고 transform — 본인 팀이 월급을 받는 이유인 그 로직이 앉는 자리가 여기야. 규율은 모든 transformation 을 명시적으로, 이름 붙여서, 테스트 가능하게 유지하는 것 — 200줄짜리 notebook cell 속에 파묻지 않는 것.

Transform 의 네 종류

청소 — 타입 고치기, 날짜 파싱, null 처리, string 정규화.
Derive — 있는 컬럼에서 새 컬럼 계산 (amount_local = amount_usd * fx_rate).
Join — 테이블 결합.
Reshape — pivot, unpivot, group-then-aggregate.

DataFrame 을 받아 DataFrame 을 돌려주는 작은 이름 있는 함수들로 쌓아. raw.pipe(clean).pipe(derive).pipe(join_customers).pipe(reshape) 라고 적힌 파이프라인은 계약서처럼 읽혀. do_everything(raw) 안의 inline 200줄은 안 읽히고.

Code

.pipe() 로 plug 하는 composable transform 함수들·python

import pandas as pd

def clean(df: pd.DataFrame) -> pd.DataFrame:
    return (df
        .assign(
            order_date=lambda d: pd.to_datetime(d['order_date'], errors='raise'),
            amount_usd=lambda d: pd.to_numeric(d['amount_usd'].astype(str).str.replace(',', ''), errors='raise'),
            customer_id=lambda d: d['customer_id'].str.strip().str.upper(),
        )
        .dropna(subset=['order_id', 'customer_id', 'amount_usd'])
        .drop_duplicates('order_id')
    )

def derive(df: pd.DataFrame) -> pd.DataFrame:
    return df.assign(
        month=lambda d: d['order_date'].dt.to_period('M'),
        amount_local=lambda d: d['amount_usd'] * d.get('fx_rate', 1.0),
        is_high_value=lambda d: d['amount_usd'] > 500,
    )

def attach_customers(df: pd.DataFrame, customers: pd.DataFrame) -> pd.DataFrame:
    return df.merge(
        customers[['customer_id', 'country', 'tier']],
        on='customer_id', how='left', validate='many_to_one',
    )

result = (
    raw
      .pipe(clean)
      .pipe(derive)
      .pipe(attach_customers, customers=customers)
)

Exercise

본인이 쓴 다단계 transformation 하나를 골라 DataFrame 을 받고 돌려주는 이름 있는 함수 3–4개로 쪼개 봐. .pipe() 로 다시 이어 붙이고, 함수마다 10 row 짜리 합성 입력으로 pytest 테스트를 하나씩 달아. 이 테스트 suite 가 바로 겁 없이 리팩터링하게 해 주는 산출물이야.

Transform — 청소, cast, derive

로직이 사는 곳

Transform 의 네 종류

Code

External links

Exercise

Progress

댓글 0