A notebook is for finding things out. A pipeline is for producing the same answer every week without anyone watching. Turning one into the other is mostly four moves: functions with signatures, one entry point, idempotence, and failing in the right place.
Functions with inputs and outputs
import pandas as pd
# A notebook cell: reads a global, writes a global, impossible to test.
#
# df = df[df['status'] != 'cancelled']
# df['revenue'] = df['units'] * df['price']
# total = df.groupby('region')['revenue'].sum()
# The same work as a function: inputs in, value out, nothing shared.
def exclude_cancelled(orders: pd.DataFrame) -> pd.DataFrame:
return orders[orders['status'] != 'cancelled'].copy()
def add_revenue(orders: pd.DataFrame) -> pd.DataFrame:
return orders.assign(revenue=(orders['units'] * orders['price']).round(2))
def revenue_by_region(orders: pd.DataFrame) -> pd.DataFrame:
return (orders.groupby('region', as_index=False, observed=True)['revenue']
.sum().sort_values('revenue', ascending=False))
frame = pd.DataFrame({
'region': ['North', 'South', 'North'],
'status': ['complete', 'cancelled', 'complete'],
'units': [10, 5, 20],
'price': [12.5, 12.5, 11.0],
})
result = revenue_by_region(add_revenue(exclude_cancelled(frame)))
print(result.to_string(index=False))
Open this example in the editor and change it. Nothing you type is saved.
Three properties make those functions worth the extra lines. They take their input as an argument, so they can be called with a two-row fixture in a test. They return a new frame, so the caller's data is unchanged and the order of calls is visible. And they are named after what they do, so the pipeline reads as a description.
.pipe() chains them without the nesting:
import pandas as pd
def exclude(orders, statuses):
return orders[~orders['status'].isin(statuses)].copy()
def add_revenue(orders):
return orders.assign(revenue=(orders['units'] * orders['price']).round(2))
frame = pd.DataFrame({
'region': ['North', 'South', 'North'],
'status': ['complete', 'cancelled', 'complete'],
'units': [10, 5, 20],
'price': [12.5, 12.5, 11.0],
})
result = (frame
.pipe(exclude, statuses={'cancelled', 'test'})
.pipe(add_revenue)
.groupby('region', as_index=False, observed=True)['revenue'].sum())
print(result.to_string(index=False))
Open this example in the editor and change it. Nothing you type is saved.
One entry point
import argparse
import sys
# The shape every pipeline script has.
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description='Monthly revenue report')
parser.add_argument('--input', required=True)
parser.add_argument('--output', required=True)
parser.add_argument('--month', required=True, help='YYYY-MM')
parser.add_argument('--dry-run', action='store_true')
return parser
def main(argv: list[str]) -> int:
args = build_parser().parse_args(argv)
print(f'would read {args.input}, write {args.output}, for {args.month}')
if args.dry_run:
print('dry run: nothing written')
return 0
return 0
# The guard matters: it means the module can be imported by a test
# without running anything.
if __name__ == '__main__':
sys.exit(main(['--input', 'in.csv', '--output', 'out.parquet', '--month', '2026-03', '--dry-run']))
Open this example in the editor and change it. Nothing you type is saved.
Everything the pipeline needs comes in through arguments or environment variables. Nothing is a constant in the middle of the file, and nothing is read from a path that only exists on one laptop.
A --dry-run flag is worth the three lines it costs. It lets you run the whole thing in anger against production inputs without writing anything.
Idempotent, or you will regret it
Running the pipeline twice must give the same result as running it once. Three things break that:
import pandas as pd
history = pd.DataFrame({'month': ['2026-01'], 'revenue': [1000.0]})
new = pd.DataFrame({'month': ['2026-02'], 'revenue': [1200.0]})
# Not idempotent: run twice and February is in there twice.
appended = pd.concat([history, new], ignore_index=True)
appended = pd.concat([appended, new], ignore_index=True)
print('appended twice:', len(appended), appended['revenue'].sum())
# Idempotent: replace the partition, whatever was there before.
def upsert(existing: pd.DataFrame, incoming: pd.DataFrame, key: str) -> pd.DataFrame:
keep = existing[~existing[key].isin(incoming[key])]
return pd.concat([keep, incoming], ignore_index=True).sort_values(key)
once = upsert(history, new, 'month')
twice = upsert(once, new, 'month')
print('upserted twice:', len(twice), twice['revenue'].sum())
print('idempotent:', once.equals(twice))
Open this example in the editor and change it. Nothing you type is saved.
The three fixes, in order of preference:
- Write a whole partition, replacing it. One file per month, overwritten. Re-running the month rewrites that file and touches nothing else.
- Upsert on a key, as above, when appending to one table.
- Make the output path contain the period -
revenue-2026-03.parquet- so a re-run overwrites itself and never doubles.
A pipeline that appends is a pipeline that cannot be re-run after a failure, which is exactly when you need to re-run it.
Fail loudly, in the right place
import pandas as pd
def load(frame: pd.DataFrame) -> pd.DataFrame:
"""Validate at the boundary, so a bad input is rejected here rather
than producing a wrong number three steps later."""
required = {'order_id', 'region', 'units', 'price', 'status'}
missing = required - set(frame.columns)
if missing:
raise ValueError(f'input is missing columns: {sorted(missing)}')
if not frame['order_id'].is_unique:
duplicated = frame.loc[frame['order_id'].duplicated(), 'order_id'].tolist()
raise ValueError(f'duplicate order_id: {duplicated[:5]}')
bad = frame[frame['units'].notna() & (frame['units'] <= 0)]
if len(bad):
raise ValueError(f'{len(bad)} row(s) have non-positive units')
return frame
good = pd.DataFrame({
'order_id': [1, 2], 'region': ['N', 'S'], 'units': [10, 20],
'price': [1.0, 2.0], 'status': ['ok', 'ok'],
})
print('good input:', len(load(good)), 'rows')
for broken, label in [
(good.drop(columns=['price']), 'missing column'),
(pd.concat([good, good.head(1)]), 'duplicate id'),
(good.assign(units=[10, -1]), 'negative units'),
]:
try:
load(broken)
except ValueError as error:
print(f'{label:<16} rejected: {error}')
Open this example in the editor and change it. Nothing you type is saved.
The rule: validate at the boundary, assert in the middle, and never silently continue. A pipeline that drops bad rows without counting them produces a smaller number every week and nobody notices until the quarter.
Logging rather than printing, so the output is usable when it runs unattended:
import logging
import sys
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)-7s %(message)s',
stream=sys.stdout,
)
log = logging.getLogger('pipeline')
log.info('loaded %d rows from %s', 1234, 'orders-2026-03.csv')
log.info('excluded %d cancelled, %d test', 12, 3)
log.warning('%d rows had missing units, imputed with %d', 4, 95)
log.info('wrote %d rows to %s', 1219, 'revenue-2026-03.parquet')
Open this example in the editor and change it. Nothing you type is saved.
Log the counts at every stage. When a number is wrong next month, the log tells you which stage lost the rows, and that is usually the whole debugging session.
Parquet over CSV for anything downstream
import io
import pandas as pd
frame = pd.DataFrame({
'region': pd.Categorical(['North', 'South'] * 50),
'units': pd.array([10, 20] * 50, dtype='int16'),
'placed_on': pd.to_datetime(['2026-01-15'] * 100),
})
csv = frame.to_csv(index=False)
round_tripped = pd.read_csv(io.StringIO(csv))
print('original ', frame.dtypes.to_dict())
print('after csv ', round_tripped.dtypes.to_dict())
print('dtypes survived:', frame.dtypes.equals(round_tripped.dtypes))
Open this example in the editor and change it. Nothing you type is saved.
CSV loses every dtype. The category becomes an object, the int16 becomes int64, the timestamp becomes a string, and the next stage has to guess all three again. Parquet stores them, compresses better, and reads only the columns asked for.
Keep CSV for the thing a human opens in a spreadsheet. Use Parquet between stages.
What stays in the notebook
The division that works:
- Notebook: the question, the exploration, the charts, the narrative. Things you do once.
- Module: loading, cleaning, joining, aggregating. Things you do every time, with tests.
- Script: the entry point that calls the module in order, with arguments and logging.
The notebook then imports the module, which means a fix to the cleaning logic reaches both the exploration and the production run - rather than being made twice and diverging.
A worked example
import io
import logging
import sys
from dataclasses import dataclass
import pandas as pd
logging.basicConfig(level=logging.INFO, format='%(levelname)-7s %(message)s',
stream=sys.stdout, force=True)
log = logging.getLogger('revenue')
EXCLUDED_STATUSES = frozenset({'cancelled', 'test'})
@dataclass(frozen=True)
class Config:
month: str
excluded_statuses: frozenset = EXCLUDED_STATUSES
dry_run: bool = False
# ---- stages: each takes a frame, returns a frame ---------------------
def load(text: str) -> pd.DataFrame:
frame = pd.read_csv(
io.StringIO(text),
dtype={'order_id': 'int64', 'region': 'category', 'status': 'category'},
parse_dates=['placed_on'],
)
required = {'order_id', 'placed_on', 'region', 'units', 'price', 'status'}
missing = required - set(frame.columns)
if missing:
raise ValueError(f'input is missing columns: {sorted(missing)}')
if not frame['order_id'].is_unique:
raise ValueError('order_id is not unique')
log.info('loaded %d rows', len(frame))
return frame
def restrict_to_month(frame: pd.DataFrame, month: str) -> pd.DataFrame:
period = frame['placed_on'].dt.to_period('M').astype(str)
kept = frame[period == month].copy()
log.info('month %s: kept %d of %d rows', month, len(kept), len(frame))
if kept.empty:
raise ValueError(f'no rows for {month}')
return kept
def exclude_statuses(frame: pd.DataFrame, statuses: frozenset) -> pd.DataFrame:
kept = frame[~frame['status'].isin(statuses)].copy()
log.info('excluded %d row(s) by status', len(frame) - len(kept))
return kept
def impute_units(frame: pd.DataFrame) -> pd.DataFrame:
missing = frame['units'].isna()
if not missing.any():
return frame.assign(units_imputed=False)
median = round(float(frame['units'].median()))
log.warning('%d row(s) missing units, imputed with %d', int(missing.sum()), median)
return frame.assign(units_imputed=missing,
units=frame['units'].fillna(median).astype('int64'))
def add_revenue(frame: pd.DataFrame) -> pd.DataFrame:
out = frame.assign(revenue=(frame['units'] * frame['price']).round(2))
if (out['revenue'] <= 0).any():
raise ValueError('a revenue is not positive')
return out
def summarise(frame: pd.DataFrame) -> pd.DataFrame:
summary = (frame.groupby('region', as_index=False, observed=True)
.agg(orders=('order_id', 'size'),
units=('units', 'sum'),
revenue=('revenue', 'sum'))
.sort_values('revenue', ascending=False)
.round(2))
total = round(float(frame['revenue'].sum()), 2)
assert abs(float(summary['revenue'].sum()) - total) < 0.01, 'groups do not sum to the total'
log.info('summarised into %d region(s), total %.2f', len(summary), total)
return summary
def upsert(existing: pd.DataFrame, incoming: pd.DataFrame, keys: list[str]) -> pd.DataFrame:
"""Idempotent: the same incoming rows replace, never duplicate."""
index = pd.MultiIndex.from_frame(incoming[keys])
keep = existing[~pd.MultiIndex.from_frame(existing[keys]).isin(index)]
return (pd.concat([keep, incoming], ignore_index=True)
.sort_values(keys).reset_index(drop=True))
# ---- the entry point -------------------------------------------------
def run(text: str, config: Config, store: pd.DataFrame) -> pd.DataFrame:
summary = (
load(text)
.pipe(restrict_to_month, month=config.month)
.pipe(exclude_statuses, statuses=config.excluded_statuses)
.pipe(impute_units)
.pipe(add_revenue)
.pipe(summarise)
.assign(month=config.month)
)
if config.dry_run:
log.info('dry run: nothing written')
return store
return upsert(store, summary, keys=['month', 'region'])
RAW = """order_id,placed_on,region,units,price,status
1001,2026-02-15,North,120,12.50,complete
1002,2026-02-16,South,95,12.50,complete
1003,2026-02-16,North,,11.00,complete
1004,2026-02-01,East,60,14.00,cancelled
1005,2026-02-03,West,45,13.25,complete
1006,2026-03-14,North,143,11.00,complete
1007,2026-02-20,South,72,12.00,test
"""
store = pd.DataFrame(columns=['month', 'region', 'orders', 'units', 'revenue'])
print('--- first run ---')
store = run(RAW, Config(month='2026-02'), store)
print(store.to_string(index=False))
print()
print('--- same run again ---')
rows_before = len(store)
total_before = float(store['revenue'].sum())
store = run(RAW, Config(month='2026-02'), store)
print(f'rows {rows_before} -> {len(store)}, '
f'total {total_before:,.2f} -> {float(store["revenue"].sum()):,.2f}')
print('idempotent:', len(store) == rows_before)
print()
print('--- the next month ---')
store = run(RAW, Config(month='2026-03'), store)
print(store.to_string(index=False))
print()
print('--- a month with no data ---')
try:
run(RAW, Config(month='2026-09'), store)
except ValueError as error:
print('rejected:', error)
Open this example in the editor and change it. Nothing you type is saved.
Six properties make that a pipeline rather than a notebook. Every stage is a function with a signature and no globals. The configuration is one frozen object. The counts are logged at every stage. The boundary validates and raises rather than dropping rows quietly. The write is an upsert, so the second run changes nothing. And the empty month raises instead of writing an empty summary that would look like a bad month.
When it goes wrong
| Symptom | Cause |
|---|---|
| Re-running doubled the numbers | The write appends; upsert or replace a partition |
| A failure left half the output written | Write to a temporary path and rename |
| Rows disappeared and nobody noticed | Silent drops; log the count at every stage |
| The script only runs on one machine | Hard-coded paths; pass them as arguments |
| Dtypes are re-guessed at every stage | CSV between stages; use Parquet |
| A test needs the real database | The function reads a global; pass the frame in |
| An empty input produced an empty report | Validate and raise on empty |
| Importing the module ran the pipeline | Missing if __name__ == '__main__' |
A check you can run
Run your pipeline twice in a row against the same input and compare the two outputs byte for byte.
If they differ, it is not idempotent, which means a retry after a failure will corrupt the result. That is a one-command test and it is the single most important property a scheduled job has - more important than its speed, and more important than its correctness on the happy path, because the happy path is not when you will need to re-run it.