Learn how to use Python to build practical, reliable data pipelines. This chapter covers API ingestion, file processing, nested JSON transformation, data quality, aggregation, reusable ETL functions, error handling, retries, and efficient processing at scale.
1. Processing API Responses
Data pipelines often ingest data from REST APIs. A robust client should handle HTTP errors, timeouts, pagination, response parsing, and rate limits rather than assuming every request succeeds.
Make a request and validate the response
import requests
url = "https://api.example.com/orders"
response = requests.get(
url,
params={"status": "complete"},
timeout=20
)
response.raise_for_status()
payload = response.json()
print(payload)This example uses the third-party requests package. A timeout prevents the client from waiting indefinitely, while raise_for_status() raises an exception for unsuccessful HTTP status codes.
Process paginated responses
APIs may return one page at a time. Follow the provider's documented pagination mechanism, which may use a next URL, page number, or cursor.
import requests
def fetch_all_orders(base_url, token):
headers = {"Authorization": f"Bearer {token}"}
next_url = base_url
params = {"limit": 100}
while next_url:
response = requests.get(
next_url,
headers=headers,
params=params,
timeout=20
)
response.raise_for_status()
payload = response.json()
for order in payload.get("results", []):
yield order
# Example cursor-based API response
next_url = payload.get("next")
params = None # next URL already contains pagination details
# for order in fetch_all_orders(api_url, api_token):
# process_order(order)Pagination field names and authentication vary by API. Some APIs return a cursor rather than a complete next URL; adapt the loop to the provider's contract.
Handle request failures
import requests
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
data = response.json()
except requests.Timeout:
print("API request timed out")
except requests.HTTPError as exc:
print(f"API returned an HTTP error: {exc}")
except requests.RequestException as exc:
print(f"Request failed: {exc}")
except ValueError as exc:
print(f"Response was not valid JSON: {exc}")2. Reading and Writing Files
Python's context manager syntax (with) closes files reliably. Specify an encoding for text and process large inputs incrementally when possible.
Read a file line by line
def read_nonempty_lines(path):
with open(path, "r", encoding="utf-8") as file:
for line in file:
line = line.strip()
if line:
yield line
for line in read_nonempty_lines("events.log"):
print(line)Write output incrementally
def write_lines(path, lines):
with open(path, "w", encoding="utf-8") as file:
for line in lines:
file.write(f"{line}\\n")
write_lines("output.txt", ["row-1", "row-2", "row-3"])Read and write JSON
import json
with open("input.json", "r", encoding="utf-8") as file:
records = json.load(file)
with open("output.json", "w", encoding="utf-8") as file:
json.dump(records, file, indent=2, ensure_ascii=False)Use json.load() for a file object and json.loads() for a JSON string. A regular JSON document is usually parsed as a whole, so it may not be appropriate for very large files.
Write JSON Lines (JSONL)
import json
records = [
{"event_id": "e1", "type": "click"},
{"event_id": "e2", "type": "purchase"}
]
with open("events.jsonl", "w", encoding="utf-8") as file:
for record in records:
file.write(json.dumps(record) + "\\n")JSONL stores one JSON value per line and is convenient for streaming, incremental writes, and many data processing tools.
3. Flattening Nested JSON
API payloads often contain nested dictionaries and arrays. Flattening converts selected nested fields into columns while preserving repeated child entities where necessary.
Extract nested fields
order = {
"order_id": 1001,
"customer": {
"id": "C42",
"profile": {"city": "Bengaluru"}
},
"items": [
{"sku": "A1", "quantity": 2},
{"sku": "B7", "quantity": 1}
]
}
order_row = {
"order_id": order["order_id"],
"customer_id": order["customer"]["id"],
"customer_city": order["customer"]["profile"]["city"]
}
print(order_row)Flatten an array into child rows
item_rows = [
{
"order_id": order["order_id"],
"sku": item["sku"],
"quantity": item["quantity"]
}
for item in order.get("items", [])
]
print(item_rows)This produces one row per item, which is often a better relational model than forcing an array of items into a single column. It also means the order-level fields repeat across child rows.
Use pandas.json_normalize for deeper structures
import pandas as pd
payload = [
{
"id": 1,
"customer": {"name": "Asha", "city": "Pune"},
"amount": 250
},
{
"id": 2,
"customer": {"name": "Ravi", "city": "Delhi"},
"amount": 175
}
]
df = pd.json_normalize(payload, sep="_")
print(df)4. Deduplication and Data Validation
Deduplication starts with a clear business key and a rule for choosing the surviving record. Validation checks whether records meet required structural and business rules.
Deduplicate records by key
records = [
{"event_id": "e1", "version": 1, "value": "old"},
{"event_id": "e2", "version": 1, "value": "other"},
{"event_id": "e1", "version": 2, "value": "latest"}
]
latest_by_id = {}
for record in records:
event_id = record["event_id"]
current = latest_by_id.get(event_id)
if current is None or record["version"] > current["version"]:
latest_by_id[event_id] = record
deduplicated = list(latest_by_id.values())
print(deduplicated)This example keeps the highest version per event ID. If the source uses timestamps, define how ties are resolved and what should happen when a timestamp is missing.
Validate required fields and data types
def validate_order(record):
required = ("order_id", "customer_id", "amount")
missing = [key for key in required if key not in record]
if missing:
return False, f"Missing fields: {missing}"
if not isinstance(record["amount"], (int, float)):
return False, "amount must be numeric"
if record["amount"] < 0:
return False, "amount cannot be negative"
return True, None
valid, error = validate_order({
"order_id": 1001,
"customer_id": "C42",
"amount": 250.0
})bool is a subclass of int, so the numeric check above also accepts True and False. If booleans must be rejected, explicitly check not isinstance(amount, bool). Consider decimals for precise monetary calculations.Separate valid and invalid records
valid_records = []
rejected_records = []
for record in records:
is_valid, reason = validate_order(record)
if is_valid:
valid_records.append(record)
else:
rejected_records.append({
"record": record,
"reason": reason
})For production pipelines, persist rejected records with a reason and source metadata so issues can be investigated and replayed.
5. Grouping Records and Aggregations
Grouping is common in ETL jobs: count events, calculate totals, identify the latest event per entity, or prepare summary tables.
Aggregate with defaultdict
from collections import defaultdict
sales_by_region = defaultdict(float)
sales = [
{"region": "West", "amount": 120.0},
{"region": "East", "amount": 80.0},
{"region": "West", "amount": 40.0}
]
for row in sales:
sales_by_region[row["region"]] += row["amount"]
print(dict(sales_by_region))
# {'West': 160.0, 'East': 80.0}Count categories with Counter
from collections import Counter
statuses = ["SUCCESS", "FAILED", "SUCCESS", "SUCCESS", "FAILED"]
status_counts = Counter(statuses)
print(status_counts)
print(status_counts.most_common(1))Group records and compute multiple metrics
from collections import defaultdict
summary = defaultdict(lambda: {"count": 0, "amount": 0.0})
for row in sales:
region = row["region"]
summary[region]["count"] += 1
summary[region]["amount"] += row["amount"]
summary_rows = [
{"region": region, **metrics}
for region, metrics in summary.items()
]
print(summary_rows)6. Building Reusable ETL Functions
Good ETL code separates extraction, transformation, validation, and loading. Small functions are easier to test and reuse than a single large script.
def extract(source):
"""Read records from a source."""
return source # Replace with API, file, or database logic.
def transform(record):
"""Normalize a single record."""
return {
"order_id": str(record["order_id"]),
"amount": float(record["amount"])
}
def validate(record):
"""Raise an error for invalid business values."""
if record["amount"] < 0:
raise ValueError("amount cannot be negative")
return record
def load(records, target_writer):
"""Send transformed records to a target writer."""
for record in records:
target_writer(record)
def run_pipeline(source, target_writer):
extracted = extract(source)
transformed = (validate(transform(row)) for row in extracted)
load(transformed, target_writer)The code is a structural example. A production implementation should define source and target contracts, batch boundaries, transaction semantics, idempotency, metrics, and failure recovery.
Make transformations testable
def normalize_order(row):
return {
"order_id": str(row["order_id"]).strip(),
"amount": float(row["amount"]),
"currency": row.get("currency", "USD").upper()
}
sample = {"order_id": 1001, "amount": "25.50", "currency": "usd"}
expected = {
"order_id": "1001",
"amount": 25.5,
"currency": "USD"
}
assert normalize_order(sample) == expected7. Logging, Exceptions, and Retries
Logging makes pipeline behavior observable. Exceptions communicate failures, while retries can recover from transient issues such as timeouts or temporary service errors.
Use structured logging basics
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s %(message)s"
)
logger = logging.getLogger("orders_pipeline")
logger.info("Pipeline started")
logger.warning("Skipping invalid record_id=%s", "e-42")
logger.error("Load failed for batch_id=%s", "batch-7")Retry transient errors with exponential backoff
import time
import random
import logging
logger = logging.getLogger(__name__)
def retry(operation, attempts=3, base_delay=1.0):
for attempt in range(1, attempts + 1):
try:
return operation()
except (TimeoutError, ConnectionError):
if attempt == attempts:
logger.exception("Operation failed after retries")
raise
delay = base_delay * (2 ** (attempt - 1))
delay += random.uniform(0, 0.25 * delay)
logger.warning(
"Transient failure; attempt=%s retrying in %.2fs",
attempt, delay
)
time.sleep(delay)Preserve useful exception context
try:
process_batch(batch)
except ValueError:
logger.exception("Invalid data in batch_id=%s", batch_id)
raise
except Exception:
logger.exception("Unexpected pipeline failure batch_id=%s", batch_id)
raiseUse logger.exception() inside an exception handler to include traceback details. Re-raise failures when the orchestration layer must mark the task as failed or trigger recovery.
8. Processing Large Datasets Efficiently
Large-data processing is often constrained by memory, network I/O, CPU, or database throughput. Choose the processing pattern based on data size and source capabilities.
Stream records instead of loading everything
import json
def read_jsonl(path):
with open(path, "r", encoding="utf-8") as file:
for line in file:
if line.strip():
yield json.loads(line)
for event in read_jsonl("large_events.jsonl"):
process_event(event)Process in batches
from itertools import islice
def batched(iterable, batch_size):
iterator = iter(iterable)
while True:
batch = list(islice(iterator, batch_size))
if not batch:
break
yield batch
for batch in batched(read_jsonl("large_events.jsonl"), 1000):
process_batch(batch)Batch size is a tradeoff: small batches reduce peak memory and retry scope but can increase overhead; large batches may improve throughput but require more memory and make retries more expensive.
Use sets for membership and generators for pipelines
valid_customer_ids = {"C1", "C2", "C3"}
def filter_known_customers(rows):
for row in rows:
if row.get("customer_id") in valid_customer_ids:
yield row
filtered = filter_known_customers(read_jsonl("events.jsonl"))Know when Python is not the right processing engine
- Use Python for orchestration, API ingestion, moderate transformations, and custom business logic.
- Use SQL engines for set-based transformations close to the warehouse data.
- Use distributed engines such as Spark or Apache Beam when data volume or processing requirements exceed a single machine.
- Use vectorized libraries such as pandas or NumPy for suitable workloads that fit memory and benefit from vectorized operations.
Interview Questions to Practice
1. How do you build a reliable API ingestion pipeline?
Use timeouts, authentication through a secrets manager, pagination, schema validation, rate-limit handling, retries for transient errors, observability, and idempotent writes. Track progress so a failed run can resume safely.
2. How do you flatten nested JSON without losing data?
Identify parent and child entities, preserve stable keys, and decide whether arrays should become child rows or separate tables. Validate optional fields and avoid aggregating after a one-to-many expansion without checking row multiplication.
3. How would you deduplicate incoming events?
Define the event's unique key and a deterministic survivor rule, such as highest source sequence or latest valid timestamp. Track duplicates and ensure the target write is idempotent.
4. How do you design reusable ETL code?
Separate extraction, transformation, validation, and loading into small functions. Make dependencies explicit, keep transformations testable, define clear input/output contracts, and expose metrics and failure context.
5. Which errors should be retried?
Usually transient network errors, timeouts, and selected temporary service failures. Do not blindly retry invalid data, authentication failures, or non-idempotent writes. Apply bounded exponential backoff and jitter.
6. How would you process a file too large for memory?
Stream it line by line or in batches, write output incrementally, and avoid collecting every record in a list. If the workload needs distributed computation, use a suitable processing engine instead of forcing it into one Python process.
7. How do you make a pipeline safe to rerun?
Use stable keys, deterministic transformations, idempotent upserts or overwrite-safe partitions, checkpoints, and a clear commit boundary. Ensure partial failures don't cause duplicate records when the job restarts.
Chapter Summary
- Build API clients with timeouts, pagination, authentication, and deliberate error handling.
- Read and write files with context managers; stream JSONL when datasets are large.
- Flatten nested JSON according to its data model, preserving parent-child relationships.
- Define deduplication keys and validation rules explicitly, and quarantine invalid records when needed.
- Use running aggregations to limit memory usage when raw group members are unnecessary.
- Structure ETL into reusable, testable stages and add logging, bounded retries, and idempotent writes.
- Measure the bottleneck and choose Python, SQL, pandas, Spark, or Beam based on the workload.
Note: Examples use Python 3. The API examples use the third-party requests package, and the flattening example uses pandas. Adapt authentication, pagination, schema rules, and retry behavior to the actual service and production environment.
Comments
Post a Comment