Introduction
You've built a pipeline that reads customer orders from a CSV file and loads them into your data warehouse. Everything looks fine in development — your test data is clean, your types line up, and rows flow smoothly from source to sink. Then Monday morning arrives, and you're staring at a stack trace because a field called order_total contains the string "N/A", your pipeline tried to cast it to a float, and the whole job crashed halfway through a 500,000-row load.
This scenario plays out constantly in production data engineering. Real-world data is messy. Sources send strings where you expect numbers, timestamps arrive in three different formats, and nullable fields are sometimes empty strings, sometimes null, sometimes the literal text "null". Your pipeline's transformation layer is the gatekeeper between that chaos and your clean, typed, queryable data store — and how you handle type coercion and nullability at that boundary determines whether your pipeline is fragile or robust.
By the end of this lesson, you'll know how to write transformation logic that handles these situations gracefully: safely attempting type conversions, applying sensible defaults, and routing genuinely bad records away from your main dataset without crashing the pipeline.
What you'll learn:
- What type coercion is and why it fails silently or loudly in different contexts
- How to safely cast fields using try/except patterns and schema validation
- How to distinguish between legitimate nulls and malformed data
- How to apply defaults, fallbacks, and conditional logic to keep pipelines flowing
- How to reject and quarantine records that can't be salvaged, rather than letting them corrupt your output
Prerequisites
You should be comfortable writing Python functions and understand what a data pipeline does at a basic level. Familiarity with pandas or basic SQL will help you follow the examples, but isn't strictly required. If you're brand new to pipelines, spend a few minutes with What is a Data Pipeline? Architecture and Core Concepts for Data Engineers before continuing.
What Is Type Coercion, and Why Does It Matter?
Type coercion is the process of converting a value from one data type to another. When your pipeline reads a CSV file, every value starts as a string — the number 42 arrives as the two-character sequence "42", and true arrives as the four-character sequence "true". Your transformation layer is responsible for converting those strings into the types your downstream system actually expects: integers, floats, booleans, timestamps, and so on.
This sounds simple, but it breaks in three distinct ways:
1. Hard crashes. Python's int("N/A") raises a ValueError. Pandas' pd.to_datetime("not-a-date") raises a ParserError. If you don't handle these exceptions, your pipeline stops mid-run.
2. Silent data corruption. Some tools are too forgiving. SQL databases might silently truncate a string to fit a column's max length. Pandas' astype(int) on a float column will truncate 3.9 to 3 without warning. You end up with data that loaded successfully but is subtly wrong.
3. Null propagation. A field that's None in Python, NULL in SQL, or NaN in a NumPy array can quietly infect downstream calculations. A single null in a sum aggregation doesn't crash anything — it just returns null, which then propagates through every query that touches it.
Understanding which failure mode you're dealing with is the first step to handling it correctly.
Key insight: The goal of type coercion handling isn't to accept every record — it's to make an explicit, intentional decision about every record. Crash, default, or reject are all valid outcomes. Silent corruption is never acceptable.
The Three Strategies: Cast, Default, Reject
Every malformed field in your pipeline forces you to choose one of three strategies:
Cast safely: Attempt the conversion, and if it fails, handle the exception explicitly. The record can still pass through, but you know what happened.
Default: Replace the bad value with a known-good sentinel — zero, an empty string, a "unknown" category, or a business-defined fallback. The record passes through with an imputed value.
Reject: Decide that this record cannot be trusted at all, remove it from the main pipeline, and route it somewhere else for investigation.
These aren't mutually exclusive — you might cast safely, then default if casting fails, then reject if the defaulted value would violate a business rule downstream. Let's build each pattern from the ground up.
Safe Casting with Try/Except
The most fundamental tool is a small helper function that wraps a conversion attempt in a try/except block and returns a sentinel value on failure.
def safe_cast(value, cast_fn, default=None):
"""
Attempt to cast `value` using `cast_fn`.
Returns `default` if the cast fails.
"""
try:
return cast_fn(value)
except (ValueError, TypeError):
return default
Now you can use it across different types:
# Integer casting
safe_cast("42", int) # Returns 42
safe_cast("N/A", int) # Returns None
safe_cast("3.7", int) # Returns None (int("3.7") raises ValueError)
safe_cast(None, int) # Returns None (TypeError caught)
# Float casting
safe_cast("19.99", float) # Returns 19.99
safe_cast("$19.99", float) # Returns None
# Boolean casting (custom function)
def parse_bool(v):
if str(v).lower() in ("true", "1", "yes"):
return True
if str(v).lower() in ("false", "0", "no"):
return False
raise ValueError(f"Cannot parse boolean from: {v!r}")
safe_cast("yes", parse_bool) # Returns True
safe_cast("maybe", parse_bool) # Returns None
The key insight here is that safe_cast never crashes. It always returns something, and that something is either a properly typed value or an explicit None that you can reason about later.
Warning: Be careful about what you put in the default parameter. Returning 0 as a default for a failed integer cast means a "N/A" order total becomes 0.0 in your analytics — which looks like a real order for zero dollars. In many cases, returning None and handling it later is safer than choosing a numeric default eagerly.
Not all nulls are born equal. You need to distinguish between two fundamentally different situations:
Legitimately null: The field is nullable by design. A customer's middle_name might genuinely be absent. A discount_amount might be null because no discount was applied. These nulls are valid data that your schema should accommodate.
Malformed: The field is expected to have a value, but what's there is unparseable. An order_date of "ASAP" is not a null — it's a broken value that someone entered incorrectly.
The distinction matters because they call for different actions. Legitimate nulls should pass through with NULL/None/NaN as-is. Malformed values need to be flagged, defaulted, or rejected.
Here's how to make that distinction explicit in code:
EXPLICIT_NULL_MARKERS = {"", "null", "none", "n/a", "na", "nan", "#n/a"}
def normalize_null(value):
"""
Returns None if the value represents an intentional missing value,
otherwise returns the raw value for further parsing.
"""
if value is None:
return None
if str(value).strip().lower() in EXPLICIT_NULL_MARKERS:
return None
return value
def parse_field(raw_value, cast_fn, nullable=True, default=None):
"""
Full field parsing pipeline:
1. Normalize explicit null markers to None
2. If None and nullable, return None
3. If None and not nullable, return default (which may itself be None — caller decides)
4. Attempt the cast
5. On failure, return default
"""
normalized = normalize_null(raw_value)
if normalized is None:
return None if nullable else default
return safe_cast(normalized, cast_fn, default=default)
Now let's see this in practice with a realistic record:
raw_record = {
"order_id": "10042",
"customer_id": "C-889",
"order_total": "N/A", # malformed — should be a float
"discount_amount": "", # legitimately null — no discount applied
"item_count": "3",
"order_date": "2024-13-45", # malformed date
"is_priority": "yes",
}
from datetime import datetime
def parse_date(v):
return datetime.strptime(v, "%Y-%m-%d")
parsed = {
"order_id": parse_field(raw_record["order_id"], int, nullable=False),
"customer_id": raw_record["customer_id"], # stays a string
"order_total": parse_field(raw_record["order_total"], float, nullable=False),
"discount_amount": parse_field(raw_record["discount_amount"], float, nullable=True),
"item_count": parse_field(raw_record["item_count"], int, nullable=False),
"order_date": parse_field(raw_record["order_date"], parse_date, nullable=False),
"is_priority": parse_field(raw_record["is_priority"], parse_bool, nullable=False),
}
print(parsed)
# {
# 'order_id': 10042,
# 'customer_id': 'C-889',
# 'order_total': None, ← cast failed, field is non-nullable, default is None
# 'discount_amount': None, ← legitimately null, nullable=True
# 'item_count': 3,
# 'order_date': None, ← cast failed
# 'is_priority': True,
# }
Both order_total and discount_amount are None in the output, but they got there for completely different reasons. That distinction matters when you decide what to do next.
Applying Defaults: When None Isn't Good Enough
Sometimes a None value will break downstream consumers — a SQL column with a NOT NULL constraint, a model that can't handle missing input features, a dashboard metric that divides by a field that must have a value. In those cases, you need to apply a business-defined default.
Tip: Always document where a default came from. A hardcoded 0 buried in transformation logic will confuse the next engineer who looks at your data and wonders why every record with a bad order total shows a $0.00 transaction. Add it to your pipeline's metadata or at minimum leave a comment.
There are three kinds of defaults worth knowing:
Static defaults — a fixed value applied regardless of context:
# If item_count is missing, assume 1 (minimum viable order)
item_count = parse_field(raw["item_count"], int, nullable=False) or 1
Contextual defaults — derived from other fields in the same record:
# If order_date is missing, fall back to record ingestion timestamp
order_date = parse_field(raw["order_date"], parse_date, nullable=False)
if order_date is None:
order_date = ingestion_timestamp
Lookup defaults — pulled from a reference table or config:
# If customer region is missing, look it up from the customer master
region = parse_field(raw["region"], str, nullable=True)
if region is None:
region = customer_lookup.get(raw["customer_id"], {}).get("region", "UNKNOWN")
Be conservative with defaults. Every default you apply is an assumption about what the real value should have been. Those assumptions compound. A pipeline that defaults aggressively will keep running, but it's accumulating silent inaccuracies that show up as mysterious discrepancies in reports.
Some records are too broken to fix. An order with no order_id, a transaction with a negative quantity, a user record where email fails format validation — these aren't worth coercing or defaulting. Loading them would introduce garbage data downstream that's harder to find and fix than a missing record.
The right pattern here is to validate after parsing and route bad records to a dead letter store. This is related to the broader pattern described in Implementing Dead Letter Queues and Poison Message Handling in Data Pipelines — records that can't be processed don't disappear, they go somewhere you can inspect and reprocess them.
Here's a simple validation-and-rejection layer:
def validate_record(parsed):
"""
Returns a list of validation errors.
An empty list means the record is clean.
"""
errors = []
if parsed.get("order_id") is None:
errors.append("order_id is missing or unparseable")
if parsed.get("order_total") is None:
errors.append("order_total is missing or unparseable")
elif parsed["order_total"] < 0:
errors.append(f"order_total is negative: {parsed['order_total']}")
if parsed.get("order_date") is None:
errors.append("order_date is missing or unparseable")
if parsed.get("item_count") is not None and parsed["item_count"] <= 0:
errors.append(f"item_count must be positive, got: {parsed['item_count']}")
return errors
def process_records(raw_records):
good_records = []
dead_letter = []
for raw in raw_records:
parsed = parse_all_fields(raw)
errors = validate_record(parsed)
if errors:
dead_letter.append({
"original": raw,
"parsed": parsed,
"errors": errors,
"rejected_at": datetime.utcnow().isoformat(),
})
else:
good_records.append(parsed)
return good_records, dead_letter
The dead letter list should be written to a separate location — a database table, a file, a queue — where it can be reviewed, fixed, and reprocessed without touching your main pipeline run.
Note: Your data quality validation and monitoring tooling should track reject rates over time. A sudden spike in rejections is often the first signal that a source system changed its schema or started producing bad data — long before anyone notices missing records downstream.
Let's assemble everything into a clean, cohesive transformation function that handles a batch of raw order records end-to-end:
import csv
from datetime import datetime
from typing import Any
EXPLICIT_NULL_MARKERS = {"", "null", "none", "n/a", "na", "nan", "#n/a"}
def normalize_null(value):
if value is None:
return None
if str(value).strip().lower() in EXPLICIT_NULL_MARKERS:
return None
return value
def safe_cast(value, cast_fn, default=None):
try:
return cast_fn(value)
except (ValueError, TypeError):
return default
def parse_field(raw_value, cast_fn, nullable=True, default=None):
normalized = normalize_null(raw_value)
if normalized is None:
return None if nullable else default
return safe_cast(normalized, cast_fn, default=default)
def parse_bool(v):
if str(v).lower() in ("true", "1", "yes"):
return True
if str(v).lower() in ("false", "0", "no"):
return False
raise ValueError(f"Cannot parse boolean: {v!r}")
def parse_date(v):
for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%d-%b-%Y"):
try:
return datetime.strptime(v, fmt)
except ValueError:
continue
raise ValueError(f"No matching date format for: {v!r}")
def transform_order(raw: dict) -> dict:
return {
"order_id": parse_field(raw.get("order_id"), int, nullable=False),
"customer_id": parse_field(raw.get("customer_id"), str.strip, nullable=False),
"order_total": parse_field(raw.get("order_total"), float, nullable=False),
"discount_amount": parse_field(raw.get("discount_amount"), float, nullable=True),
"item_count": parse_field(raw.get("item_count"), int, nullable=False, default=1),
"order_date": parse_field(raw.get("order_date"), parse_date, nullable=False),
"is_priority": parse_field(raw.get("is_priority"), parse_bool, nullable=False, default=False),
}
def validate_order(parsed: dict) -> list[str]:
errors = []
if parsed["order_id"] is None:
errors.append("Missing order_id")
if parsed["customer_id"] is None:
errors.append("Missing customer_id")
if parsed["order_total"] is None:
errors.append("Missing or invalid order_total")
elif parsed["order_total"] < 0:
errors.append(f"Negative order_total: {parsed['order_total']}")
if parsed["order_date"] is None:
errors.append("Missing or invalid order_date")
return errors
def run_transform(raw_records):
good, rejected = [], []
for raw in raw_records:
parsed = transform_order(raw)
errors = validate_order(parsed)
if errors:
rejected.append({"record": raw, "errors": errors})
else:
good.append(parsed)
return good, rejected
This structure is clean, testable, and extensible. Adding a new field means adding one line to transform_order and optionally one check in validate_order. Each concern — parsing, nullability, validation, routing — lives in its own function. It also pairs cleanly with pipeline testing patterns because each helper function is independently unit-testable.
Hands-On Exercise
Work through this exercise to solidify the concepts. You'll need Python 3.10+ and no external libraries.
The scenario: You're building a pipeline that ingests a product catalog CSV from a retail partner. The file has five columns: product_id, price_usd, stock_quantity, is_active, and last_updated. The partner's data team is... inconsistent.
Step 1: Create a list of raw records to simulate reading a CSV:
raw_records = [
{"product_id": "P001", "price_usd": "29.99", "stock_quantity": "150", "is_active": "true", "last_updated": "2024-03-15"},
{"product_id": "P002", "price_usd": "N/A", "stock_quantity": "32", "is_active": "yes", "last_updated": "2024/03/10"},
{"product_id": "", "price_usd": "15.00", "stock_quantity": "0", "is_active": "false", "last_updated": "2024-03-12"},
{"product_id": "P004", "price_usd": "-5.00", "stock_quantity": "10", "is_active": "1", "last_updated": "null"},
{"product_id": "P005", "price_usd": "199.00", "stock_quantity": "none","is_active": "maybe", "last_updated": "2024-03-01"},
]
Step 2: Write a transform_product function using the patterns from this lesson. Decide which fields are nullable and which aren't.
Step 3: Write a validate_product function. Add a business rule: price_usd must be greater than zero.
Step 4: Run your transform over all five records. Which ones pass? Which get rejected and why?
Expected outcomes to check your work: Records P001 and P003 should pass (a stock quantity of 0 is valid). P002 should pass if you treat a missing price as nullable, or be rejected if you don't. P004 should be rejected because of the negative price. P005 should be rejected because is_active of "maybe" can't be parsed.
Common Mistakes & Troubleshooting
Catching too broadly. Using a bare except Exception in your safe_cast will hide bugs in your cast functions — like a date parser that has a logic error that raises an AttributeError. Catch only the exceptions that genuinely represent bad input: ValueError, TypeError, OverflowError for numeric types, and KeyError for dictionary lookups.
Defaulting too eagerly. Applying a 0 default to every failed numeric cast means you can never tell the difference between "this order genuinely had zero discount" and "this discount field was unparseable." Keep a separate parse_errors column in your output schema or write rejections to a dead letter table.
Forgetting whitespace. " 42 ".strip() is "42", which casts fine. " 42 " raises a ValueError on int(). Always strip whitespace before type conversion.
Treating all date formats the same. "03/04/2024" could be March 4th or April 3rd depending on locale. If your source doesn't specify, pick a canonical format and reject anything that doesn't match it — don't guess.
Not tracking rejection rates. A pipeline that quietly routes 30% of records to dead letter every day isn't "working" — it's failing slowly. Track reject rates as a pipeline health metric. See Logging, Alerting, and Observability for Data Pipelines for patterns on surfacing this kind of operational data.
Warning: Schema changes at the source are one of the most common causes of sudden spikes in type coercion failures. If your price_usd field suddenly starts arriving as "USD 29.99" instead of "29.99", every single record will fail your float cast. Build alerts on rejection rate and read about schema evolution strategies to handle this gracefully.
Summary & Next Steps
Type coercion and nullability handling are not glamorous — but they're what separates a pipeline that works in a demo from one that runs reliably in production for years. Here's what you built in this lesson:
- A
safe_cast helper that wraps type conversions in explicit exception handling
- A
normalize_null function that standardizes the many ways sources express "missing"
- A
parse_field function that distinguishes nullable from required fields
- A validation layer that applies business rules and routes bad records to dead letter
- A full transformation pipeline that assembles these pieces into clean, testable functions
The next natural step is to think about what happens when your schema itself changes — when the source starts sending a new field, renames an existing one, or changes a column's type. That's the territory covered in Schema Evolution Strategies for Production Data Pipelines.
You should also think about how to test the transformation layer you've built. The patterns here lend themselves naturally to unit testing — each function is pure and has a clear input/output contract. Pipeline Testing: Unit Tests, Integration Tests, and Data Contracts will walk you through building a proper test suite around exactly this kind of transformation code.
Finally, if you're working with high-volume pipelines where millions of records pass through these transforms, the choices you make here have performance implications. Validating and coercing field by field in pure Python doesn't scale the same way as expressing those rules in a vectorized Pandas operation or a SQL CAST statement. Building Your First Data Pipeline with Python covers where to go from here on the implementation side.
