Цепочки методов с pipe()
Пишите понятные конвейеры преобразования данных, используя pipe() для объединения пользовательских функций со встроенными методами Pandas.
«Цепочки методов с pipe()» — бесплатный урок Pandas & NumPy Academy на CoddyKit. Это урок 4 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения Pandas & NumPy Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс Pandas & NumPy Academy содержит 4 уроков всего.
Проблема промежуточных переменных
В конвейере очистки данных без pipe() часто накапливается множество промежуточных переменных: df1 = clean(df), df2 = transform(df1), df3 = enrich(df2). Эти переменные загромождают пространство имён, усложняют отладку и провоцируют разработчиков использовать их повторно неправильно. В результате код трудно читать сверху вниз как последовательность преобразований.
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()
DataFrame.pipe(func) вызывает func(df) и возвращает результат, позволяя объединять пользовательские функции в цепочку так же, как встроенные методы Pandas, например .dropna().query(). Главное преимущество состоит в том, что каждый шаг преобразования явно обозначен и читается слева направо (или сверху вниз при оформлении в скобках), отражая логический порядок конвейера.
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()
Передавайте дополнительные аргументы в функцию конвейера с помощью именованных аргументов после имени функции: df.pipe(func, arg1=val1). Сигнатура функции должна принимать df первым параметром. Параметризованные функции делают конвейер настраиваемым: можно изменять пороги, имена столбцов или поведение, не меняя тело функции, а только аргументы вызова 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() со встроенными методами
Преимущество pipe() в том, что он бесшовно объединяется со встроенными методами Pandas в одной цепочке. Можно комбинировать .dropna(), .query(), .rename() и .pipe(custom_func) в любом порядке. Благодаря этому цепочка одновременно остаётся краткой (где возможно, используются встроенные методы) и гибкой (пользовательские функции применяются там, где встроенных методов недостаточно).
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()
Отлаживать длинную цепочку может быть сложно, поскольку нельзя вставить оператор печати в середину и проверить промежуточные состояния. Одно из решений — написать сквозную функцию отладки, которая выводит форму и сведения о столбцах, а затем возвращает DataFrame без изменений. Вставьте её в любую точку цепочки, чтобы проверить состояние на этом шаге, не прерывая цепочку.
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')Создание полноценного конвейера очистки
Объедините все шаги очистки в одну функцию конвейера с помощью pipe(). Если обернуть цепочку в функцию clean_pipeline(df), конвейер можно будет тестировать как единое целое. Передайте ей исходный DataFrame и получите очищенный. Этот шаблон соответствует парадигме ETL (извлечение, преобразование, загрузка), используемой в промышленной инженерии данных.
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(func) передаёт функции func весь DataFrame и ожидает получить обратно DataFrame (или преобразованный объект). apply(func, axis=1) передаёт по одной строке за раз. Используйте pipe() для преобразований всего DataFrame, сохраняющих его форму (или намеренно изменяющих её), а 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')Повторно используемые компоненты конвейера
Оформляйте каждый шаг конвейера как чистую функцию: без глобального состояния, принимающую DataFrame и возвращающую DataFrame. Чистые функции легко проверять модульными тестами: передайте небольшой тестовый DataFrame и проверьте форму результата и значения столбцов. Библиотека проверенных и повторно используемых функций конвейера значительно ускоряет анализ новых наборов данных с похожими требованиями к очистке.
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.')Условные шаги в конвейере
Иногда шаг очистки должен выполняться только при определённом флаге или содержимом данных. Можно добавить условные шаги в цепочку pipe, вставив функцию, которая либо возвращает данные без изменений, либо выполняет преобразование: она проверяет условие и выбирает нужное действие. Это сохраняет структуру цепочки и одновременно поддерживает необязательные шаги.
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)Экспорт результата конвейера
Последним шагом в цепочке pipe() часто бывает экспорт. Пользовательскую функцию экспорта можно добавить в цепочку с помощью pipe() или просто вызвать встроенный метод экспорта Pandas после завершения цепочки. Шаг pipe(save_to_parquet) сохраняет экспорт в документации цепочки и гарантирует, что он всегда выполняется для окончательного очищенного DataFrame, а не для промежуточной версии.
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.')Быстрая проверка
Проверьте своё понимание концепций анализа данных из этого урока.
Итоги урока
В этом уроке Вы узнали, как использовать pipe() для объединения пользовательских функций со встроенными методами Pandas, создавать повторно используемые, параметризованные и тестируемые шаги конвейера в виде чистых функций, а также добавлять отладочное журналирование и условные шаги в цепочку pipe. Далее мы рассмотрим, как оформлять шаги преобразования в виде функций для производственного ETL-конвейера.
Часто задаваемые вопросы
Урок «Цепочки методов с pipe()» бесплатный?
Да — полный текст урока «Цепочки методов с pipe()» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс Pandas & NumPy Academy, подпишись на CoddyKit PRO. Курс Pandas & NumPy Academy содержит 4 уроков всего.
Чему я научусь в уроке «Цепочки методов с pipe()»?
Пишите понятные конвейеры преобразования данных, используя pipe() для объединения пользовательских функций со встроенными методами Pandas. Ты практикуешь Pandas & NumPy Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать Pandas & NumPy Academy?
Предыдущий опыт не требуется. Pandas & NumPy Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 4 из 4.
Сколько времени занимает урок «Цепочки методов с pipe()»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке Pandas & NumPy Academy?
Да. Каждый урок Pandas & NumPy Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- apply() для столбцов и строк
- apply() с GroupBy
- map() и applymap() для поэлементных операций
- Цепочки методов с pipe()