Parsing AWS CUR Parquet Files with Python Pandas

Production FinOps automation routinely fractures at the parse stage when engineers call pandas.read_parquet() against AWS Cost and Usage Report (CUR) Parquet exports with default settings. The bottleneck is rarely S3 throughput — it is uncontrolled memory allocation during schema inference and aggressive expansion of the JSON-encoded resource_tags column. Against a CUR object larger than 2 GB, the naive path triggers MemoryError inside a constrained worker pod, silently narrows DECIMAL(18,6) cost columns to float64, and re-reads partitions the pipeline has already processed. This page solves that specific problem: a deterministic, PyArrow-backed parse that enforces an explicit schema, preserves exact decimal precision, defers tag normalization until after filtering, and streams record batches so a 10 GB part file processes in a 2–4 GB container. It is the Parquet-specific deep dive for the parse step of the AWS CUR to Data Lake Pipeline.

# Root Cause & Failure Modes

CUR Parquet exports use a wide, append-only schema: a single report can carry several hundred columns, and AWS adds new ones without notice. Three of pandas’ convenience defaults turn that schema into production incidents.

1. Tag explosion and OOM. The line_item_resource_tags (and resource_tags/user:*) data arrives as JSON strings. Engineers reflexively feed it to pd.json_normalize(), which materializes every distinct tag key as its own column. In a multi-account organization with tens of thousands of unique keys — many ephemeral or developer-generated — this produces a sparse matrix whose width scales with tag cardinality, exhausting RAM before type casting even begins. The worker is OOM-killed and the pipeline SLA is missed.
2. Decimal precision loss. pandas.read_parquet() casts DECIMAL(18,6) cost columns to IEEE-754 float64. Rounding drift at the $0.000001 boundary compounds across millions of line items, and the monthly allocation total no longer reconciles against the AWS invoice to the cent — a finding that fails audit and breaks any cross-cloud cost allocation ledger downstream.
3. Whole-file materialization and re-reads. read_parquet() loads the entire object into one DataFrame, and without partition awareness the job re-scans historical months on every run. AWS finalizes a billing period over a 24–48 hour window, so the parser must read only the partitions in play and skip the rest.

The fix is to bypass pandas’ inference engine entirely: scan with pyarrow.dataset, push filters down to prune partitions, pin every column type with an explicit pa.schema, convert to pandas with types_mapper=pd.ArrowDtype so decimals survive, and expand tags row-wise only after the batch is filtered. This mirrors the memory-bounded contract the parent reference defines for Cloud Billing Data Ingestion & Parsing.

# Production Pipeline Architecture

The parse stage decomposes into four phases, each with a single responsibility, sitting between manifest discovery upstream and partitioned persistence downstream.

• Phase 1 — Bounded scan. pyarrow.dataset opens the export as a logical dataset and a scanner streams it as record batches sized by batch_size, never the whole file. Partition filter expressions are pushed down so irrelevant months are skipped at the storage layer.
• Phase 2 — Schema projection. An explicit pa.schema pins identity, temporal, and financial columns to their exact types, guaranteeing decimal128(18, 6) survives and rejecting silent coercion.
• Phase 3 — Arrow-backed conversion. Each batch converts to pandas via types_mapper=pd.ArrowDtype, keeping costs as decimal.Decimal-backed Arrow types rather than float64.
• Phase 4 — Deferred tag normalization. Only after filtering, the JSON tag column is parsed row-wise into a single dict column, restricted to business-relevant prefixes, then the raw JSON is dropped to free memory before the batch is yielded.

The upstream manifest resolution that hands this stage its file list is covered in the parent AWS CUR to Data Lake Pipeline; the same trailing-window, watermark logic for “which partitions are still mutating” appears on the GCP side in Incremental Sync Strategies for GCP Billing Exports. Once parsed, these batches feed daily roll-ups described in Time-Series Aggregation for Daily Cloud Cost Tracking.

# Step-by-Step Python Implementation

The module below targets containerized workers with a hard memory limit. It enforces the schema, streams record batches, preserves decimals, and parses tags safely. Dependencies: pyarrow, pandas. It is idempotent — re-running over the same partition filter yields identical output — and degrades loudly on malformed tag JSON rather than crashing the batch.

import gc
import json
import logging
from typing import Iterator, Dict, Optional, Set

import pyarrow as pa
import pyarrow.dataset as ds
import pyarrow.compute as pc
import pandas as pd

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("cur.parquet")

# Explicit CUR schema. Pinning types here is what stops pandas from
# narrowing DECIMAL(18,6) to float64 or re-inferring tag columns.
CUR_SCHEMA = pa.schema([
    ("identity_line_item_id", pa.string()),
    ("line_item_usage_account_id", pa.string()),
    ("line_item_usage_start_date", pa.timestamp("us", tz="UTC")),
    ("line_item_usage_end_date", pa.timestamp("us", tz="UTC")),
    ("line_item_unblended_cost", pa.decimal128(18, 6)),
    ("line_item_blended_cost", pa.decimal128(18, 6)),
    ("line_item_resource_tags", pa.string()),
    ("line_item_product_code", pa.string()),
    ("line_item_usage_type", pa.string()),
])


def parse_tags_safe(tag_json: str, allowed_prefixes: Optional[Set[str]] = None) -> Dict[str, str]:
    """Parse a CUR JSON tag string without exploding into a sparse matrix.

    Restricting to business prefixes (cost_center:, env:, project:) keeps the
    output width bounded regardless of how many ephemeral keys exist upstream.
    """
    if not tag_json or tag_json == "{}":
        return {}
    try:
        raw = json.loads(tag_json)
    except (json.JSONDecodeError, TypeError) as exc:
        logger.warning("invalid resource_tags JSON, skipping row: %s", exc)
        return {}
    if allowed_prefixes:
        return {
            k: str(v) for k, v in raw.items()
            if any(k.startswith(p) for p in allowed_prefixes)
        }
    return {k: str(v) for k, v in raw.items() if isinstance(v, (str, int, float))}


def stream_cur_partitions(
    cur_path: str,
    batch_size: int = 250_000,
    tag_prefixes: Optional[Set[str]] = None,
    partition_filter: Optional[ds.Expression] = None,
    use_threads: bool = True,
) -> Iterator[pd.DataFrame]:
    """Partition-pruned, record-batch CUR parse with explicit schema enforcement.

    Yields Arrow-backed DataFrames so DECIMAL precision is preserved end to end.
    Memory stays bounded by batch_size rather than file size.
    """
    logger.info("opening pyarrow dataset: %s", cur_path)
    dataset = ds.dataset(cur_path, format="parquet", partitioning="hive")

    # Project only the columns we declared; pushing the filter down lets the
    # scanner skip whole row groups and partitions at the storage layer.
    scanner = dataset.scanner(
        columns=[f.name for f in CUR_SCHEMA],
        filter=partition_filter,
        batch_size=batch_size,
        use_threads=use_threads,
    )

    rows_seen = 0
    for batch_idx, batch in enumerate(scanner.to_batches()):
        if batch.num_rows == 0:
            continue
        # ArrowDtype keeps decimals exact; NumPy backend would cast to float64.
        df = batch.to_pandas(types_mapper=pd.ArrowDtype)

        # Deferred, bounded tag expansion AFTER filtering, never before.
        df["parsed_tags"] = df["line_item_resource_tags"].apply(
            lambda x: parse_tags_safe(x, tag_prefixes)
        )
        df = df.drop(columns=["line_item_resource_tags"])

        rows_seen += len(df)
        logger.info("batch %d -> %d rows (cumulative %d)", batch_idx, len(df), rows_seen)
        yield df
        del df
        gc.collect()

    logger.info("scan complete: %d rows across %s", rows_seen, cur_path)


def main() -> None:
    # Process only the finalizing month; historical partitions are skipped.
    partition_filter = (pc.field("year") == 2026) & (pc.field("month") == 5)
    tag_prefixes = {"cost_center:", "env:", "project:"}

    total = 0
    for df in stream_cur_partitions(
        "s3://finops-cur-exports/parquet/",
        partition_filter=partition_filter,
        tag_prefixes=tag_prefixes,
    ):
        # Hand each Arrow-backed batch to the persistence stage (Iceberg, Delta,
        # BigQuery). Costs remain decimal128 here, not float64.
        total += int(df["line_item_unblended_cost"].notna().sum())
    logger.info("parsed %d costed line items", total)


if __name__ == "__main__":
    main()

# Why these choices

• types_mapper=pd.ArrowDtype keeps PyArrow’s native types instead of NumPy equivalents, so line_item_unblended_cost stays exact. See the pandas PyArrow backend guide for the dtype semantics.
• scanner.to_batches() streams configurable chunks instead of one giant DataFrame, so a multi-gigabyte part file processes without swapping.
• Prefix-filtered tags cap output width at known business domains, preventing column proliferation from throwaway keys.

# Verification & Testing

Confirm correctness before wiring the parser into the persistence stage:

• Decimal fidelity assertion. After conversion, assert the cost column is still Arrow-backed and reconciles to the source. A fast check: assert str(df["line_item_unblended_cost"].dtype) == "decimal128(18, 6)[pyarrow]", then compare df["line_item_unblended_cost"].sum() against the manifest’s reported total — it must match to six decimal places, not “approximately”.
• Memory dry-run. Run the scan against the largest part file with batch_size set low and sample RSS between yields with psutil.Process().memory_info().rss. Peak resident memory should track batch_size, not file size; if it grows monotonically a downstream consumer is retaining batch references.
• Idempotency check. Run the same partition_filter twice and hash the concatenated output (row count plus a checksum of sorted identity_line_item_id). Identical hashes prove the parse is deterministic and safe to replay against the 24–48 hour finalization window.
• Tag-bound assertion. Assert df["parsed_tags"].map(len).max() never exceeds the count of allowed_prefixes domains in play — a regression here means a prefix filter was bypassed and the sparse-matrix failure mode is back.

# Common Pitfalls Checklist

• Calling pd.json_normalize() on the tag column. It materializes every key as a column and OOMs at scale — keep tags as a single dict column and expand only the prefixes you allocate on.
• Trusting read_parquet() for money. It returns float64 and quietly loses cents — always go through pyarrow.dataset with types_mapper=pd.ArrowDtype.
• No partition filter. Without filter pushdown the job re-scans every historical month each run — pass a pyarrow.compute expression bound to the finalizing year/month.
• use_threads=True in a tiny pod. Thread-pool overhead can push a 2 GB pod over its limit — set use_threads=False when memory is the binding constraint, not CPU.
• Ignoring additive schema drift. AWS appends columns silently; projecting a fixed column list (as above) tolerates new columns, but a hard-coded positional read does not — never index CUR columns by position.

# Frequently Asked Questions

# Related

• AWS CUR to Data Lake Pipeline — the parent reference; this page is the Parquet-specific deep dive for its parse stage.
• Cloud Billing Data Ingestion & Parsing — the four-stage acquisition→normalization→allocation→persistence model this parser implements.
• Incremental Sync Strategies for GCP Billing Exports — the GCP analogue of CUR’s finalization window, using watermark-bounded partition fetches.
• Time-Series Aggregation for Daily Cloud Cost Tracking — the downstream roll-up stage that consumes these decimal-exact parsed batches.
• Cross-Cloud Cost Allocation Strategies — why decimal fidelity here is non-negotiable for an auditable allocation ledger.

Up: AWS CUR to Data Lake Pipeline · Home: Cloud Cost Optimization & FinOps Automation