Normalizing Multi-Cloud Billing Schemas

AWS CUR line items, GCP BigQuery billing export rows, and Azure Cost Management exports describe the same underlying fact — money spent against a resource, in a period — through three incompatible schemas with different field names, cost-type semantics, and time granularity. line_item/UnblendedCost is not cost, which is not CostInBillingCurrency; lineItem/UsageStartDate is hourly, usage_start_time is provider-aggregated, and Azure’s export defaults to daily. This page covers the normalization stage of the pipeline defined in Cloud Billing Data Ingestion & Parsing: the engineering problem is not fetching the data — that is solved by acquisition — it is converging three schemas into one canonical dimensional model that allocation, amortization, and anomaly detection can consume without knowing which cloud a row came from. Get this wrong and every downstream number is provider-shaped instead of business-shaped: a multi-cloud showback report either double-counts credits, mixes currencies, or silently drops Azure’s amortized reservation cost because nobody mapped EffectiveCost.

# Architecture Context & Data-Flow Position

Normalization sits second in the four-stage pipeline, immediately after acquisition and before allocation. Acquisition — the AWS CUR to Data Lake Pipeline, the GCP BigQuery Billing Export Sync, and the Azure Cost Management API Integration — delivers raw, provider-shaped rows. Normalization’s isolation contract is strict: it accepts only raw rows plus a mapping-registry version, and it emits only canonical records — it never talks to a cloud API and never applies tag policy or shared-cost splits, which are allocation’s job. That separation is what lets you re-normalize a full historical dataset against a corrected mapping without re-pulling a single byte from S3 or BigQuery.

The convergence point is a single canonical record shape. Every provider adapter, regardless of source schema, must populate the same eight core fields: billing_period, service, resource_id, usage_type, unblended_cost, amortized_cost, currency, and a flattened tags map. Anything a provider exposes that does not map to this model — AWS’s lineItemType, GCP’s sku.id, Azure’s MeterCategory — is either folded into usage_type/service or dropped after being logged, never silently forwarded.

The table below is the top-level convergence map — the detailed per-field type and constraint breakdown lives in the Schema Reference Table further down, and the JSON-Schema-backed resolver itself is covered in Mapping Provider Cost Fields to a Unified Schema.

# Core Implementation Patterns

# 1. The Mapping Registry as a Versioned Artifact

Hard-coding field names inline in adapter code makes a provider’s schema change a silent production incident — a renamed CUR column returns None instead of raising. The fix is to externalize every field mapping into a single registry object with a schema_version, stored in Git and code-reviewed like any other contract:

MAPPING_REGISTRY = {
    "schema_version": "2026.07.1",
    "aws": {"unblended_cost": "lineItem/UnblendedCost", "amortized_cost": "lineItem/NetUnblendedCost"},
    "gcp": {"unblended_cost": "cost", "credits_field": "credits"},
    "azure": {"unblended_cost": "CostInBillingCurrency", "amortized_cost": "EffectiveCost"},
}

Every canonical record carries the schema_version that produced it. When a provider adds a field or renames one, you bump the version, add a migration entry, and re-normalize affected periods — you never mutate history in place under an unversioned mapping, because that makes a re-run silently non-reproducible.

# 2. Currency Conversion and a Pinned Settlement Currency

AWS CUR rows arrive per-account in the account’s billing currency; GCP export rows carry a currency_conversion_rate alongside local-currency cost; Azure exposes both CostInBillingCurrency and CostInPricingCurrency. Multi-currency estates cannot be summed as-is. The registry stores the native currency and amount unchanged — never overwritten — and a separate FX step multiplies into a pinned settlement currency (typically USD) using a dated rate table, so historical reports stay reproducible even as spot rates move:

def to_settlement(amount: Decimal, rate: Decimal) -> Decimal:
    # Quantize to 6 places to match provider invoice precision before summation.
    return (amount * rate).quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)

# 3. Cost-Type Reconciliation: Blended, Unblended, Amortized, Net

The four cost types answer different questions and providers do not expose them uniformly. Unblended cost is what an individual linked account actually pays before any consolidated-billing averaging — AWS’s default and the canonical model’s baseline (lineItem/UnblendedCost, cost, CostInBillingCurrency). Blended cost averages Reserved Instance and Savings Plan rates across an AWS consolidated billing family; it exists only on AWS and is excluded from the canonical model because GCP and Azure have no equivalent — mixing it in would make cross-cloud totals incomparable. Amortized cost spreads upfront reservation and commitment charges evenly across their term instead of spiking the purchase month; AWS exposes it as reservation/EffectiveCost / savingsPlan/SavingsPlanEffectiveCost, GCP folds committed-use discounts into the credits array, and Azure requires requesting the AmortizedCost dataset explicitly (the default ActualCost dataset omits it). The join logic that attaches amortization schedules to consumption is detailed in Reserved Instance Mapping Logic. Net cost is unblended cost plus every credit, discount, and refund — cost + SUM(credits.amount) on GCP, lineItem/NetUnblendedCost on AWS. The canonical model populates both unblended_cost and amortized_cost explicitly rather than picking one, so allocation logic — see Cross-Cloud Cost Allocation Strategies — can choose per report which lens to aggregate on.

# 4. Tag and Label Flattening

AWS exposes user tags as dozens of individually-prefixed CUR columns (resourceTags/user:cost-center), GCP as a repeated key/value record array, and Azure as a JSON object column. All three collapse into the canonical model’s single tags: Dict[str, str] map, with provider-reserved keys (AWS aws:, Azure hidden-) stripped at normalization time so they never reach tag-validation gates as if they were user-authored.

# Production-Grade Python Normalization Engine

The module below loads the versioned mapping registry, runs a per-provider adapter that reads raw rows through it, reconciles cost type and currency, and emits deduplicated canonical records. pydantic validates and coerces every field at construction time, so a malformed row fails loudly at normalization rather than corrupting a downstream aggregate. Dependencies: pydantic>=2.4, tenacity>=8.2.

"""Multi-cloud billing normalization engine.

Converges AWS CUR, GCP BigQuery billing export, and Azure Cost Management
rows into one canonical cost record via a versioned mapping registry.
"""
import hashlib
import json
import logging
from abc import ABC, abstractmethod
from datetime import date, datetime
from decimal import Decimal, ROUND_HALF_UP
from typing import Any, Dict, List, Optional

from pydantic import BaseModel, field_validator
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

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

# ---------------------------------------------------------------------------
# Versioned mapping registry — the single source of truth for field names.
# Bump schema_version on any provider field addition, rename, or removal.
# ---------------------------------------------------------------------------
MAPPING_REGISTRY: Dict[str, Any] = {
    "schema_version": "2026.07.1",
    "aws": {
        "service": "product/ProductName",
        "resource_id": "lineItem/ResourceId",
        "usage_type": "lineItem/UsageType",
        "usage_start": "lineItem/UsageStartDate",
        "unblended_cost": "lineItem/UnblendedCost",
        "amortized_cost": "lineItem/NetUnblendedCost",
        "currency": "lineItem/CurrencyCode",
        "tag_prefix": "resourceTags/user:",
    },
    "gcp": {
        "service": "service.description",
        "resource_id": "sku.id",
        "usage_type": "sku.description",
        "usage_start": "usage_start_time",
        "unblended_cost": "cost",
        "credits_field": "credits",
        "currency": "currency",
        "label_field": "labels",
    },
    "azure": {
        "service": "ConsumedService",
        "resource_id": "ResourceId",
        "usage_type": "MeterCategory",
        "usage_start": "Date",
        "unblended_cost": "CostInBillingCurrency",
        "amortized_cost": "EffectiveCost",
        "currency": "BillingCurrency",
        "tag_field": "Tags",
    },
}


class CanonicalCostRecord(BaseModel):
    """The unified dimensional model every provider adapter converges into."""

    provider: str
    schema_version: str
    billing_period: str          # ISO month, "2026-07"
    usage_start: datetime
    granularity: str             # "hourly" | "daily" — daily is the safe floor
    service: str
    resource_id: Optional[str] = None
    usage_type: Optional[str] = None
    unblended_cost: Decimal
    amortized_cost: Decimal
    currency: str
    tags: Dict[str, str] = {}
    row_hash: str = ""

    @field_validator("unblended_cost", "amortized_cost", mode="before")
    @classmethod
    def _coerce_decimal(cls, v: Any) -> Decimal:
        # Providers emit cost as string, float, or Decimal depending on SDK;
        # coercing through str() avoids float binary-representation drift.
        return Decimal(str(v)) if v not in (None, "") else Decimal("0")

    def with_fingerprint(self) -> "CanonicalCostRecord":
        # Content hash over the fields that define identity; used for
        # idempotent dedup across replayed or overlapping source batches.
        basis = "|".join([
            self.provider, self.billing_period, self.usage_start.isoformat(),
            self.service, str(self.resource_id), str(self.unblended_cost),
        ])
        return self.model_copy(update={"row_hash": hashlib.sha256(basis.encode()).hexdigest()})


FX_RETRYABLE = (ConnectionError, TimeoutError)


@retry(
    retry=retry_if_exception_type(FX_RETRYABLE),
    wait=wait_exponential(multiplier=1, min=1, max=15),
    stop=stop_after_attempt(4),
    reraise=True,
)
def fetch_fx_rate(currency: str, as_of: date, _rates: Dict[str, Decimal]) -> Decimal:
    """Dated FX lookup to the pinned settlement currency (USD).

    In production this calls a rate service; the in-memory table here keeps
    the example runnable. Retrying protects against transient network
    failures without masking a genuinely missing currency.
    """
    if currency == "USD":
        return Decimal("1")
    try:
        return _rates[(currency, as_of)]
    except KeyError as exc:
        raise ValueError(f"no FX rate for {currency} on {as_of}") from exc


class ProviderAdapter(ABC):
    """Base contract: read one raw row through the mapping registry."""

    provider: str

    def __init__(self, fx_rates: Dict[str, Decimal]) -> None:
        self.mapping = MAPPING_REGISTRY[self.provider]
        self.schema_version = MAPPING_REGISTRY["schema_version"]
        self.fx_rates = fx_rates

    @abstractmethod
    def normalize(self, raw: Dict[str, Any]) -> CanonicalCostRecord:
        ...

    def _flatten_tags(self, raw: Dict[str, Any]) -> Dict[str, str]:
        return {}


class AwsCurAdapter(ProviderAdapter):
    provider = "aws"

    def normalize(self, raw: Dict[str, Any]) -> CanonicalCostRecord:
        m = self.mapping
        usage_start = datetime.fromisoformat(raw[m["usage_start"]].replace("Z", "+00:00"))
        record = CanonicalCostRecord(
            provider=self.provider,
            schema_version=self.schema_version,
            billing_period=usage_start.strftime("%Y-%m"),
            usage_start=usage_start,
            granularity="hourly",
            service=raw[m["service"]],
            resource_id=raw.get(m["resource_id"]) or None,
            usage_type=raw.get(m["usage_type"]),
            unblended_cost=raw.get(m["unblended_cost"], "0"),
            # Fall back to unblended when a row has no RI/SP amortization component.
            amortized_cost=raw.get(m["amortized_cost"]) or raw.get(m["unblended_cost"], "0"),
            currency=raw.get(m["currency"], "USD"),
            tags=self._flatten_tags(raw),
        )
        return record.with_fingerprint()

    def _flatten_tags(self, raw: Dict[str, Any]) -> Dict[str, str]:
        prefix = self.mapping["tag_prefix"]
        return {k[len(prefix):]: v for k, v in raw.items() if k.startswith(prefix) and v}


class GcpBillingExportAdapter(ProviderAdapter):
    provider = "gcp"

    def normalize(self, raw: Dict[str, Any]) -> CanonicalCostRecord:
        m = self.mapping
        usage_start = raw[m["usage_start"]]
        if isinstance(usage_start, str):
            usage_start = datetime.fromisoformat(usage_start.replace("Z", "+00:00"))
        credits_total = sum(
            (Decimal(str(c.get("amount", 0))) for c in raw.get(m["credits_field"], [])),
            start=Decimal("0"),
        )
        unblended = Decimal(str(raw.get(m["unblended_cost"], 0)))
        record = CanonicalCostRecord(
            provider=self.provider,
            schema_version=self.schema_version,
            billing_period=usage_start.strftime("%Y-%m"),
            usage_start=usage_start,
            granularity="daily",
            service=raw[m["service"]],
            resource_id=raw.get(m["resource_id"]),
            usage_type=raw.get(m["usage_type"]),
            unblended_cost=unblended,
            amortized_cost=unblended + credits_total,  # credits are negative
            currency=raw.get(m["currency"], "USD"),
            tags=self._flatten_tags(raw),
        )
        return record.with_fingerprint()

    def _flatten_tags(self, raw: Dict[str, Any]) -> Dict[str, str]:
        labels = raw.get(self.mapping["label_field"]) or []
        return {lbl["key"]: lbl["value"] for lbl in labels}


class AzureCostManagementAdapter(ProviderAdapter):
    provider = "azure"

    def normalize(self, raw: Dict[str, Any]) -> CanonicalCostRecord:
        m = self.mapping
        usage_start = datetime.fromisoformat(str(raw[m["usage_start"]]))
        record = CanonicalCostRecord(
            provider=self.provider,
            schema_version=self.schema_version,
            billing_period=usage_start.strftime("%Y-%m"),
            usage_start=usage_start,
            granularity="daily",  # Cost Management exports do not offer hourly
            service=raw[m["service"]],
            resource_id=raw.get(m["resource_id"]),
            usage_type=raw.get(m["usage_type"]),
            unblended_cost=raw.get(m["unblended_cost"], "0"),
            # EffectiveCost only exists under the AmortizedCost dataset request.
            amortized_cost=raw.get(m["amortized_cost"]) or raw.get(m["unblended_cost"], "0"),
            currency=raw.get(m["currency"], "USD"),
            tags=self._flatten_tags(raw),
        )
        return record.with_fingerprint()

    def _flatten_tags(self, raw: Dict[str, Any]) -> Dict[str, str]:
        tags = raw.get(self.mapping["tag_field"]) or {}
        return {k: v for k, v in tags.items() if not k.startswith("hidden-")}


class NormalizationEngine:
    """Orchestrates per-provider adapters with idempotent, hash-based dedup."""

    def __init__(self, fx_rates: Optional[Dict[str, Decimal]] = None) -> None:
        fx_rates = fx_rates or {}
        self.adapters: Dict[str, ProviderAdapter] = {
            "aws": AwsCurAdapter(fx_rates),
            "gcp": GcpBillingExportAdapter(fx_rates),
            "azure": AzureCostManagementAdapter(fx_rates),
        }
        self._seen_hashes: set = set()  # swap for a persistent store in production

    def normalize_batch(self, provider: str, raw_rows: List[Dict[str, Any]]) -> List[CanonicalCostRecord]:
        adapter = self.adapters[provider]
        out: List[CanonicalCostRecord] = []
        for raw in raw_rows:
            try:
                record = adapter.normalize(raw)
            except (KeyError, ValueError) as exc:
                # A missing mapped field means the registry is stale for this
                # provider's current schema — fail loud, do not coerce to null.
                logger.error("normalization failed for %s row: %s", provider, exc)
                continue
            if record.row_hash in self._seen_hashes:
                logger.debug("skip duplicate row_hash=%s", record.row_hash[:12])
                continue
            self._seen_hashes.add(record.row_hash)
            out.append(record)
        logger.info("normalized %d/%d %s rows (schema_version=%s)",
                    len(out), len(raw_rows), provider, MAPPING_REGISTRY["schema_version"])
        return out


def main() -> None:
    sample_aws_row = {
        "product/ProductName": "Amazon Elastic Compute Cloud",
        "lineItem/ResourceId": "i-0abc123",
        "lineItem/UsageType": "BoxUsage:m5.large",
        "lineItem/UsageStartDate": "2026-07-01T00:00:00Z",
        "lineItem/UnblendedCost": "4.32",
        "lineItem/NetUnblendedCost": "3.10",
        "lineItem/CurrencyCode": "USD",
        "resourceTags/user:cost-center": "platform",
    }
    engine = NormalizationEngine()
    records = engine.normalize_batch("aws", [sample_aws_row])
    for r in records:
        print(json.dumps(r.model_dump(mode="json"), default=str))


if __name__ == "__main__":
    main()

The design choices that matter most: Decimal coercion happens once, in a field_validator, so every adapter is free of float arithmetic; the row_hash fingerprint makes re-normalizing an overlapping batch a no-op rather than a duplicate; and each adapter fails loud on a KeyError/ValueError instead of writing a partially-null canonical record. Pagination and rate-limit handling for the Azure source rows this engine consumes are covered in Azure Cost Management API Integration; the pandas-based parquet read that produces the AWS raw-row dictionaries is covered in Parsing AWS CUR Parquet Files with Python & pandas.

# Schema Reference Table

# Operational Considerations

• Granularity floor. AWS CUR can deliver hourly rows, but Azure Cost Management exports and GCP’s practical export cadence are daily. The canonical model normalizes every provider down to the coarsest common grain — daily — for cross-cloud reports; keep raw hourly AWS data in a separate fine-grained table for AWS-only drill-downs rather than upsampling Azure or GCP to match it.
• Registry version pinning. Every canonical record stores the schema_version that produced it. Never re-run an old batch through a newer registry silently — bump the version, add an explicit migration, and re-normalize the affected billing_period range so historical figures stay reproducible and auditable.
• FX rate staleness. Currency conversion must use the rate dated to the usage_start period, not the rate at normalization time — providers restate historical rows, and re-normalizing a restated GCP row with today’s spot rate instead of the period rate silently shifts prior-period totals.
• Credits sign convention. GCP’s credits.amount and Azure’s discount line items are negative; a normalization bug that treats them as positive doubles reported spend instead of netting it out. Assert amortized_cost <= unblended_cost as a cheap sanity check on every batch.
• Untagged and malformed rows. A row failing field_validator coercion is logged and dropped rather than raising the whole batch — track a per-run drop-rate metric and alert above 0.5%, since a sudden spike almost always means an upstream schema change the registry has not caught up to.
• Throughput. A single normalization pass processes tens of thousands of rows per second in pure Python for the dict-based adapters above; at genuinely high volume (multi-million row daily CUR batches) push the same field-mapping logic into a vectorized pandas/polars transform rather than a per-row Python loop, keeping the mapping registry as the shared source of truth for both paths.

# Troubleshooting

Cross-cloud total is higher than the sum of provider invoices. Root cause: amortized_cost was computed by adding GCP credits as positive values, or by summing AWS UnblendedCost and NetUnblendedCost instead of treating the latter as the net figure. Detection: amortized_cost > unblended_cost for a nontrivial fraction of rows. Remediation: assert the sign convention in the adapter (credits.amount is always ≤ 0) and add the amortized_cost <= unblended_cost guard to the batch validator.

Azure amortized cost is always equal to unblended cost. Root cause: the Cost Management query requested the default ActualCost dataset, which does not include EffectiveCost; the adapter’s fallback silently masked the missing field. Detection: amortized_cost == unblended_cost for every Azure row with an active reservation. Remediation: switch the export/query type to AmortizedCost for reservation-bearing scopes, per Azure Cost Management API Integration.

Historical month total changes after a re-run with no new source data. Root cause: the FX rate table was queried at run time instead of at the period’s dated rate, so a re-normalization months later picks up a different spot rate. Detection: cost_usd for a closed period diverges between two runs against identical raw rows. Remediation: key the FX lookup on (currency, usage_start.date()) and treat the rate table itself as append-only and versioned, never mutated in place.

Registry mapping silently returns None for a new provider field. Root cause: AWS, GCP, or Azure added or renamed an export column and raw.get(...) swallowed the KeyError that should have failed the row. Detection: a spike in rows with resource_id=None or usage_type=None after a provider export version bump. Remediation: use raw[...] (not .get()) for fields the registry marks required, so a missing mapped column raises immediately and is caught by the adapter’s except (KeyError, ValueError) logging path instead of propagating a null.

Duplicate canonical records after re-ingesting an overlapping batch. Root cause: the dedup store (_seen_hashes in the reference engine) is process-local and does not survive a restart or scale across workers. Detection: SUM(unblended_cost) for a period exceeds the provider invoice after a re-run. Remediation: back the fingerprint store with a persistent, shared table (Postgres unique constraint or DynamoDB conditional write) keyed on row_hash, mirroring the state-registry pattern used at the acquisition stage.

# Frequently Asked Questions

# Related

• Cloud Billing Data Ingestion & Parsing — the parent reference defining the four-stage pipeline this normalization stage implements.
• Mapping Provider Cost Fields to a Unified Schema — the deep dive on the registry resolver, versioning, and migration mechanics behind this page’s mapping table.
• Parsing AWS CUR Parquet Files with Python & pandas — produces the raw AWS row dictionaries the AwsCurAdapter consumes.
• Azure Cost Management API Integration — the acquisition-stage source of the raw Azure rows normalized here, including the AmortizedCost dataset request.
• Cross-Cloud Cost Allocation Strategies — how allocation consumes the canonical record’s unblended_cost/amortized_cost split downstream of this stage.
• Reserved Instance Mapping Logic — the amortization-schedule join logic that populates amortized_cost for commitment-bearing resources.

Up: Cloud Billing Data Ingestion & Parsing