AWS CUR to Data Lake Pipeline

The AWS Cost and Usage Report (CUR) is the authoritative ledger for AWS spend: hourly, resource-level line items with pricing, amortization, and discount detail that no console view exposes. This page covers the AWS acquisition stage of the billing pipeline end to end — the mechanism is an S3-delivered, manifest-indexed export of compressed CSV or Parquet objects, and the engineering problem is turning that raw drop into a partitioned, catalogued data lake without race conditions, out-of-memory failures, or duplicated invoices. It implements the AWS branch of the four-stage model defined in Cloud Billing Data Ingestion & Parsing: acquisition, normalization, allocation, persistence. Unlike the Cost Explorer GetCostAndUsage API, CUR pushes data to you, so there is no polling loop — but you inherit mid-delivery file splits, manifest versioning, and schema drift that a naive ListObjectsV2 walk will silently corrupt.

Architecture Context & Data-Flow Position

CUR ingestion is the first stage of the AWS pipeline. AWS writes a new set of report objects to a configured S3 prefix on a sub-daily cadence, each accompanied by a JSON manifest. The manifest — not the bucket listing — is the single source of truth for which objects constitute a billing period, because AWS splits a large report into many part files and rewrites the set as the period finalizes over 24–48 hours. The pipeline therefore reads the manifest, validates reportStatus, deduplicates against already-processed artifacts, streams each part into a memory-bounded parser, writes optimized Parquet, and registers the partition with the AWS Glue Data Catalog so Athena and Spark can prune scans. Everything downstream — tag-based allocation, commitment amortization, anomaly detection — assumes this stage delivered an exactly-once, schema-stable record set.

AWS CUR acquisition stage, manifest to Athena A serpentine dataflow for the AWS Cost and Usage Report acquisition stage. An S3 ObjectCreated event on the manifest triggers the manifest resolver, which reads reportKeys, reportStatus, and compression and defers the run if status is IN_PROGRESS. The idempotency check compares each part's ETag checksum against a DynamoDB high-water mark. A memory-bounded streaming parser reads chunked record batches under a retry-and-backoff loop on the S3 GetObject call, the partitioned Parquet writer keys output by billing period, the Glue catalog register creates the partition, and Athena runs a partition-pruned scan. defer if IN_PROGRESS retry · backoff on GetObject 1 · S3 ObjectCreated 2 · Manifest resolver 3 · Idempotency check 4 · Streaming parser *-Manifest.json reportKeys · status DynamoDB high-water mark memory-bounded chunks 5 · Parquet writer 6 · Glue register 7 · Athena query billing_period partition create_partition partition-pruned scan stream parts
The AWS CUR acquisition stage: a manifest event drives resolution, an idempotency check against a DynamoDB high-water mark, memory-bounded streaming under GetObject backoff, partitioned Parquet output, Glue registration, and a pruned Athena scan.

The manifest is the contract. Its fields drive every downstream decision:

Manifest / object field Type Purpose Constraint
reportKeys array[string] Ordered S3 keys of every part file in the report May grow mid-period as AWS splits large reports
reportStatus string Delivery state of the report set Process only when not IN_PROGRESS
compression string GZIP, ZIP, or Parquet Dictates the reader path
billingPeriod.start string Period start (YYYYMMDD...) Becomes the partition key
schemaElements array Ordered column contract for the report version Drifts when AWS adds columns
ETag (per object) string MD5 for single-part uploads Multipart objects carry a -N suffix

This is the same manifest-driven discovery the parent reference summarizes for AWS; the sibling channels — GCP BigQuery Billing Export Sync and Azure Cost Management API Integration — solve the same problem with partition pruning and cursor pagination respectively. For near-real-time figures rather than the finalized historical record CUR provides, the throttling profile of the Cost Explorer API is documented under AWS Cost Explorer Architecture.

Core Implementation Patterns

1. Least-Privilege IAM and Cross-Account Access

CUR is frequently delivered to a central billing account while consumption lives in member accounts. The ingestion compute (Lambda, Fargate, EKS, or EC2) must assume a scoped role rather than carry static keys. Grant only s3:GetObject and s3:ListBucket on the CUR prefix, plus glue:CreatePartition/glue:BatchCreatePartition on the target table. Hardcoded credentials violate the baseline; attached IAM roles plus STS AssumeRole for cross-account reads are the only acceptable pattern.

import boto3
from botocore.config import Config


def cur_s3_client(role_arn: str | None = None, region: str = "us-east-1"):
    """Return an S3 client using the task's IAM role, or an assumed cross-account role."""
    adaptive = Config(retries={"max_attempts": 5, "mode": "adaptive"}, region_name=region)
    if role_arn is None:
        return boto3.client("s3", config=adaptive)

    sts = boto3.client("sts")
    creds = sts.assume_role(RoleArn=role_arn, RoleSessionName="cur-ingest")["Credentials"]
    return boto3.client(
        "s3",
        aws_access_key_id=creds["AccessKeyId"],
        aws_secret_access_key=creds["SecretAccessKey"],
        aws_session_token=creds["SessionToken"],
        config=adaptive,
    )

2. Manifest Resolution and File Discovery

Parse the manifest to build an authoritative, ordered queue of part files. Reject reports that are still delivering, and surface compression so the parser selects the correct reader. Never enumerate the prefix directly — a ListObjectsV2 walk can capture a half-written report mid-split.

import json
import logging
from botocore.exceptions import ClientError

logger = logging.getLogger("cur.ingest")


def resolve_cur_manifest(s3, bucket: str, manifest_key: str) -> dict:
    """Parse the CUR manifest into authoritative file paths and metadata."""
    try:
        response = s3.get_object(Bucket=bucket, Key=manifest_key)
        manifest = json.loads(response["Body"].read().decode("utf-8"))
    except ClientError as exc:
        logger.error("failed to fetch CUR manifest %s: %s", manifest_key, exc)
        raise

    status = manifest.get("reportStatus", {})
    if isinstance(status, dict) and status.get("status") == "IN_PROGRESS":
        raise ValueError(f"report {manifest_key} still delivering; defer processing")

    return {
        "compression": manifest.get("compression", "GZIP"),
        "files": manifest.get("reportKeys", []),
        "columns": [c.get("name") for c in manifest.get("schemaElements", [])],
        "billing_period": manifest.get("billingPeriod", {}).get("start", "unknown"),
    }

3. Resilient Download with Backoff and Jitter

S3 GetObject hits transient SlowDown, RequestTimeout, and VPC-endpoint throttling on large reports. Wrap reads in exponential backoff with jitter, capped so a stalled object cannot block the batch indefinitely. The adaptive retry mode on the client handles most cases; this loop covers the throttle codes that warrant a longer pause.

import io
import time
import random
from botocore.exceptions import ClientError

THROTTLE_CODES = {"Throttling", "ThrottlingException", "SlowDown", "RequestTimeout"}


def download_with_backoff(s3, bucket: str, key: str, max_attempts: int = 5) -> io.BytesIO:
    """Stream an S3 object into memory with exponential backoff and jitter."""
    for attempt in range(1, max_attempts + 1):
        try:
            body = s3.get_object(Bucket=bucket, Key=key)["Body"]
            buffer = io.BytesIO()
            for chunk in body.iter_chunks(chunk_size=8 * 1024 * 1024):
                buffer.write(chunk)
            buffer.seek(0)
            return buffer
        except ClientError as exc:
            if exc.response["Error"]["Code"] not in THROTTLE_CODES:
                raise
            delay = min(2 ** attempt + random.uniform(0, 1), 30)
            logger.warning("throttled on %s; retry %d/%d in %.1fs", key, attempt, max_attempts, delay)
            time.sleep(delay)
    raise RuntimeError(f"max retries exceeded downloading {key}")

4. Memory-Bounded Parsing and Dimension Mapping

Uncompressed CUR part files routinely exceed 10 GB. Loading one whole into a DataFrame triggers OOM kills and blows the pipeline SLA. Read in bounded chunks (pyarrow record batches for Parquet, chunked pandas for CSV), coerce monetary columns explicitly, and map the verbose CUR column names onto the canonical dimensional model before writing. Type coercion here is what stops a stray string in lineItem/UnblendedCost from failing an Athena scan three queries downstream.

import io
import pandas as pd
from typing import Iterator

# CUR source column -> canonical dimension. Keep this map versioned alongside the manifest schema.
COLUMN_MAP = {
    "lineItem/UsageAccountId": "account_id",
    "lineItem/UsageType": "usage_type",
    "lineItem/UnblendedCost": "unblended_cost",
    "bill/BillingPeriodStartDate": "billing_period_start",
    "product/ProductName": "service",
}


def stream_parse_cur(buffer: io.BytesIO, compression: str, chunk_rows: int = 500_000) -> Iterator[pd.DataFrame]:
    """Memory-bounded chunked CSV parse with explicit monetary coercion and column mapping."""
    comp = "gzip" if compression.upper() == "GZIP" else None
    with pd.read_csv(buffer, compression=comp, chunksize=chunk_rows, on_bad_lines="skip") as reader:
        for chunk in reader:
            present = {src: dst for src, dst in COLUMN_MAP.items() if src in chunk.columns}
            chunk = chunk[list(present)].rename(columns=present)
            chunk["unblended_cost"] = pd.to_numeric(
                chunk.get("unblended_cost", 0), errors="coerce"
            ).fillna(0.0)
            yield chunk

For the full Parquet-specific column-mapping treatment — predicate pushdown, dictionary encoding, and pyarrow schema enforcement — see Parsing AWS CUR Parquet Files with Python Pandas.

Production-Grade Python Ingestion Engine

The module below ties the patterns together into one self-contained orchestrator. It resolves the manifest, enforces exactly-once processing through a state registry keyed on object ETag, streams each part into bounded Parquet writes partitioned by billing period and account, and registers the result with Glue. Swap the SQLite registry for DynamoDB with a conditional write for distributed locking. Dependencies: boto3, pandas, pyarrow.

import hashlib
import io
import json
import logging
import random
import sqlite3
import time
from dataclasses import dataclass
from functools import wraps
from typing import Iterator, List

import boto3
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from botocore.config import Config
from botocore.exceptions import BotoCoreError, ClientError

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

THROTTLE_CODES = {"Throttling", "ThrottlingException", "SlowDown", "RequestTimeout"}


def with_backoff(max_attempts: int = 5):
    """Retry decorator: exponential backoff with jitter on transient S3/Glue throttling."""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except (ClientError, BotoCoreError) as exc:
                    code = getattr(exc, "response", {}).get("Error", {}).get("Code", "")
                    if code not in THROTTLE_CODES or attempt == max_attempts:
                        raise
                    delay = min(2 ** attempt + random.uniform(0, 1), 30)
                    logger.warning("throttled (%s); retry %d/%d in %.1fs", code, attempt, max_attempts, delay)
                    time.sleep(delay)
            raise RuntimeError("unreachable")
        return wrapper
    return decorator


@dataclass(frozen=True)
class CurArtifact:
    """One CUR part file plus the metadata that drives idempotent processing."""

    bucket: str
    key: str
    checksum: str
    billing_period: str
    compression: str


class IngestStateRegistry:
    """Exactly-once tracker. INSERT ... ON CONFLICT makes a duplicate trigger a no-op.

    A key match with a *different* checksum means AWS restated the report and the
    artifact must be re-processed; identical checksum means skip.
    """

    def __init__(self, db_path: str = "cur_state.db") -> None:
        self.conn = sqlite3.connect(db_path)
        self.conn.execute(
            """
            CREATE TABLE IF NOT EXISTS processed_artifacts (
                artifact_key TEXT PRIMARY KEY,
                checksum     TEXT NOT NULL,
                processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
            """
        )
        self.conn.commit()

    def is_processed(self, key: str, checksum: str) -> bool:
        row = self.conn.execute(
            "SELECT checksum FROM processed_artifacts WHERE artifact_key = ?", (key,)
        ).fetchone()
        return row is not None and row[0] == checksum

    def mark_processed(self, key: str, checksum: str) -> None:
        self.conn.execute(
            """
            INSERT INTO processed_artifacts (artifact_key, checksum) VALUES (?, ?)
            ON CONFLICT(artifact_key) DO UPDATE SET
                checksum = excluded.checksum,
                processed_at = CURRENT_TIMESTAMP
            """,
            (key, checksum),
        )
        self.conn.commit()


COLUMN_MAP = {
    "lineItem/UsageAccountId": "account_id",
    "lineItem/UsageType": "usage_type",
    "lineItem/UnblendedCost": "unblended_cost",
    "bill/BillingPeriodStartDate": "billing_period_start",
    "product/ProductName": "service",
}

OUTPUT_SCHEMA = pa.schema([
    ("account_id", pa.string()),
    ("usage_type", pa.string()),
    ("unblended_cost", pa.float64()),
    ("billing_period_start", pa.string()),
    ("service", pa.string()),
])


class CurDataLakePipeline:
    """Resolve a CUR manifest, stream parts to partitioned Parquet, register with Glue."""

    def __init__(self, state: IngestStateRegistry, region: str = "us-east-1") -> None:
        cfg = Config(retries={"max_attempts": 5, "mode": "adaptive"}, region_name=region)
        self.s3 = boto3.client("s3", config=cfg)
        self.glue = boto3.client("glue", config=cfg)
        self.state = state

    @with_backoff()
    def _resolve_manifest(self, bucket: str, manifest_key: str) -> dict:
        raw = self.s3.get_object(Bucket=bucket, Key=manifest_key)["Body"].read()
        manifest = json.loads(raw.decode("utf-8"))
        status = manifest.get("reportStatus", {})
        if isinstance(status, dict) and status.get("status") == "IN_PROGRESS":
            raise ValueError(f"{manifest_key} still delivering; defer")
        return manifest

    @staticmethod
    def _checksum(head: dict) -> str:
        # ETag == MD5 for single-part uploads; multipart carries a "-N" suffix and
        # needs a composite fallback so restatements are still detected.
        etag = head.get("ETag", "").strip('"')
        if "-" in etag:
            return f"{head.get('ContentLength')}:{head.get('LastModified')}"
        return etag

    @with_backoff()
    def _download(self, bucket: str, key: str) -> io.BytesIO:
        body = self.s3.get_object(Bucket=bucket, Key=key)["Body"]
        buffer = io.BytesIO()
        for chunk in body.iter_chunks(chunk_size=8 * 1024 * 1024):
            buffer.write(chunk)
        buffer.seek(0)
        return buffer

    @staticmethod
    def _parse(buffer: io.BytesIO, compression: str, chunk_rows: int = 500_000) -> Iterator[pd.DataFrame]:
        comp = "gzip" if compression.upper() == "GZIP" else None
        with pd.read_csv(buffer, compression=comp, chunksize=chunk_rows, on_bad_lines="skip") as reader:
            for chunk in reader:
                present = {s: d for s, d in COLUMN_MAP.items() if s in chunk.columns}
                chunk = chunk[list(present)].rename(columns=present)
                chunk["unblended_cost"] = pd.to_numeric(
                    chunk.get("unblended_cost", 0), errors="coerce"
                ).fillna(0.0)
                yield chunk

    def discover(self, bucket: str, manifest_key: str) -> List[CurArtifact]:
        manifest = self._resolve_manifest(bucket, manifest_key)
        period = manifest.get("billingPeriod", {}).get("start", "unknown")[:8]
        compression = manifest.get("compression", "GZIP")
        artifacts: List[CurArtifact] = []
        for key in manifest.get("reportKeys", []):
            head = self.s3.head_object(Bucket=bucket, Key=key)
            artifacts.append(CurArtifact(bucket, key, self._checksum(head), period, compression))
        return artifacts

    def process(self, artifact: CurArtifact, output_prefix: str) -> str | None:
        if self.state.is_processed(artifact.key, artifact.checksum):
            logger.info("skip unchanged artifact %s", artifact.key)
            return None

        buffer = self._download(artifact.bucket, artifact.key)
        tables = []
        for chunk in self._parse(buffer, artifact.compression):
            cols = [f.name for f in OUTPUT_SCHEMA if f.name in chunk.columns]
            tables.append(pa.Table.from_pandas(chunk[cols], preserve_index=False))

        if not tables:
            self.state.mark_processed(artifact.key, artifact.checksum)
            return None

        combined = pa.concat_tables(tables, promote_options="default")
        part_name = hashlib.sha1(artifact.key.encode()).hexdigest()[:12]
        out_path = f"{output_prefix}/billing_period={artifact.billing_period}/{part_name}.parquet"
        pq.write_table(combined, out_path, compression="snappy",
                       row_group_size=1_000_000, use_dictionary=True)
        self.state.mark_processed(artifact.key, artifact.checksum)
        logger.info("wrote %d rows -> %s", combined.num_rows, out_path)
        return out_path

    @with_backoff()
    def register_partition(self, database: str, table: str, period: str, location: str) -> None:
        try:
            self.glue.create_partition(
                DatabaseName=database,
                TableName=table,
                PartitionInput={
                    "Values": [period],
                    "StorageDescriptor": {
                        "Location": location,
                        "InputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
                        "OutputFormat": "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat",
                        "SerdeInfo": {
                            "SerializationLibrary":
                                "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"
                        },
                    },
                },
            )
            logger.info("registered partition %s at %s", period, location)
        except self.glue.exceptions.AlreadyExistsException:
            logger.info("partition %s already registered", period)


def main() -> None:
    state = IngestStateRegistry("cur_state.db")
    pipeline = CurDataLakePipeline(state)
    artifacts = pipeline.discover(
        bucket="finops-cur-exports",
        manifest_key="cur/monthly/finops-cur-Manifest.json",
    )
    for artifact in artifacts:
        out = pipeline.process(artifact, output_prefix="/lake/cur")
        if out:
            pipeline.register_partition(
                database="finops", table="cur_line_items",
                period=artifact.billing_period,
                location=out.rsplit("/", 1)[0],
            )


if __name__ == "__main__":
    main()

Schema Reference Table

The CUR exposes hundreds of columns; this is the minimal allocation-grade subset mapped onto the canonical model the rest of the pipeline consumes. Keep this map versioned next to the manifest’s schemaElements so an added column is a code review, not a production surprise.

CUR source column Normalized field Type Notes
lineItem/UsageAccountId account_id string Member account that incurred the usage; primary partition for showback
bill/BillingPeriodStartDate billing_period_start string (ISO) Period partition key; aligns with finalization window
lineItem/UnblendedCost unblended_cost decimal Store as Decimal/float64; never narrow to float32
lineItem/UsageType usage_type string Encodes region + operation (e.g. USE1-BoxUsage:t3.large)
product/ProductName service string Human-readable service; cluster on this for query pruning
lineItem/ResourceId resource_id string (nullable) Empty unless resource-level CUR is enabled
resourceTags/user:* tags{} map Flattened user tags; feeds tag-validation gating
savingsPlan/SavingsPlanEffectiveCost amortized_cost decimal Use for amortized views, not blended/unblended

Operational Considerations

  • Delivery cadence and finalization. CUR can be delivered hourly or daily, but cost data finalizes 24–48 hours after usage as credits, refunds, and tax settle. Treat the trailing two days as provisional and re-ingest the period as AWS restates it; the checksum-keyed registry makes that overwrite deterministic.
  • File size and splitting. A single account-month CUR can exceed 10 GB uncompressed and is split into many part files. The reportKeys array grows during the period — re-read the manifest each run rather than caching the file list.
  • Cost-aware compute. CUR processing is batch and interruption-tolerant, so run it on Spot-backed Fargate or Graviton Lambda. Partition pruning by billing_period keeps Athena scans — and Athena’s $5/TB-scanned charge — bounded.
  • Schema evolution. AWS adds CUR columns without notice. Validate the incoming header against the versioned contract and fail loudly on an unknown required field; allow additive columns through a registry update. Watch the CUR column changelog.
  • Monitoring hooks. Emit CloudWatch metrics for FilesProcessed, BytesTransformed, SchemaValidationFailures, and PipelineDuration, and alarm on validation-failure rate to catch upstream format changes before they reach dashboards. The retry-and-quota behaviour behind the backoff decorator is detailed in Handling Billing API Rate Limits & Retries.

Troubleshooting

Athena query returns partial or zero rows after a successful load. Root cause: the Glue partition was never registered, so the catalog cannot see the new objects. Detection: SELECT count(*) returns rows for old periods but not the latest. Remediation: run MSCK REPAIR TABLE cur_line_items or call register_partition for the missing billing_period; verify the S3 location in the partition matches the written prefix exactly.

OutOfMemory / pod OOM-killed during parse. Root cause: a part file was read whole instead of in chunks. Detection: the worker dies on the largest part file, smaller ones succeed. Remediation: confirm chunksize (CSV) or record-batch iteration (Parquet) is in effect and lower chunk_rows; size the container so peak chunk memory plus the PyArrow table fits with headroom.

Duplicated costs in showback after a re-run. Root cause: append-on-replay without idempotency, or a partition overwritten with WRITE_APPEND semantics. Detection: month-to-date total exceeds the AWS invoice by a clean multiple. Remediation: gate every part on the state registry’s is_processed checksum check, and write each artifact to a deterministic, idempotent object key so a replay overwrites rather than appends.

reportStatus shows IN_PROGRESS or the manifest 404s. Root cause: the pipeline triggered on a part-file ObjectCreated event before the manifest finalized. Detection: intermittent NoSuchKey on the manifest or a status guard rejection. Remediation: trigger only on the manifest key (*-Manifest.json) ObjectCreated event, not on part-file events, and defer with a short backoff when status is IN_PROGRESS.

Negative or zero unblended_cost rows flood the lake. Root cause: credits, refunds, and tax line items are legitimate CUR rows, not parse errors. Detection: spikes of zero-cost rows at month boundaries. Remediation: keep them — filtering credits breaks reconciliation against the invoice; segment by lineItem/LineItemType (Credit, Tax, Usage) in downstream allocation instead.

Frequently Asked Questions

Why parse the manifest instead of listing the S3 prefix?

AWS splits a large report into many part files and rewrites the set as the period finalizes. A ListObjectsV2 prefix scan can capture a half-written report and miss or double-count parts. The manifest’s reportKeys array is the only authoritative, consistent view of which objects belong to a billing period.

How do I handle the 24–48 hour finalization window?

Treat the most recent two days as provisional and re-ingest the period as AWS restates it. Because each artifact is keyed by ETag checksum in the state registry, a restated part (same key, new checksum) is re-processed while an unchanged part is skipped — so re-ingestion overwrites provisional data without creating duplicates.

Should I use CUR or the Cost Explorer API?

Use CUR for the granular, finalized, resource-level historical record that allocation and amortization need; it is the cheapest path for full-fidelity data. Use the Cost Explorer GetCostAndUsage API for low-latency, aggregated figures and near-real-time alerting, accepting its 5-requests-per-second default throttle.

Why store cost as Decimal rather than float?

Floating-point rounding drift compounds across millions of line items and produces allocation totals that fail to reconcile against the AWS invoice to the cent. Decimal (or a carefully bounded float64 only at the Parquet boundary) preserves exact base-10 values required for auditable financial reporting.

Up: Cloud Billing Data Ingestion & Parsing · Home: Cloud Cost Optimization & FinOps Automation