使用 pipe() 链式调用方法
使用 pipe() 将自定义函数与 Pandas 原生方法串联起来，编写易读的数据转换 pipeline。
使用 pipe() 链式调用方法 是 CoddyKit 上的免费 Pandas & NumPy Academy 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Pandas & NumPy Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Pandas & NumPy Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
The Problem with Intermediate Variables
A data cleaning pipeline without pipe() often accumulates many intermediate variables: df1 = clean(df), df2 = transform(df1), df3 = enrich(df2). These variables clutter the namespace, make debugging harder, and tempt developers to reuse them incorrectly. The result is code that is hard to read top-to-bottom as a sequence of transformations.
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)Introducing pipe()
DataFrame.pipe(func) calls func(df) and returns the result, allowing you to chain custom functions in the same way you chain native Pandas methods like .dropna().query(). The key benefit is that every transformation step is explicit and readable left-to-right (or top-to-bottom when formatted with parentheses), mirroring the logical order of the pipeline.
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)Passing Arguments Through pipe()
Pass extra arguments to the piped function using keyword arguments after the function name: df.pipe(func, arg1=val1). The function signature must accept df as its first parameter. Parameterised functions make the pipeline configurable: you can change thresholds, column names, or behaviour without modifying the function body, just by changing the pipe call arguments.
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)Mixing pipe() with Native Methods
The power of pipe() is that it integrates seamlessly with native Pandas methods in the same chain. You can mix .dropna(), .query(), .rename(), and .pipe(custom_func) in any order. This makes the chain both concise (using built-in methods where possible) and flexible (using custom functions where built-ins fall short).
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())Debugging a pipe() Chain
Debugging a long chain can be tricky because you cannot inspect intermediate states by adding a print statement in the middle. One solution is to write a passthrough debug function that prints shape and column info and then returns the DataFrame unchanged. Insert it at any point in the chain to inspect the state at that step without breaking the chain.
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')Building a Full Cleaning Pipeline
Combine all cleaning steps into a single pipeline function using pipe(). Wrapping the chain in a function called clean_pipeline(df) makes the pipeline testable as a unit. Call it with a raw DataFrame and receive a clean one. This pattern aligns with the ETL (Extract, Transform, Load) paradigm used in production data engineering.
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() vs. apply(): Key Differences
pipe(func) passes the entire DataFrame to func and expects a DataFrame (or transformed object) back. apply(func, axis=1) passes one row at a time. Use pipe() for whole-DataFrame transformations that maintain the same shape (or deliberately change it), and use apply() for row or column level calculations. They are complementary, not competing.
# 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')Reusable Pipeline Components
Write each pipeline step as a pure function — no global state, receives a DataFrame, returns a DataFrame. Pure functions are easy to unit-test: call with a small test DataFrame and assert the output shape and column values. A library of tested, reusable pipeline functions dramatically speeds up analysis of new datasets that share similar cleaning requirements.
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)Logging Pipeline Steps with pipe()
Add structured logging inside each pipeline function so you can audit every transformation in production. Include the input row count, output row count, and any relevant stats (e.g. rows dropped by a filter). This gives you a complete trace of every pipeline run without needing an external workflow orchestrator for basic audit trails.
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.')Conditional Steps in a Pipeline
Sometimes a cleaning step should only run based on a flag or the data content. You can add conditional steps to a pipe chain by inserting an identity-or-transform function: it checks a condition and either applies a transformation or returns the DataFrame unchanged. This keeps the chain structure intact while supporting optional steps.
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)Exporting the Pipeline Result
The final step in a pipe() chain is often an export. You can chain a custom export function using pipe() or simply call the native Pandas export method after the chain. Using a pipe(save_to_parquet) step keeps the export part of the chain documentation and ensures it always runs on the final cleaned DataFrame, not an intermediate version.
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.')Quick Check
Test your understanding of Data Analysis concepts from this lesson.
Lesson Recap
In this lesson you learned: using pipe() to chain custom functions with native Pandas methods, building reusable, parameterised, and testable pipeline steps as pure functions, and adding debug logging and conditional steps inside a pipe chain. Next up we explore structuring transformation steps as functions for a production ETL pipeline.
常见问题解答
「使用 pipe() 链式调用方法」课时是免费的吗？
是的 — 「使用 pipe() 链式调用方法」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Pandas & NumPy Academy 课程的其余内容，请升级到 CoddyKit PRO。 Pandas & NumPy Academy 课程共包含 4 节课。
「使用 pipe() 链式调用方法」这节课中我会学到什么？
使用 pipe() 将自定义函数与 Pandas 原生方法串联起来，编写易读的数据转换 pipeline。 你通过在浏览器中直接运行的动手代码来练习 Pandas & NumPy Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 Pandas & NumPy Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 Pandas & NumPy Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「使用 pipe() 链式调用方法」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 Pandas & NumPy Academy 课中编写并运行代码吗？
能。每节 Pandas & NumPy Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 对列和行使用 apply()
- 对 GroupBy 使用 apply()
- 用于逐元素操作的 map() 和 applymap()
- 使用 pipe() 链式调用方法