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

Transform 이 로직 사는 곳

Extract 가 byte 를 끌어오고, load 가 쓰고, transform 이 본인 팀이 존재하는 이유의 로직이 앉는 곳. 규율은 모든 transformation 을 명시적, 명명된, 테스트 가능하게 만드는 거 — 200줄 짜리 notebook cell 에 묻혀있지 않게.

Transform 의 네 종류

청소 — type 고치기, 날짜 파싱, 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) 인 파이프라인이 contract 처럼 읽히고; do_everything(raw) 의 200 줄 inline 연산은 안 그래.

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 개로 split. .pipe() 로 compose. 그리고 각 함수에 대해 10 row 합성 입력으로 pytest 테스트 하나씩. 테스트 suite 가 두려움 없이 리팩터링 가능하게 만드는 산출물.

Transform — 청소, cast, derive

Transform 이 로직 사는 곳

Transform 의 네 종류

Code

External links

Exercise

Progress

댓글 0