Prefect — The Pythonic Alternative

import pandas as pd
from prefect import flow, task, get_run_logger

@task(retries=3, retry_delay_seconds=30)
def extract(ds: str) -> pd.DataFrame:
    log = get_run_logger()
    df = pd.read_csv(f'source/orders_{ds}.csv', dtype=str)
    log.info('extracted %d rows', len(df))
    return df

@task
def transform(raw: pd.DataFrame) -> pd.DataFrame:
    return (raw
        .assign(amount_usd=lambda d: pd.to_numeric(d['amount_usd']))
        .dropna(subset=['amount_usd']))

@task
def load(clean: pd.DataFrame, ds: str) -> str:
    path = f'warehouse/orders/date={ds}/part-0.parquet'
    clean.to_parquet(path, index=False)
    return path

@flow(name='orders_pipeline')
def orders_pipeline(ds: str = '2026-04-30') -> str:
    raw = extract(ds)
    clean = transform(raw)
    return load(clean, ds)

if __name__ == '__main__':
    orders_pipeline('2026-04-30')