Move beyond Python basics with practical techniques for transforming collections, grouping records, processing streams, parsing text and nested data, and reasoning about performance. These patterns appear frequently in data engineering scripts, ETL jobs, and coding interviews.
1. Dictionary and List Manipulation
Lists are useful for ordered sequences, while dictionaries provide key-based lookups. In data engineering, you often transform a list of records, build lookup maps, and aggregate fields.
Transform a list of dictionaries
orders = [
{"order_id": 101, "customer_id": "C1", "amount": 250},
{"order_id": 102, "customer_id": "C2", "amount": 175},
{"order_id": 103, "customer_id": "C1", "amount": 300},
]
# Select only the fields needed downstream
rows = [
{"id": row["order_id"], "amount": row["amount"]}
for row in orders
]
print(rows)Build a lookup dictionary
customer_names = {
"C1": "Asha",
"C2": "Ravi",
}
for order in orders:
order["customer_name"] = customer_names.get(
order["customer_id"], "UNKNOWN"
)
print(orders)Aggregate values by key
total_by_customer = {}
for order in orders:
customer_id = order["customer_id"]
total_by_customer[customer_id] = (
total_by_customer.get(customer_id, 0) + order["amount"]
)
print(total_by_customer)
# {'C1': 550, 'C2': 175}Safely update nested dictionaries
record = {
"customer": {
"id": "C1",
"address": {"city": "Bengaluru"}
}
}
city = record.get("customer", {}).get("address", {}).get("city")
print(city) # Bengalurudict.get() helps when a key may be absent, but chained get() calls do not protect you if an intermediate value exists but is None or has the wrong type. Validate incoming records when schemas can vary.Shallow copy versus deep copy
import copy
original = {"tags": ["new", "priority"]}
shallow = original.copy()
deep = copy.deepcopy(original)
original["tags"].append("processed")
print(shallow["tags"]) # shares the nested list
print(deep["tags"]) # independent nested listA shallow copy creates a new outer collection but retains references to nested objects. Use a deep copy only when you need independent nested structures and can afford its cost.
2. Sorting and Grouping
Sort records by one or more fields
employees = [
{"name": "Mina", "team": "Data", "salary": 120000},
{"name": "Asha", "team": "Platform", "salary": 135000},
{"name": "Ravi", "team": "Data", "salary": 110000},
]
# Sort by salary descending
by_salary = sorted(
employees,
key=lambda employee: employee["salary"],
reverse=True
)
# Sort by team, then salary descending
by_team_salary = sorted(
employees,
key=lambda employee: (employee["team"], -employee["salary"])
)
print(by_team_salary)sorted() returns a new list; list.sort() sorts a list in place and returns None. Python sorting is stable, so equal keys preserve their prior relative order.
Group records with itertools.groupby
from itertools import groupby
# groupby groups consecutive matching keys,
# so sort by the grouping key first.
employees_sorted = sorted(employees, key=lambda row: row["team"])
for team, rows in groupby(employees_sorted, key=lambda row: row["team"]):
team_rows = list(rows)
print(team, len(team_rows), team_rows)itertools.groupby() does not automatically collect all equal keys from an unsorted list. It groups adjacent runs. Sort by the same key first, or use a dictionary-based aggregation when order is not important.Group and aggregate without groupby
from collections import defaultdict
orders_by_customer = defaultdict(list)
for order in orders:
orders_by_customer[order["customer_id"]].append(order)
for customer_id, customer_orders in orders_by_customer.items():
total = sum(order["amount"] for order in customer_orders)
print(customer_id, total)3. Collections: Counter, defaultdict, and deque
The collections module provides efficient, purpose-built data structures for common tasks.
Counter: count values
from collections import Counter
statuses = ["SUCCESS", "FAILED", "SUCCESS", "SKIPPED", "FAILED", "SUCCESS"]
counts = Counter(statuses)
print(counts)
print(counts.most_common(2))
print(counts["SUCCESS"]) # 3Use Counter for frequency counts, duplicate detection, and top-N values.
defaultdict: build grouped collections
from collections import defaultdict
events_by_user = defaultdict(list)
events = [
("U1", "login"),
("U2", "login"),
("U1", "purchase"),
]
for user_id, event_type in events:
events_by_user[user_id].append(event_type)
print(dict(events_by_user))A defaultdict(list) creates an empty list automatically when a missing key is first accessed. Note that accessing a missing key creates it.
deque: efficient queue operations
from collections import deque
queue = deque(["file1.csv", "file2.csv"])
queue.append("file3.csv") # add to right
queue.appendleft("priority.csv")
next_file = queue.popleft() # remove from left
print(next_file)
print(queue)Appending and popping at either end of a deque are approximately O(1). Removing the first item from a regular list with pop(0) is O(n), because remaining elements must shift.
Maintain a rolling window
from collections import deque
window = deque(maxlen=3)
for value in [10, 20, 30, 40]:
window.append(value)
print(list(window))With maxlen, appending to a full deque automatically discards an item from the opposite end. This is useful for bounded rolling buffers, but calculate rolling aggregates carefully if values leave the window.
4. Iterators and Generators
An iterable can produce an iterator. An iterator yields one item at a time using next(). A generator is a convenient way to create an iterator using yield.
Iterator basics
values = [10, 20, 30]
iterator = iter(values)
print(next(iterator)) # 10
print(next(iterator)) # 20
print(next(iterator)) # 30
# Another next(iterator) raises StopIterationCreate a generator with yield
def read_batches(records, batch_size):
for start in range(0, len(records), batch_size):
yield records[start:start + batch_size]
records = list(range(1, 11))
for batch in read_batches(records, 3):
print(batch)This example slices a list into batches. For a truly large or streaming source, iterate over the input incrementally instead of requiring the full list in memory.
Stream records from an input iterator
def positive_amounts(rows):
for row in rows:
amount = row.get("amount")
if amount is not None and amount > 0:
yield row
# The source can itself be a generator or file iterator.
valid_rows = positive_amounts(orders)
for row in valid_rows:
print(row)Generator expression versus list comprehension
numbers = range(1_000_000)
# Builds a list in memory
squares_list = [n * n for n in numbers]
# Produces values on demand
squares_generator = (n * n for n in numbers)
# Both can be consumed by sum()
total = sum(n * n for n in range(100))5. String Processing
Text cleanup and parsing are common in ingestion pipelines, log processing, and data normalization.
Trim, split, join, and replace
raw = " Alice Smith, Bengaluru, IN "
clean = raw.strip()
parts = [part.strip() for part in clean.split(",")]
normalized = " | ".join(parts)
print(parts)
print(normalized)
print(clean.replace("Bengaluru", "Bangalore"))Check prefixes, suffixes, and case
filename = "ORDERS_2026.CSV"
print(filename.lower().endswith(".csv")) # True
print(filename.startswith("ORDERS")) # True
print("failed".upper()) # FAILEDExtract patterns with regular expressions
import re
text = "Job run_id=run-482 status=FAILED duration=31s"
match = re.search(r"run_id=(\\S+).*status=(\\w+)", text)
if match:
run_id, status = match.groups()
print(run_id, status)Use regular expressions for pattern-based parsing, but prefer straightforward string operations for simple, fixed delimiters. For complex formats such as CSV, use a CSV parser rather than splitting on commas manually.
Normalize whitespace and compare text
def normalize_name(value):
if value is None:
return None
return " ".join(value.split()).casefold()
print(normalize_name(" Asha KUMAR ")) # asha kumarcasefold() is intended for caseless matching and handles some Unicode cases more thoroughly than lower(). Whether normalization is appropriate depends on the field and business rules.
6. Working with CSV, JSON, and Nested Data
Read CSV safely
import csv
with open("orders.csv", "r", newline="", encoding="utf-8") as file:
reader = csv.DictReader(file)
for row in reader:
order_id = row["order_id"] # CSV values are strings
amount = float(row["amount"])
print(order_id, amount)CSV values are usually strings, so convert them deliberately and validate empty or malformed fields. Use newline="" as recommended by Python's CSV module.
Write CSV records
import csv
rows = [
{"order_id": 101, "amount": 250.0},
{"order_id": 102, "amount": 175.0},
]
with open("output.csv", "w", newline="", encoding="utf-8") as file:
writer = csv.DictWriter(file, fieldnames=["order_id", "amount"])
writer.writeheader()
writer.writerows(rows)Parse JSON and navigate nested data
import json
payload = {
"event_id": "e-101",
"user": {"id": "u-7", "location": {"city": "Pune"}},
"items": [
{"sku": "A1", "quantity": 2},
{"sku": "B2", "quantity": 1}
]
}
text = json.dumps(payload)
event = json.loads(text)
user_id = event["user"]["id"]
city = event["user"]["location"]["city"]
skus = [item["sku"] for item in event["items"]]
print(user_id, city, skus)Flatten nested records
flattened = {
"event_id": event["event_id"],
"user_id": event["user"]["id"],
"city": event["user"]["location"]["city"],
"item_count": len(event["items"]),
"total_quantity": sum(item["quantity"] for item in event["items"])
}
print(flattened)Process JSON Lines incrementally
import json
def read_jsonl(path):
with open(path, "r", encoding="utf-8") as file:
for line_number, line in enumerate(file, start=1):
if not line.strip():
continue
try:
yield json.loads(line)
except json.JSONDecodeError as exc:
# Replace with structured logging/quarantine in production.
print(f"Invalid JSON at line {line_number}: {exc}")
for event in read_jsonl("events.jsonl"):
print(event.get("event_id"))7. Time and Space Complexity
Big O notation describes how an algorithm's resource needs grow with input size. It helps compare approaches before running them on production-scale data.
| Operation | Typical complexity | Notes |
|---|---|---|
| List index access | O(1) | Direct access by position |
| Search in a list | O(n) | May inspect every element |
| Append to a list | Amortized O(1) | Occasional resizing can cost more |
| Insert/remove at list front | O(n) | Elements shift positions |
| Dictionary lookup/insert | Average O(1) | Worst-case behavior can differ |
| Set membership | Average O(1) | Hash-based lookup |
| Comparison sort | O(n log n) | Python sorting is stable |
| Build a list of n items | O(n) space | Stores all items in memory |
| Generator pipeline | Often O(1) extra space | Depends on upstream and downstream operations |
Example: List membership versus set membership
allowed_ids_list = list(range(100_000))
allowed_ids_set = set(allowed_ids_list)
target = 99_999
# List search is O(n) in the worst case
found_in_list = target in allowed_ids_list
# Set membership is average O(1)
found_in_set = target in allowed_ids_setBuilding the set costs additional time and memory, so it is most useful when you perform many membership checks or need uniqueness.
Avoid accidental quadratic behavior
# Less efficient: repeated list membership can be O(n*m)
unique = []
for value in values:
if value not in unique:
unique.append(value)
# Often more efficient for membership, while preserving first-seen order
seen = set()
unique = []
for value in values:
if value not in seen:
seen.add(value)
unique.append(value)Interview Questions to Practice
1. How do you group a list of dictionaries by a key?
Use defaultdict(list) to append records under each key, or use a dictionary and initialize missing keys. For aggregations such as sums or counts, store the running aggregate instead of every record if the original rows are not needed.
2. What is the difference between sorted() and list.sort()?
sorted() returns a new sorted list from an iterable. list.sort() mutates an existing list in place and returns None. Both use a stable sort.
3. How is itertools.groupby different from SQL GROUP BY?
itertools.groupby() groups consecutive runs of equal keys, so data usually needs sorting by the grouping key first. SQL GROUP BY groups matching keys across the input regardless of their original order.
4. Why use deque instead of a list for a queue?
A deque supports efficient appends and pops at both ends. Removing index zero from a list is O(n) because the remaining elements must shift.
5. What is the benefit of generators in ETL pipelines?
Generators yield records on demand and can reduce peak memory. They are useful for streaming transformations, but a generator is normally consumed once and doesn't eliminate memory used by other pipeline stages.
6. How do you process a large CSV or JSONL file?
Use the CSV reader or iterate over JSONL one line at a time. Validate and transform each record incrementally, handle malformed rows explicitly, and write output in batches or as a stream where possible.
7. How can you improve repeated membership checks?
Convert the lookup collection to a set when appropriate. Set construction costs O(n) time and memory, but repeated membership checks are average O(1), compared with O(n) per search in a list.
Chapter Summary
- Use lists and dictionaries to transform records and build efficient lookup or aggregation structures.
- Understand sorting stability and remember that
itertools.groupby()groups consecutive keys. - Use Counter for frequencies, defaultdict for grouped values, and deque for queues or rolling buffers.
- Use generators to process records lazily and reduce memory pressure where the full result need not be retained.
- Use standard CSV and JSON parsers rather than manually splitting structured formats.
- Analyze both time and space complexity, and measure performance with realistic data volumes.
Note: Examples use Python 3. Adapt parsing, validation, logging, and memory-management choices to your data volume and production requirements.
Comments
Post a Comment