変換手順の関数化
ノートブックを抽出、変換、読み込みの各関数に分割し、それぞれがDataFrameを受け取り返す構成にして、簡単にテストできるようにします。
「変換手順の関数化」はCoddyKit上の無料Pandas & NumPy Academyレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはPandas & NumPy Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Pandas & NumPy Academyコースには全4レッスンが含まれています。
このレッスンの一部はまだ翻訳されておらず、英語で表示されています。
Moving Beyond Notebooks
Jupyter notebooks are great for exploration but poor for production data pipelines. Code scattered across 50 cells with global state and no tests is fragile: changing one cell can silently break another. The professional alternative is to extract each transformation into a named function that accepts a DataFrame and returns a DataFrame. This separation of concerns is the foundation of maintainable data engineering code.
# Notebook-style (fragile)
df = pd.read_csv('orders.csv')
df = df.dropna(subset=['revenue'])
df = df[df['quantity'] > 0]
df['revenue_per_unit'] = df['revenue'] / df['quantity']
# Function-style (robust)
def extract(path):
return pd.read_csv(path)
def transform(df):
return (
df.dropna(subset=['revenue'])
.query('quantity > 0')
.assign(revenue_per_unit=lambda d: d['revenue'] / d['quantity'])
)
df = transform(extract('orders.csv'))The ETL Pattern: Extract, Transform, Load
The ETL pattern divides a data pipeline into three phases: Extract (read from source), Transform (clean and enrich), and Load (write to destination). Each phase is a separate function. This separation makes it easy to swap data sources (CSV vs. database), change cleaning logic, or change the output format without touching the other two phases. Every production pipeline should follow this structure.
def extract(config):
return pd.read_csv(config['input_path'], parse_dates=['order_date'])
def transform(df, config):
return (
df
.dropna(subset=config['required_cols'])
.query('quantity > 0')
.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
)
def load(df, config):
df.to_parquet(config['output_path'], index=False)
print(f'Saved {len(df)} rows.')Single Responsibility Principle
Each transformation function should do exactly one thing. A function called clean_data() that drops nulls, caps outliers, parses dates, and encodes categories is hard to test and debug. Instead, write drop_nulls(), cap_outliers(), parse_dates(), and encode_categories() as separate functions. This granularity makes it easy to skip, replace, or reorder any single step.
def drop_null_rows(df, required_cols):
return df.dropna(subset=required_cols)
def remove_returns(df):
return df[df['quantity'] > 0]
def compute_revenue(df):
return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
def add_date_features(df):
return df.assign(
year=lambda d: d['order_date'].dt.year,
month=lambda d: d['order_date'].dt.month
)Chaining Steps with pipe()
Connect single-responsibility functions using pipe() to build the full transform phase as a readable chain. The chain reads like a recipe: each line is a step, and the arrow of data flows top to bottom. Any step can be commented out or reordered without renaming variables. The final result is the clean, enriched DataFrame ready for the load phase.
import pandas as pd
REQUIRED = ['order_id', 'revenue', 'order_date']
def transform(raw_df):
return (
raw_df
.pipe(drop_null_rows, required_cols=REQUIRED)
.pipe(remove_returns)
.pipe(compute_revenue)
.pipe(add_date_features)
)
df_clean = transform(pd.read_csv('orders.csv', parse_dates=['order_date']))
print(df_clean.shape)Returning Row Counts for Audit
Each transformation function should optionally log how many rows it received and how many it returned. Wrapping the function body in a before/after row count log is a lightweight audit trail that lets you quickly see where rows are being dropped in a pipeline run. Store these counts in a list and print them as a report at the end of each run.
audit_log = []
def audited(func):
def wrapper(df, *args, **kwargs):
before = len(df)
result = func(df, *args, **kwargs)
after = len(result)
audit_log.append({'step': func.__name__, 'in': before, 'out': after, 'dropped': before - after})
return result
return wrapper
@audited
def drop_null_rows(df, required_cols):
return df.dropna(subset=required_cols)Testable Functions with Small DataFrames
The biggest advantage of named functions is testability. Write a small test DataFrame that represents a realistic edge case and assert the function's output. Test the drop_null_rows function with one row that has a null and verify it is dropped, and another without a null and verify it is kept. Unit tests on transformation functions catch regressions when the pipeline code changes.
import pandas as pd
def test_drop_null_rows():
test_df = pd.DataFrame({
'order_id': [1, 2, 3],
'revenue': [100.0, None, 200.0]
})
result = drop_null_rows(test_df, required_cols=['revenue'])
assert len(result) == 2, 'Should have 2 non-null rows'
assert result['revenue'].isna().sum() == 0, 'No nulls in revenue'
print('test_drop_null_rows PASSED')
test_drop_null_rows()Organising Functions into Modules
As the pipeline grows, split functions into Python module files: extract.py, transform.py, load.py, and validate.py. The main pipeline.py script imports and orchestrates them. This file structure is readable, testable with pytest, and deployable as a Python package. It mirrors the standard layout used by data engineering teams using tools like dbt or Airflow.
# pipeline.py
# from extract import extract_orders
# from transform import transform
# from load import load_to_parquet
# from validate import validate_sales_df
# def run_pipeline(config):
# raw = extract_orders(config)
# clean = transform(raw, config)
# validate_sales_df(clean)
# load_to_parquet(clean, config)
print('Module-based pipeline structure shown above (imports commented for demo)')Idempotency: Run Multiple Times Safely
A well-designed pipeline function is idempotent: running it twice on the same input produces the same output and does not cause side effects. Avoid in-place mutations (df.drop(..., inplace=True)) and always use df.copy() at the start of functions that modify columns. Idempotent functions can be re-run after failures without corrupting the output data.
def add_revenue_flag(df, threshold=500):
# Use copy to avoid mutating the input
df = df.copy()
df['is_large_order'] = df['revenue'] >= threshold
return df
# Running twice gives the same result
df1 = add_revenue_flag(df_clean)
df2 = add_revenue_flag(df_clean)
assert df1.equals(df2), 'Function is not idempotent!'
print('Idempotency check passed.')Type Hints for Self-Documentation
Add Python type hints to transformation function signatures to make the contract explicit: def transform(df: pd.DataFrame) -> pd.DataFrame. Type hints are self-documenting — any developer reading the function knows exactly what it expects and returns without reading the body. They also enable static analysis tools like mypy and IDE auto-complete to catch type errors before runtime.
from typing import List
def drop_null_rows(df: pd.DataFrame, required_cols: List[str]) -> pd.DataFrame:
return df.dropna(subset=required_cols)
def compute_revenue(df: pd.DataFrame) -> pd.DataFrame:
return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
print('Type-annotated functions ready for production use.')Documenting Each Step with Docstrings
Each transformation function should have a one-line docstring stating what it does, what columns it requires, and what columns it adds or removes. Good docstrings make the function discoverable via help() and in IDE tooltips. They also serve as specification documents: a failing assertion that contradicts the docstring is a bug; a docstring that disagrees with the code is a documentation error.
def compute_revenue(df: pd.DataFrame) -> pd.DataFrame:
"""Add 'revenue' column as quantity * unit_price.
Requires: 'quantity' (numeric), 'unit_price' (numeric) columns.
Returns: df with new 'revenue' float column appended.
"""
return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])
help(compute_revenue)Running the Full Pipeline
Orchestrate the complete ETL by calling the three phases in sequence: extract, transform, and load. Add timing around each phase to measure where time is spent. Catch exceptions from each phase separately so error messages identify which phase failed. Log the start and end time of the full run and whether it succeeded or failed for monitoring purposes.
import time
CONFIG = {
'input_path': 'orders.csv',
'output_path': 'orders_clean.parquet',
'required_cols': ['order_id', 'revenue']
}
t0 = time.time()
raw = extract(CONFIG)
clean = transform(raw, CONFIG)
load(clean, CONFIG)
print(f'Pipeline completed in {time.time()-t0:.1f}s')
print(f'Audit log: {audit_log}')Quick Check
Test your understanding of Data Analysis concepts from this lesson.
Lesson Recap
In this lesson you learned: structuring pipelines as ETL functions with single responsibility, making functions testable, idempotent, and self-documenting with type hints and docstrings, and orchestrating the full pipeline with timing and audit logging. Next up we explore parameterising pipelines with config dicts for reusability across different datasets.
よくある質問
「変換手順の関数化」レッスンは無料ですか？
はい。「変換手順の関数化」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Pandas & NumPy Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Pandas & NumPy Academyコースには全4レッスンが含まれています。
「変換手順の関数化」で何を学びますか？
ノートブックを抽出、変換、読み込みの各関数に分割し、それぞれがDataFrameを受け取り返す構成にして、簡単にテストできるようにします。 ブラウザで直接実行するハンズオンコードでPandas & NumPy Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
Pandas & NumPy Academyを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのPandas & NumPy Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。
「変換手順の関数化」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このPandas & NumPy Academyレッスンでコードを書いて実行できますか？
はい。すべてのPandas & NumPy Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。