pipe()によるメソッドチェーン
pipe()を使ってカスタム関数とPandasの標準メソッドをつなぎ、読みやすいデータ変換パイプラインを記述します。
「pipe()によるメソッドチェーン」はCoddyKit上の無料Pandas & NumPy Academyレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これは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) を呼び出して結果を返します。これにより、.dropna().query() のような Pandas 標準メソッドと同じように、カスタム関数をチェーンできます。主な利点は、各変換ステップが左から右へ（かっこで整形した場合は上から下へ）明示的かつ読みやすく並ぶことです。パイプラインの論理的な順序もそのまま反映されます。
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()チェーンのデバッグ
長いチェーンのデバッグは、途中にprint文を追加して中間状態を確認できないため、難しい場合があります。解決策の一つは、形状と列の情報を出力した後、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を渡すと、クリーニング済みのDataFrameが返されます。このパターンは、実運用のデータエンジニアリングで使われるETL（Extract、Transform、Load）パラダイムに沿ったものです。
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)はDataFrame全体をfuncに渡し、DataFrame（または変換後のオブジェクト）が返されることを想定します。apply(func, axis=1)は、1行ずつ関数に渡します。同じ形状を保つ（または意図的に変更する）DataFrame全体の変換には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')再利用可能なパイプラインコンポーネント
パイプラインの各ステップを純粋関数として記述します。つまり、グローバル状態を使わず、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.')パイプライン内の条件付きステップ
クリーニング手順は、フラグやデータの内容に応じて実行するかどうかを決めたい場合があります。条件を確認し、条件を満たす場合は変換を適用し、そうでない場合はDataFrameを変更せずに返すアイデンティティまたは変換関数を挿入することで、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ではなく、最終的にクリーニングされた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時間対応のAIチューター）、Pandas & NumPy Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Pandas & NumPy Academyコースには全4レッスンが含まれています。
「pipe()によるメソッドチェーン」で何を学びますか？
pipe()を使ってカスタム関数とPandasの標準メソッドをつなぎ、読みやすいデータ変換パイプラインを記述します。 ブラウザで直接実行するハンズオンコードでPandas & NumPy Academyを演習し、24時間対応の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()によるメソッドチェーン