Python and Pandas for Data Analysis Questions

Direct answer

Write each stage (read, filter, impute, feature creation, write) as a small pure function that takes a DataFrame and returns one, and compose them with .pipe() so the pipeline reads top-to-bottom as its own table of contents. Each function can then be unit-tested in complete isolation with a two- or three-row DataFrame, with no need to run the rest of the pipeline or touch the filesystem.

Approach

import pandas as pd

def read_csv(path):
    return pd.read_csv(path, parse_dates=["event_time"])

def filter_rows(df):
    return df[df["event_time"].notna() & (df["revenue"] > 0)]

def impute(df):
    df["country"] = df["country"].fillna("unknown")
    df["revenue"] = df["revenue"].fillna(0)
    return df

def create_features(df):
    # cast the period to str: parquet's partition_cols can't hash a raw Period dtype column
    df["month"] = df["event_time"].dt.to_period("M").astype(str)
    df["rev_per_item"] = df["revenue"] / df["items"].replace(0, 1)
    return df

def run_pipeline(input_path, output_dir):
    result = (
        read_csv(input_path)   # the chain starts from the function that produces the first DataFrame,
        .pipe(filter_rows)     # not from an undefined variable piped through read_csv
        .pipe(impute)
        .pipe(create_features)
    )
    result.to_parquet(output_dir, partition_cols=["month", "country"], engine="pyarrow")
    return result

Key points:

• The chain has to start from a call that actually produces a DataFrame (read_csv("input.csv")), not from .pipe()-ing an undefined name through a reader function; .pipe() threads an existing DataFrame through the next step, it can't conjure the first one.
• create_features casts month to a plain string before it becomes a partition column: to_period("M") produces a pandas-specific Period dtype, and PyArrow's write_to_dataset (which to_parquet(partition_cols=...) uses under the hood) cannot hash that extension dtype as a partition key, so writing the raw Period column raises ArrowNotImplementedError: Keys of type extension<pandas.period<ArrowPeriodType>>.
• impute and create_features each receive the DataFrame that the previous .pipe() call already produced fresh (from filter_rows's boolean-indexed slice, which is itself already a new object under Copy-on-Write), so a plain df["col"] = ... inside them is an ordinary column assignment, not the chained-slice pattern that pandas 3.0 flags.

Worked example

pd.DataFrame({
    "event_time": ["2026-01-05", "2026-01-20", "2026-02-01", None, "2026-02-10"],
    "revenue": [10.0, 0.0, 25.0, 15.0, -5.0],
    "items": [2, 1, 0, 3, 1],
    "country": ["US", None, "DE", "US", "FR"],
}).to_csv("input.csv", index=False)

result = run_pipeline("input.csv", "out/")
print(result)

Output (verified against pandas 3.0.3, pyarrow):

  event_time  revenue  items country    month  rev_per_item
0 2026-01-05     10.0      2      US  2026-01           5.0
2 2026-02-01     25.0      0      DE  2026-02          25.0

Of the five input rows, three are filtered out by filter_rows: the row with revenue == 0, the row with a missing event_time, and the row with negative revenue (a data-quality problem, not something impute should paper over). The items == 0 row on 2026-02-01 survives filtering (its revenue is positive) and rev_per_item correctly falls back to dividing by 1 instead of 0 via .replace(0, 1). Writing this result with partition_cols=["month", "country"] succeeds and reading it back with pd.read_parquet("out/") reproduces the same two rows.

Complexity and edge cases

Complexity: each stage is a single O(n) pass over the rows it touches; .pipe() itself adds no algorithmic cost, it is function composition. Memory: each .pipe() call's result is a fresh DataFrame (a shallow copy for the untouched columns plus whatever the step added or filtered), so peak memory during the chain is bounded by roughly the size of the largest single intermediate, not the sum across every stage, since Python frees each intermediate once the next step has consumed it.

Edge cases: an items value of exactly 0 needs the .replace(0, 1) guard or rev_per_item divides by zero; a missing event_time is dropped by the filter rather than imputed, since a synthetic timestamp would be a worse assumption than dropping the row; an already-unknown or already-zero-filled value passing through impute a second time is a no-op, so the function is safe to call more than once (idempotent) if the pipeline is ever re-run on partially-processed data.

Trade-offs and pitfalls

Unit-testing each function in isolation is the main payoff of this shape: create_features can be tested with a two-row DataFrame asserting the exact rev_per_item value and that month is a plain string (not a Period), without needing a real CSV file or a filesystem to write parquet to; impute can be tested by asserting that a null country becomes "unknown" and a null revenue becomes 0, again with no dependency on the rest of the chain. Mocking I/O becomes trivial too, since read_csv and the final to_parquet call are the only two functions that touch the filesystem, everything in between is a pure DataFrame-to-DataFrame transformation that pytest fixtures can exercise directly. The pitfall in a .pipe() chain specifically: because each step only runs when the chain executes, not when it's defined, a bug like the undefined-starting-variable above or a missing import inside a step function stays completely invisible until someone actually runs the full pipeline end-to-end, which is exactly why testing each function independently (rather than trusting that the chain "looks right") is the point of building it this way in the first place.

Direct answer

Build one boolean mask per condition, amount > 100, status.isin([...]), created_at.between(...), and combine them with & for AND / | for OR, wrapping every individual comparison in parentheses. Parentheses are required because Python's & and | bind tighter than comparison operators like > and ==, so without them the expression groups incorrectly and pandas raises rather than silently misevaluating.

Approach

import pandas as pd

orders = pd.DataFrame({
    'order_id': [1, 2, 3, 4],
    'user_id': [10, 11, 12, 13],
    'amount': [50, 150, 200, 90],
    'status': ['complete', 'shipped', 'pending', 'complete'],
    'created_at': ['2024-01-15', '2024-02-20', '2024-02-25', '2024-04-01'],
})
orders['created_at'] = pd.to_datetime(orders['created_at'])

mask_amount = orders['amount'] > 100
mask_status = orders['status'].isin(['complete', 'shipped'])
mask_date = orders['created_at'].between('2024-01-01', '2024-03-31')

result = orders[mask_amount & mask_status & mask_date]

# Equivalent with .query()
result_q = orders.query(
    "amount > 100 and status in ['complete', 'shipped'] "
    "and created_at >= '2024-01-01' and created_at <= '2024-03-31'"
)
# result.equals(result_q) -> True

Output (only order_id 2 satisfies all three conditions: amount 150 > 100, status "shipped", created_at 2024-02-20 in range):

   order_id  user_id  amount   status created_at
1         2       11     150  shipped 2024-02-20

Key points

• Use & / | for elementwise boolean-Series logic, never the Python keywords and / or, which only work on single scalar truth values and raise on a Series.
• Wrap each comparison, (orders['amount'] > 100), in parentheses before combining with &/|; the operator-precedence trap is the single most common bug in hand-written boolean masks.
• .query() reads more like SQL and lets you write and/or/in directly as keywords inside the string, since the expression is parsed and evaluated separately from normal Python operator precedence.

Complexity

Each comparison, isin, or between call is a single vectorized O(n) pass over the column. Combining k masks with & is O(k*n) total. .query() compiles the expression once and evaluates it in a comparable O(n) pass (and can use numexpr under the hood for large frames to reduce the number of intermediate boolean arrays materialized). Memory: each intermediate boolean mask is O(n) at 1 byte per element, and the final result is O(m) for the m matching rows.

Edge cases

• NaN in a compared column: any comparison against NaN (not-a-number) evaluates to False, so rows with missing amount are silently excluded, never raised, which matches how NaN comparisons work generally.
• Missing parentheses: orders['amount'] > 100 & orders['status'] == 'complete' raises a TypeError at the &, because & binds to 100 and orders['status'] before the comparisons resolve; it fails loudly rather than returning a wrong-but-silent mask.
• Timezone-aware vs timezone-naive values mixed in created_at: comparing them raises a TypeError, so normalize timezone handling before filtering.
• Duplicate index labels in orders: boolean masking is positional in effect (aligned by index, but each row is independently True/False), so duplicates don't break the filter itself, though a later .loc lookup by label on the result could return more rows than expected.

Trade-offs and pitfalls

For very large frames, .query() can be more memory-efficient because it can avoid materializing every intermediate boolean mask (numexpr evaluates the whole expression in a more fused fashion), which matters if you are chaining many conditions. For readability with column names that are valid Python identifiers, .query() also tends to be easier to review at a glance than a long &-chained boolean expression. Prefer .loc[mask] over df[mask] when you also need to select specific columns in the same step, since df[mask][cols] = ... reintroduces exactly the chained-indexing risk that plain boolean filtering for reading avoids.

Direct answer

Parse the JSON string column with a wrapper around json.loads that catches malformed or missing input and returns an empty dict instead of raising, then expand the resulting dicts into real columns with pandas.json_normalize. Keeping the parser's failure mode consistent (always a dict, never an exception escaping) is what lets one bad row degrade gracefully instead of taking the whole pipeline down.

Approach

import json
import pandas as pd

def safe_parse(s):
    # Return dict for valid JSON, {} for null/empty, {} for malformed.
    if pd.isna(s):
        return {}
    if isinstance(s, dict):
        return s
    s = str(s).strip()
    if s == "":
        return {}
    try:
        return json.loads(s)
    except (json.JSONDecodeError, TypeError, ValueError):
        # log the row id/context here in a real pipeline before swallowing it
        return {}

df = pd.DataFrame({
    "id": [1, 2, 3, 4],
    "props": [
        '{"a":1, "b":{"c":2}}',
        None,
        '{"a":3, "b":{"c":4}}',
        '{"malformed": "no end"'
    ]
})

parsed = df["props"].map(safe_parse)
props_expanded = pd.json_normalize(parsed)   # nested keys flatten to "b.c" style dotted names
props_expanded = props_expanded.add_prefix("prop_")

result = pd.concat([df.drop(columns=["props"]), props_expanded], axis=1)
result = result.fillna(value={"prop_a": 0})
print(result)

Output (verified by running this exact code against these four rows, one valid, one null, one valid, one malformed with an unterminated brace):

   id  prop_a  prop_b.c
0   1     1.0       2.0
1   2     0.0       NaN
2   3     3.0       4.0
3   4     0.0       NaN

Row 2 (null) and row 4 (malformed JSON) both come through as {} from safe_parse, so both land as NaN (not-a-number) after the join, exactly like a row that simply had no props data, rather than crashing the whole json_normalize call.

Key points

• Catch json.JSONDecodeError specifically (not a bare except:), so a genuine bug elsewhere in the pipeline still surfaces instead of silently being treated as "malformed input."
• Return the SAME type ({}) from every failure branch so json_normalize sees a consistent list of dicts and produces stable columns; if you sometimes return None and sometimes {}, json_normalize can produce inconsistent output shapes.
• add_prefix avoids a column-name collision if the original DataFrame already has a column named the same as a JSON key.
• Cast or fill types explicitly after expansion (fillna, astype) since a column that is fully missing for some rows and populated for others comes back as float64 even for what should be integer data, because NaN forces a float upcast.
• Preserve the foreign key across the flattening step: keep id (or whichever column links a row back to its parent table) alongside the expanded JSON columns rather than dropping it, since pd.concat([df.drop(columns=["props"]), props_expanded], axis=1) above keeps id precisely so the flattened output can still be joined back to the table it came from. If you instead flattened into a separate table without carrying that key along, you would have no way to reconnect a parsed property back to its source row.

Complexity

O(n) time in the number of rows for the parse step; O(n * m) memory for the expanded result, where m is the number of distinct JSON keys across all rows, since json_normalize allocates one column per key seen anywhere in the input, not per row.

Edge cases

• JSON arrays at the top level of a record (rather than an object) need record_path or explicit list-handling; a bare json_normalize call assumes each parsed value is a dict.
• Mixed schemas across rows (row A has key x, row B doesn't) produce the union of all keys, with NaN filling the gaps; decide on a canonical schema and cast explicitly rather than letting every field silently drift to object or float64.
• Very large or deeply nested JSON payloads: consider a faster parser (orjson, ujson) or streaming/chunked parsing if a naive json.loads per row becomes the bottleneck at scale.
• A row where props is already a dict (not a string), which happens if the column came from a source that partially pre-parsed it: safe_parse passes it through unchanged rather than calling json.loads on a non-string and raising a TypeError.

Direct answer

A genuine left join can never drop rows from the original left DataFrame: it can only fail to
find a match, which produces not-a-number (NaN) on the right-side columns, not a missing row.
So if the row count actually shrank, either the merge is not really a left join (check the
how= argument and that you assigned the result rather than something upstream), or duplicate
keys somewhere caused a downstream drop_duplicates()/aggregation to shrink it after the fact.
The first move either way is merge(..., indicator=True) to see, per row, which side it came
from.

Structured elaboration

Work through this checklist in order; each step is a detection snippet followed by its fix.

1) Confirm counts and get an indicator breakdown

print("orders:", len(orders), "customers:", len(customers))
m = orders.merge(customers, how="left", on="customer_id", indicator=True)
print(m['_merge'].value_counts())
# len(m) should equal len(orders). If it does not, the merge itself is not really a plain
# left join; if it does but many rows are 'left_only', those orders found no match, which
# is a data problem, not a merge bug.

If the key dtypes are incompatible in a hard way (numeric on one side, string on the other),
this call raises ValueError: You are trying to merge on int64 and str columns before you even
see _merge counts, and that error IS the diagnosis: go straight to step 4. Softer mismatches
(matching-looking values, whitespace, invisible characters) do not raise; they merge silently
and show up only as unexpected 'left_only' rows, which is what the rest of this checklist is
for.

2) Rule out an accidental inner join or an overwritten variable

m_inner = orders.merge(customers, how="inner", on="customer_id")
print(len(m_inner), len(orders))

Fix: make sure how="left" is what is actually being called, and that the result is assigned
to a new variable rather than silently reused for something else downstream.

3) Duplicate keys on either side

orders_dup = orders['customer_id'].duplicated(keep=False)
customers_dup = customers['customer_id'].duplicated(keep=False)
print(orders[orders_dup].shape, customers[customers_dup].shape)

Fix: decide the intended semantics. If customers should be a unique reference table, dedupe
it with a business rule (customers.sort_values('updated_at').drop_duplicates('customer_id', keep='last')); if orders legitimately repeats a customer, that is expected and not a bug.

4) Dtype mismatch between key columns

print(orders['customer_id'].dtype, customers['customer_id'].dtype)

A merge on 1 (int) versus '1' (str) matches nothing, silently, because the two values
compare unequal even though they print the same. Fix by casting both sides to the same type:

customers['customer_id'] = customers['customer_id'].astype(orders['customer_id'].dtype)

5) Whitespace and case differences

orders['bad_ws'] = orders['customer_id'].astype(str).str.contains(r'^\s+|\s+$')
print(orders['bad_ws'].sum())

Fix:

for df, col in [(orders, 'customer_id'), (customers, 'customer_id')]:
    df[col] = df[col].astype(str).str.strip().str.lower()

6) Invisible, non-printing characters (non-breaking space, zero-width space)
These characters look identical when printed but are not equal in Python. The standard
library's re module handles this fine; no third-party regex package is needed.

import re

def has_invisible(s):
    return bool(re.search(r'[ ​‌‍]', s))

orders['has_invis'] = orders['customer_id'].astype(str).apply(has_invisible)
print(orders['has_invis'].sum())

Fix (strip the known characters, then Unicode-normalize both sides consistently):

import unicodedata

invis = [' ', '​', '‌', '‍', '']
for ch in invis:
    orders['customer_id'] = orders['customer_id'].str.replace(ch, '', regex=False)
    customers['customer_id'] = customers['customer_id'].str.replace(ch, '', regex=False)

orders['customer_id'] = orders['customer_id'].apply(lambda s: unicodedata.normalize('NFKC', s))
customers['customer_id'] = customers['customer_id'].apply(lambda s: unicodedata.normalize('NFKC', s))

7) Nulls in the key column

print(orders['customer_id'].isna().sum(), customers['customer_id'].isna().sum())

A left merge produces no match for a null key (nulls never equal each other in a join); decide
whether that row should be kept as unmatched, mapped to a sentinel value, or dropped upstream.

8) Confirmed via the validate parameter

orders.merge(customers, how='left', left_on='customer_id', right_on='customer_id', validate='m:1')
# raises pandas.errors.MergeError if customers turns out to have duplicate keys

9) Final reconciliation

m = orders.merge(customers, how='left', on='customer_id', indicator=True)
missing = m[m['_merge'] == 'left_only']
print(missing[['customer_id']].drop_duplicates().head())

Complexity

Each detection step is a single O(n) column scan (dtype check, null check, whitespace/regex
check) or an O(n log n) sort-based step (duplicated(), drop_duplicates()); running the
full checklist end to end is linear in the number of columns scanned, dominated by whichever
step sorts.

Edge cases

Keys that are numerically equal but different dtype (1 vs '1') compare unequal in a merge
even though they print the same; leading/trailing whitespace or invisible Unicode characters
that look identical when printed; null keys, which never match anything, including another
null; and duplicate keys in a table you assumed was a unique reference table.

Trade-offs and pitfalls

• Eyeballing printed values will not catch whitespace or invisible-character mismatches; you
  have to check programmatically.
• Not verifying key dtypes before merging is one of the most common causes here, because pandas
  silently returns zero matches rather than raising an error on a dtype mismatch.
• Skipping the indicator=True check first wastes time chasing deeper causes when the real
  issue (rows that never matched, versus rows that were genuinely dropped) is visible
  immediately.

Direct answer

Build the pipeline as a chain of small, pure functions (each takes a DataFrame in, returns a DataFrame out) and compose them with .pipe(). assign() creates the new is_active column without mutating the input, transform handles the elementwise recency check, and .pipe() is what makes the whole chain readable top-to-bottom and unit-testable step-by-step.

Approach

Four pure steps, each independently testable:

import pandas as pd

def drop_unused(df, cols_to_drop):
    return df.drop(columns=cols_to_drop, errors="ignore")

def add_is_active(df, reference_date, last_login_col="last_login", days=30):
    # reference_date is passed in explicitly (never pd.Timestamp.now() inside
    # the function) so the step is deterministic and testable with a fixed clock
    last_login = pd.to_datetime(df[last_login_col], errors="coerce")
    cutoff = reference_date - pd.Timedelta(days=days)
    return df.assign(**{last_login_col: last_login, "is_active": last_login.gt(cutoff)})

def encode_low_cardinality(df, cat_col):
    codes = df[cat_col].astype("category").cat.codes
    return df.assign(**{f"{cat_col}_code": codes})

def feature_pipeline(df, drop_cols, reference_date, last_login_col="last_login", cat_col="user_type"):
    return (
        df
        .pipe(drop_unused, drop_cols)
        .pipe(add_is_active, reference_date, last_login_col)
        .pipe(encode_low_cardinality, cat_col)
    )

Key points:

• assign() always returns a new DataFrame with the added column rather than mutating in place, which is the pattern pandas 3.0's mandatory Copy-on-Write (CoW, the model where any write on a DataFrame view or slice operates on an independent copy rather than the shared buffer) wants: no ambiguity about whether an in-place mutation touched a shared buffer.
• astype("category").cat.codes gives compact integer codes for a low-cardinality column; unseen or NaN (not-a-number, pandas' missing-value marker) categories become code -1.
• .pipe(func, *args) is just func(df, *args) with the DataFrame threaded through first, so each stage reads left-to-right instead of nesting step3(step2(step1(df))).

Worked example

df = pd.DataFrame({
    "user_id": [1, 2, 3],
    "last_login": ["2026-07-01", "2020-01-01", None],
    "user_type": ["gold", "silver", "gold"],
    "unused_col": ["x", "y", "z"],
})

out = feature_pipeline(df, ["unused_col"], reference_date=pd.Timestamp("2026-07-21"))
print(out)

Output (verified against pandas 3.0.3):

   user_id last_login user_type  is_active  user_type_code
0        1 2026-07-01      gold       True               0
1        2 2020-01-01    silver      False               1
2        3        NaT      gold      False               0

User 1 logged in 20 days before the reference date (inside the 30-day window), user 2's login is over 6 years stale, and user 3's missing login is coerced to NaT (not-a-time) which correctly evaluates False for is_active rather than raising.

Complexity and edge cases

Complexity: each step is a single O(n) pass over the column(s) it touches (drop, datetime coercion, comparison, categorical encoding); .pipe() adds no algorithmic overhead, it is just function composition. Memory: assign and drop each materialize a new DataFrame (a shallow copy of unaffected columns plus the new/removed one), so peak memory is roughly 2x one DataFrame's footprint mid-chain, not n times the chain length, since Python releases the previous intermediate once the next .pipe() call captures its result.

Edge cases: unparseable last_login values become NaT and read as inactive rather than erroring; a categorical column with only one distinct value still encodes fine (single code 0); an empty input DataFrame flows through cleanly and yields an empty output with the right columns since every step operates on whatever rows are present rather than assuming a fixed row count.

Trade-offs and pitfalls

The main benefit of the .pipe()-based design is testability: add_is_active can be unit-tested by calling it directly with a two-row DataFrame and a fixed reference_date, with no need to mock a clock or spin up the full pipeline. Because reference_date is threaded in as an explicit argument instead of read from pd.Timestamp.now() inside the function, the same call always produces the same is_active values, which is what makes the worked example above reproducible days or years later. A common pitfall is baking pd.Timestamp.now() directly into a pipeline step: it makes the function's output depend on wall-clock time, so a unit test asserting an exact is_active value will eventually go stale and fail for reasons unrelated to the code. Passing the "now" as a parameter (with a real default of pd.Timestamp.now() at the call site, not inside the pure function) keeps the function itself deterministic.