Building an ETL Pipeline with Python

This is sample teaching content. Publication dates and transaction records are illustrative.

An ETL pipeline extracts input, transforms it, and loads a destination. For a first exercise, use local files so you can inspect every result. This Python 3.13 example reads transaction_id and amount columns and writes accepted rows as JSON Lines. Amounts use one assumed currency.

Make a small input

Save this as transactions.csv in a new exercise directory:

transaction_id,amount
T-10,12.50
T-11,invalid
T-12,0
T-13,NaN

Zero is valid. NaN is not a useful transaction amount, even though Decimal can parse it. Check finiteness as well as conversion.

Transform with visible rejection

import csv
import json
from decimal import Decimal, InvalidOperation
from pathlib import Path

accepted, rejected = [], []
with open('transactions.csv', newline='', encoding='utf-8') as source:
    reader = csv.DictReader(source)
    if reader.fieldnames != ['transaction_id', 'amount']:
        raise ValueError('Expected transaction_id,amount header')
    for line, row in enumerate(reader, start=2):
        try:
            identifier = (row['transaction_id'] or '').strip()
            amount = Decimal(row['amount'] or '')
            if not identifier or not amount.is_finite():
                raise ValueError('Missing identifier or non-finite amount')
            accepted.append({'transaction_id': identifier,
                             'amount': str(amount)})
        except (InvalidOperation, ValueError):
            rejected.append({'line': line, 'reason': 'invalid fields'})

payload = ''.join(json.dumps(row) + '\n' for row in accepted)
Path('accepted.jsonl').write_text(payload, encoding='utf-8')
Path('rejected.json').write_text(json.dumps(rejected), encoding='utf-8')
print({'accepted': len(accepted), 'rejected': len(rejected)})

Run the file with Python in the exercise directory. Expect two accepted rows and two rejected rows. Amounts become strings so serialization does not convert exact decimal values to binary floating point. A downstream loader must cast those strings to its own fixed-precision numeric type.

Know the boundary

Opening CSV with newline='' lets the CSV module handle quoted newlines correctly. Catch only expected field failures; a missing file or permission error should stop the run. The example keeps all rows in memory and overwrites output, so it fits small exercises rather than a large production feed.

For production, add file identity, row-width checks, a currency/scale contract, durable rejection storage, and an atomic publication step. Never publish a half-written output as a complete batch. Deduplicate business identifiers separately: this transform does not enforce uniqueness.