pipe()를 사용한 메서드 연결

중간 변수의 문제점

import pandas as pd

df = pd.read_csv('orders.csv')

# Without pipe — intermediate variables everywhere
df1 = df.dropna(subset=['revenue'])
df2 = df1[df1['quantity'] > 0]
df3 = df2.assign(revenue_per_unit=df2['revenue'] / df2['quantity'])
print(df3.shape)

pipe() 소개

def drop_nulls(df):
    return df.dropna(subset=['revenue'])

def filter_positive_qty(df):
    return df[df['quantity'] > 0]

def add_revenue_per_unit(df):
    return df.assign(revenue_per_unit=df['revenue'] / df['quantity'])

# With pipe — clean chain
df_clean = (df
    .pipe(drop_nulls)
    .pipe(filter_positive_qty)
    .pipe(add_revenue_per_unit)
)
print(df_clean.shape)

pipe()로 인수 전달하기

def filter_by_region(df, regions):
    return df[df['region'].isin(regions)]

def cap_revenue(df, upper):
    df = df.copy()
    df['revenue'] = df['revenue'].clip(upper=upper)
    return df

df_result = (df
    .pipe(filter_by_region, regions=['North', 'East'])
    .pipe(cap_revenue, upper=1000)
)
print(df_result.shape)

pipe()와 기본 메서드 함께 사용하기

df_result = (
    df
    .dropna(subset=['revenue', 'order_date'])
    .query('quantity > 0')
    .rename(columns={'unit_price': 'price'})
    .pipe(add_revenue_per_unit)
    .reset_index(drop=True)
)
print(df_result.head())

pipe() 체인 디버깅

def debug(df, label=''):
    print(f'[{label}] shape: {df.shape}')
    print(f'[{label}] columns: {df.columns.tolist()}')
    return df

df_result = (
    df
    .pipe(drop_nulls)
    .pipe(debug, label='after drop_nulls')
    .pipe(filter_positive_qty)
    .pipe(debug, label='after filter')
)
print('Done')

전체 정리 파이프라인 구축

def clean_pipeline(df):
    return (
        df
        .dropna(subset=['order_id', 'revenue'])
        .drop_duplicates(subset=['order_id'])
        .query('quantity > 0 and revenue >= 0')
        .pipe(add_revenue_per_unit)
        .reset_index(drop=True)
    )

df_clean = clean_pipeline(df)
print('Clean rows:', len(df_clean))

pipe()와 apply()의 주요 차이점

# pipe: receives the whole DataFrame
def scale_revenue(df, factor=1.0):
    df = df.copy()
    df['revenue'] = df['revenue'] * factor
    return df

# apply: receives one row at a time
df['revenue_x2'] = df.apply(lambda row: row['revenue'] * 2, axis=1)

df_scaled = df.pipe(scale_revenue, factor=1.1)
print('pipe scales all rows at once; apply does row-by-row')

재사용 가능한 파이프라인 구성 요소

def normalise_strings(df, cols):
    df = df.copy()
    for col in cols:
        df[col] = df[col].str.strip().str.lower()
    return df

def parse_dates(df, cols):
    df = df.copy()
    for col in cols:
        df[col] = pd.to_datetime(df[col])
    return df

df_result = (
    df
    .pipe(normalise_strings, cols=['region', 'category'])
    .pipe(parse_dates, cols=['order_date'])
)
print(df_result.dtypes)

pipe()로 파이프라인 단계 기록하기

import logging
logging.basicConfig(level=logging.INFO)

def logged_dropna(df, subset):
    before = len(df)
    df = df.dropna(subset=subset)
    after = len(df)
    logging.info(f'dropna: {before - after} rows removed, {after} remaining')
    return df

df_clean = df.pipe(logged_dropna, subset=['revenue'])
print('Logged pipeline complete.')

파이프라인의 조건부 단계

def maybe_cap_revenue(df, cap=None):
    if cap is None:
        return df
    df = df.copy()
    df['revenue'] = df['revenue'].clip(upper=cap)
    return df

CAPPING_ENABLED = True
CAP_VALUE = 1000 if CAPPING_ENABLED else None

df_result = df.pipe(maybe_cap_revenue, cap=CAP_VALUE)
print('Conditional step applied:', CAPPING_ENABLED)

파이프라인 결과 내보내기

def save_parquet(df, path):
    df.to_parquet(path, index=False)
    print(f'Saved {len(df)} rows to {path}')
    return df  # return df so the chain can continue if needed

df_final = (
    df
    .pipe(clean_pipeline)
    .pipe(save_parquet, path='orders_final.parquet')
)
print('Pipeline complete.')

빠른 확인

단원 요약