Working with Tabular Data at Scale

A table can fit on disk yet require much more memory when loaded, copied, transformed, or joined. Before changing tools, inspect its column types and the operations that create large intermediates. The examples below compare representations, process a file in chunks, and trace the row counts and numerical quantities that an operation produces.

These examples use NumPy and pandas; the Adult example also uses Fairlearn, and the Parquet example uses PyArrow. File examples write into temporary directories and remove them afterward. They intentionally use moderately sized synthetic tables so the comparisons can run locally; file creation and reference calculations are not themselves demonstrations of bounded-memory ingestion.

Inspect column types and memory use

A dtype specifies how a column stores values, including its numeric range and precision. The NumPy-generated numeric columns below start as 64-bit values. Text storage depends on the pandas version and backend: pandas 3 uses a dedicated string dtype by default, with Arrow storage when available. Inspect the printed types instead of assuming that every string column stores Python objects. A MiB is 2²⁰ bytes.

import numpy as np, pandas as pd

rng = np.random.default_rng(0)
n = 2_000_000
df = pd.DataFrame({
    "id":    rng.integers(0, n, n),
    "small": rng.integers(0, 100, n),
    "flag":  rng.integers(0, 2, n),
    "price": rng.normal(50, 10, n),
    "city":  rng.choice(["seoul", "busan", "daegu", "incheon"], n),
})
mb = lambda d: d.memory_usage(deep=True).sum() / 2 ** 20

narrow = df.astype({"id": "int32", "small": "int8", "flag": "bool",
                    "price": "float32", "city": "category"})
print(f"default dtypes  {mb(df):8.1f} MiB")
print(f"narrowed        {mb(narrow):8.1f} MiB   ratio {mb(df) / mb(narrow):.1f}x")
for c in df.columns:
    print(f"  {c:6s} {str(df.dtype):8s} {df.memory_usage(deep=True) / 2**20:7.1f}"
          f"  ->  {str(narrow.dtype):8s} {narrow.memory_usage(deep=True) / 2**20:6.1f}")
# default dtypes      86.8 MiB
# narrowed            21.0 MiB   ratio 4.1x
#   id     int64       15.3  ->  int32       7.6
#   small  int64       15.3  ->  int8        1.9
#   flag   int64       15.3  ->  bool        1.9
#   price  float64     15.3  ->  float32     7.6
#   city   str         25.7  ->  category    1.9

In this run the estimated frame storage falls from 86.8 to 21.0 MiB, about 4.1 times smaller. Compare both the total and the per-column changes. Replacing 64-bit integers by narrower types saves space in the numeric columns, while city has only four repeated values and can use categorical codes. The saving for text depends on its original storage backend, so a ratio measured for Python objects need not hold for Arrow-backed strings. memory_usage(deep=True) estimates the frame’s storage; it is not peak process memory and can account for shared Python objects differently from an operating-system measurement.

The category dtype stores integer codes and a table of distinct values. Repetition makes that table economical. With nearly unique identifiers, the codes and category table can cost as much as or more than another string representation. Check the actual memory use and keep enough capacity for new categories at inference time.

Check ranges and missing-value requirements before narrowing. int8 holds −128 through 127; some casts can wrap out-of-range values, while other assignment paths raise errors. Nullable integer columns need a representation that also supports missing entries. float32 provides roughly seven significant decimal digits, which can lose meaningful distinctions in large-offset features or precise measurements. It also changes rounding during aggregation. Choose precision for the task and consider accumulating sums in float64; narrowing floats is not lossless.

Use array operations for column arithmetic

For this numeric Series, apply(lambda ...) calls a Python function for each element. The equivalent array expression performs arithmetic in compiled loops and avoids those repeated calls, although it can still allocate intermediate arrays. This is a statement about the tested arithmetic, not every pandas operation: text operations and extension dtypes can behave differently. The benchmark checks equal results, warms each path, and reports the best of three timings.

import numpy as np, pandas as pd, time

rng = np.random.default_rng(0)
s = pd.Series(rng.normal(size=200_000))
paths = {
    "apply": lambda: s.apply(lambda v: v * 2 + 1),
    "vectorized": lambda: s * 2 + 1,
    "numpy": lambda: np.asarray(s) * 2 + 1,
}
reference = paths["vectorized"]()
for fn in paths.values():
    np.testing.assert_allclose(fn(), reference)

def best(fn, repeats=3):
    fn()
    times = []
    for _ in range(repeats):
        t = time.perf_counter(); fn(); times.append(time.perf_counter() - t)
    return min(times)

for name, fn in paths.items():
    print(f"{name:10s} {best(fn) * 1000:8.2f} ms")
print("asarray shares storage:", np.shares_memory(np.asarray(s), s.to_numpy(copy=False)))
print("all three results agree:", True)
# apply         44.12 ms # varies by machine
# vectorized     0.21 ms # varies by machine
# numpy          0.11 ms # varies by machine
# asarray shares storage: True
# all three results agree: True

Use the printed timings to compare these three implementations on your machine. The ratio varies with hardware, dtype, allocation, and cache state; it need not increase with row count. Converting a NumPy-backed Series through np.asarray can share its storage, as checked here, rather than copying it. Other dtypes or requested conversions may allocate. The arithmetic after conversion still creates a result array.

For work that needs per-row logic, consider a compiled routine, a lookup table, or batching when those approaches fit the task. A Python loop can still be appropriate for a small dataset, sequential dependencies, or external calls. Preserve the operation’s meaning and measure its cost before reorganizing it.

Aggregate in chunks

Streaming works when each chunk can be summarized by a small state that can be combined with other chunks. Counts and sums add; a mean needs both sum and count; minima combine by taking the smaller value; variance can combine counts, means, and centered sums of squares. Exact medians and quantiles generally require growing storage, extra passes, or external sorting for unrestricted values. Approximate quantile sketches are another option.

import numpy as np, pandas as pd, tempfile
from pathlib import Path

rng = np.random.default_rng(0)
n = 1_000_000
source = pd.DataFrame({"key": rng.integers(0, 50, n), "val": rng.normal(size=n)})
full = source.groupby("key")["val"].mean().reindex(range(50))

with tempfile.TemporaryDirectory() as folder:
    path = Path(folder) / "values.csv"
    source.to_csv(path, index=False)
    del source
    sums = np.zeros(50, dtype=np.float64)
    counts = np.zeros(50, dtype=np.int64)
    largest = 0
    for chunk in pd.read_csv(path, chunksize=100_000, usecols=["key", "val"],
                             dtype={"key": "int64", "val": "float64"}):
        largest = max(largest, len(chunk))
        g = chunk.groupby("key")["val"].agg(["sum", "count"])
        sums[g.index] += g["sum"].to_numpy()
        counts[g.index] += g["count"].to_numpy()
    means = np.divide(sums, counts, out=np.full(50, np.nan), where=counts > 0)
    chunked = pd.Series(means, index=range(50))
    np.testing.assert_allclose(full, chunked, atol=1e-12, rtol=0, equal_nan=True)
    print("means match within absolute tolerance 1e-12:", True)
    print("largest reader chunk:", largest)
    print("rows counted:", int(counts.sum()))
# means match within absolute tolerance 1e-12: True
# largest reader chunk: 100000
# rows counted: 1000000

This demonstration first creates a CSV and computes a reference mean, then deletes the source DataFrame before reading the file in chunks. The reader processes at most 100,000 rows in each chunk. That is not a measurement of peak memory: parser buffers, grouped results, and retained allocator memory also count, and the setup initially held the full dataset. In a production stream the file would already exist. The result is compared with the reference using an explicit floating-point tolerance.

The mean must be combined from sums and counts, not by averaging chunk means equally. For example, means 2 and 8 from chunks of 10 and 90 rows combine to (10×2 + 90×8)/100 = 7.4, not 5. Here the keys are known integers 0 through 49; the arrays hold one sum and count per key. With many distinct keys, that aggregation state can itself outgrow memory. For variance, use centered summary statistics rather than subtracting two large raw moments; the second exercise implements a stable chunk-combination formula.

Reading fewer columns and pushing suitable filters or aggregations toward the data source can reduce work before it reaches Python. Parquet supports column-oriented access; the last exercise compares it with CSV. If the remaining data and intermediate state still exceed memory, consider an external-memory query engine or distributed processing. The appropriate choice depends on the operations, storage, and deployment constraints.

Order of operations

An early filter saves work when it leaves the meaning of the calculation unchanged. The product below depends only on each row, and the filter uses an existing column, so it can run first. Moving a filter before a rolling calculation, global normalization, or ranking can change the answer and requires separate reasoning.

import numpy as np, pandas as pd, time

rng = np.random.default_rng(0)
n = 2_000_000
df = pd.DataFrame({"a": rng.normal(size=n), "b": rng.normal(size=n),
                   "g": rng.integers(0, 10, n)})

def compute_first():
    return df.assign(c=df.a * df.b).query("g == 3")["c"].mean()

def filter_first():
    sub = df.query("g == 3")
    return (sub.a * sub.b).mean()

def best(fn, repeats=3):
    fn()
    times = []
    for _ in range(repeats):
        t = time.perf_counter(); fn(); times.append(time.perf_counter() - t)
    return min(times)

print(f"compute then filter {best(compute_first) * 1000:7.1f} ms")
print(f"filter then compute {best(filter_first) * 1000:7.1f} ms")
print("same answer:", bool(np.isclose(compute_first(), filter_first(), rtol=0, atol=1e-12)))
# compute then filter    33.5 ms # varies by machine
# filter then compute    27.6 ms # varies by machine
# same answer: True

About a tenth of the generated rows pass the filter, so the second version computes far fewer products. End-to-end runtime also includes filtering and allocations; pandas’ copy behavior can depend on its version. The timing comparison alone does not establish which component dominates or imply a tenfold upper limit on speedup. Both paths return the same result within the stated tolerance.

Applying a compatible filter in a database query can reduce the rows serialized, transferred, and parsed by the client. Whether the source also avoids scanning those rows depends on indexes, partitions, file statistics, and its execution engine.

The same saving, measured on real data

The Adult dataset provides a second memory comparison. The loader may download it on first use. We deliberately expand its categorical columns to Python-object storage and its numeric columns to int64, then narrow them. This is a controlled comparison of representations, not a claim about the loader’s default types. Converting to object preserves the existing missing markers; converting everything to strings could change their meaning.

import numpy as np, pandas as pd
from fairlearn.datasets import fetch_adult

X = fetch_adult(as_frame=True).data.copy()
cats = X.select_dtypes(exclude=[np.number]).columns
nums = X.select_dtypes(include=[np.number]).columns
raw = X.copy()
for c in cats:
    raw = raw.astype(object)
for c in nums:
    raw = raw.astype("int64")

opt = raw.copy()
for c in cats:
    opt = opt.astype("category")
for c in nums:
    opt = pd.to_numeric(opt, downcast="integer")

pd.testing.assert_frame_equal(raw.isna(), opt.isna())
for c in cats:
    mask = raw.notna()
    assert (raw.loc[mask, c] == opt.loc[mask, c].astype(object)).all()
for c in nums:
    np.testing.assert_array_equal(raw, opt)
before = raw.memory_usage(deep=True).sum() / 2**20
after = opt.memory_usage(deep=True).sum() / 2**20
print(f"object + int64      : {before:8.2f} MiB")
print(f"category + downcast : {after:8.2f} MiB   {before / after:.1f}x smaller")
print("values and missingness preserved:", True)
# object + int64      :    23.86 MiB
# category + downcast :     0.98 MiB   24.3x smaller
# values and missingness preserved: True

This run measures 23.86 MiB for the explicitly constructed object/int64 baseline and 0.98 MiB after conversion, about 24.3 times smaller. The equality and missingness checks confirm that the conversions preserve this dataset’s observed values; dtype metadata changes. Numeric downcasting succeeds here because the values are integral and fit the chosen ranges. The reduction depends on the starting representation, and future inputs may require wider ranges.

Storage estimates for the final frame are only part of the capacity plan. During conversion, both the original and converted frames can coexist, and later joins or model fitting may allocate additional copies. To benefit before memory is exhausted, specify appropriate types at ingestion or convert manageable chunks. A smaller frame does not by itself establish that the entire workload will fit on a laptop.

Exercises

1. What a join multiplies. Merge two small frames with duplicate keys on both sides and count the output rows. Then compute the output size for two 200,000-row tables with exactly 200 rows per key on each side and the same 1,000 keys.

You should get: an output larger than either input, and an estimate in the tens of millions.

import pandas as pd

left = pd.DataFrame({"user": [1, 1, 2, 2, 3], "order": range(5)})
right = pd.DataFrame({"user": [1, 1, 1, 2, 2], "addr": list("abcde")})
merged = left.merge(right, on="user", how="inner")

print(f"left {len(left)} x right {len(right)} -> merged {len(merged)}")
print("per-user counts:", left.groupby("user").size().to_dict(),
      right.groupby("user").size().to_dict())
print(f"balanced shared keys: 200k x 200k -> {200_000 * 200_000 // 1000:,} rows")
# left 5 x right 5 -> merged 10
# per-user counts: {1: 2, 2: 2, 3: 1} {1: 3, 2: 2}
# balanced shared keys: 200k x 200k -> 40,000,000 rows

2. Streaming variance, two ways. Compute the variance of a large array in one pass by accumulating \(\sum x\) and \(\sum x^2\), and by combining centered chunk statistics, on data whose mean is large relative to its spread. Compare both against the two-pass answer.

Compare numerical agreement. The raw-moment result can vary across environments because it subtracts nearly equal large numbers.

import numpy as np

rng = np.random.default_rng(0)
x = rng.normal(0, 1.0, 1_000_000) + 1e8        # float64 throughout
reference = np.var(x)

s1 = s2 = 0.0; n = 0                           # the textbook one-pass formula
for chunk in np.array_split(x, 100):
    s1 += chunk.sum(); s2 += (chunk ** 2).sum(); n += len(chunk)
naive = s2 / n - (s1 / n) ** 2

mean = 0.0; m2 = 0.0; count = 0                # combine centered chunk statistics
for v in np.array_split(x, 100):
    k = len(v); delta = v.mean() - mean; total = count + k
    m2 += v.var() * k + delta ** 2 * count * k / total
    mean += delta * k / total; count = total
combined = m2 / count

print(f"two-pass np.var   {reference:.10f}")
print(f"naive sums        {naive:.10f}")
print(f"combined chunks   {combined:.10f}")
centered = x - x[0]
shifted = np.mean(centered ** 2) - np.mean(centered) ** 2
print(f"centered moments  {shifted:.10f}")
assert np.isclose(combined, reference, rtol=1e-7, atol=0)
assert np.isclose(shifted, reference, rtol=1e-7, atol=0)
# two-pass np.var   1.0013441256
# naive sums        2.0000000000
# combined chunks   1.0013441256
# centered moments  1.0013441256

3. Column pruning at the read. Write a 50-column frame to Parquet and to CSV, then read back two columns from each. Warm up the readers first, and report both the two-column and the all-column times.

Compare column-subset and full reads. Report the environment and treat timings as observations, not fixed expected outputs.

import numpy as np, pandas as pd, time, tempfile
from pathlib import Path

rng = np.random.default_rng(0)
n, p = 300_000, 50
df = pd.DataFrame(rng.normal(size=(n, p)).astype(np.float32),
                  columns=[f"c{i}" for i in range(p)])

def best(fn, k=3):
    times = []
    for _ in range(k):
        t = time.perf_counter(); fn(); times.append(time.perf_counter() - t)
    return min(times)

with tempfile.TemporaryDirectory() as folder:
    pq, csv = Path(folder) / "x.parquet", Path(folder) / "x.csv"
    df.to_parquet(pq, engine="pyarrow", index=False)
    df.to_csv(csv, index=False)
    pq_two = pd.read_parquet(pq, columns=["c0", "c1"])
    csv_two = pd.read_csv(csv, usecols=["c0", "c1"])
    np.testing.assert_array_equal(pq_two.to_numpy(), df[["c0", "c1"]].to_numpy())
    np.testing.assert_allclose(csv_two.to_numpy(), pq_two.to_numpy(), rtol=1e-6, atol=1e-7)
    print("restored dtype: parquet", pq_two["c0"].dtype, "csv", csv_two["c0"].dtype)
    print(f"parquet {pq.stat().st_size / 2**20:6.1f} MiB   csv {csv.stat().st_size / 2**20:6.1f} MiB")
    print(f"2 of 50 cols : parquet {best(lambda: pd.read_parquet(pq, columns=['c0','c1'])) * 1000:7.1f} ms"
          f"   csv {best(lambda: pd.read_csv(csv, usecols=['c0','c1'])) * 1000:7.1f} ms")
    print(f"all 50 cols  : parquet {best(lambda: pd.read_parquet(pq)) * 1000:7.1f} ms"
          f"   csv {best(lambda: pd.read_csv(csv)) * 1000:7.1f} ms")
# restored dtype: parquet float32 csv float64
# parquet   84.3 MiB   csv  156.1 MiB
# 2 of 50 cols : parquet     6.3 ms   csv   718.4 ms # varies by machine
# all 50 cols  : parquet   115.2 ms   csv  1599.4 ms # varies by machine