Mapping Provider Cost Fields to a Unified Schema

The specific bottleneck this page solves is field-name divergence across billing providers colliding with a hardcoded transform. AWS Cost and Usage Report (CUR) calls unblended cost line_item_unblended_cost. The GCP BigQuery billing export calls the same concept cost. Azure Cost Management calls it costInBillingCurrency. A if provider == "aws": row["cost"] = raw["line_item_unblended_cost"] chain works for exactly as long as no provider ships a schema revision — then AWS renames a reservation field, Azure adds a new pricingModel enum value, and the branch logic silently drops or mis-maps a column while every downstream rollup keeps running without error. The fix is to stop hardcoding the transform and instead declare it: a versioned field-mapping registry that a resolver interprets at runtime, validated against a target schema and checked against golden records on every change. This is the field-level counterpart to the pipeline-level normalization work described in Normalizing Multi-Cloud Billing Schemas — that page owns the four-phase pipeline; this page owns the mapping registry that phase two of it calls.

Root Cause & Failure Modes

Three naive approaches to cross-provider field mapping each fail in a specific, quiet way:

  • Inline conditional branches. if provider == "aws": ... elif provider == "gcp": ... scattered through the ingestion code. Every new provider field requires a code change and a deploy. Nobody remembers to update all four places a field is referenced (transform, validation, tests, docs), so the mapping and the code drift within a quarter.
  • Positional column renaming. Renaming columns by DataFrame position (df.columns = [...]) after a raw load. This breaks the instant a provider reorders columns or a report schema version adds a new field in the middle — which AWS CUR does when it introduces a new resource_tags/user:* key, shifting every field defined by ordinal position.
  • Untyped dict-to-dict copy. canonical["cost"] = raw.get("line_item_unblended_cost", 0) with a bare default. This masks missing-field errors as zero cost instead of failing loudly, and a 0 unblended cost is indistinguishable from a genuinely free line item — the exact ambiguity that lets a broken export slip through until someone notices the monthly invoice reconciliation is off by six figures.

The correct approach treats the provider-to-canonical mapping as data, not code. A mapping spec — one YAML or dict per provider version — declares which raw field feeds which canonical field, its expected type, and whether it is required. A single resolver function interprets that spec against any raw row. Adding a provider or a schema revision means adding a mapping entry, not touching the resolver, and a jsonschema validation pass plus a golden-record regression test catch drift before it reaches production. The three providers rarely agree on units either: AWS and Azure report cost as decimal currency; GCP’s export nests cost alongside a credits array that must be summed separately to reach net cost — the registry has to carry that transform, not just the field name.

Production Pipeline Architecture

The mapping resolver sits inside the normalization stage of Cloud Billing Data Ingestion & Parsing, immediately after provider-specific parsing — such as the row extraction covered in Parsing AWS CUR Parquet Files with Python Pandas — and before allocation. It runs in four phases:

  1. Registry load. Load a versioned mapping spec per provider (aws_cur_v1, gcp_export_v1, azure_cost_mgmt_v1), each a list of field-mapping entries: source_field, target_field, value_type, required, and an optional transform name for fields that need more than a rename (currency rounding, credit summation, enum translation).
  2. Row resolution. For each raw row, the resolver walks the active provider’s mapping entries, pulls the source field (supporting dotted paths for nested JSON such as GCP’s sku.id), applies the named transform if present, and casts to the declared value_type.
  3. Schema validation. The resolved canonical row is validated against a jsonschema definition of the unified cost model. A required field that resolves to None, or a type mismatch (a string landing in a float field), raises immediately rather than propagating a corrupted row downstream.
  4. Golden-record check. In CI, the same resolver runs against a small fixed set of real anonymized rows per provider with known-correct canonical output. Any mapping-spec change that alters that output is a diff a reviewer sees before merge, not a silent production surprise.

The registry format below is intentionally provider-agnostic — it feeds the same resolver used for cost allocation joins in Cross-Cloud Cost Allocation Strategies, which assumes every row already carries the unified field names before it groups spend by owner.

Provider → canonical field table

Canonical field AWS CUR field GCP export field Azure Cost Mgmt field Type Notes
cost_amount line_item_unblended_cost cost costInBillingCurrency decimal GCP requires summing credits[].amount separately for net cost
usage_amount line_item_usage_amount usage.amount quantity decimal Units vary by SKU; not cross-comparable without the unit field
billing_period bill_billing_period_start_date invoice.month billingPeriodStartDate date AWS is a timestamp, GCP is YYYYMM string, Azure is ISO date
resource_id line_item_resource_id resource.name resourceId string Required in canonical model; nullable in all three sources
service_name product_servicename disambiguated via line_item_product_code service.description consumedService string Free-text; not enum-stable across providers
sku_id line_item_usage_type sku.id meterId string No shared vocabulary — kept as an opaque provider-native key
account_id line_item_usage_account_id project.id subscriptionGuid string The allocation join key downstream
charge_type line_item_line_item_type cost_type chargeType enum Mapped through a shared {Usage, Tax, Credit, Refund} enum via transform

Step-by-Step Python Implementation

The module below defines the mapping spec as plain dicts (swap in YAML-loaded specs for production without changing the resolver), resolves a raw provider row into a CanonicalCostRecord dataclass, validates it with jsonschema, and includes a golden-record regression test executed from __main__.

import logging
from dataclasses import dataclass, asdict
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Any, Callable, Optional

from jsonschema import validate, ValidationError

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)


@dataclass
class CanonicalCostRecord:
    cost_amount: Decimal
    usage_amount: Decimal
    billing_period: str
    resource_id: Optional[str]
    service_name: str
    sku_id: str
    account_id: str
    charge_type: str


# jsonschema definition of the unified model — required fields raise loudly
# rather than letting a mis-mapped None reach the allocation join.
CANONICAL_SCHEMA = {
    "type": "object",
    "required": ["cost_amount", "billing_period", "service_name", "sku_id", "account_id", "charge_type"],
    "properties": {
        "cost_amount": {"type": "string"},   # Decimal serialized as string for exact validation
        "usage_amount": {"type": "string"},
        "billing_period": {"type": "string"},
        "resource_id": {"type": ["string", "null"]},
        "service_name": {"type": "string", "minLength": 1},
        "sku_id": {"type": "string", "minLength": 1},
        "account_id": {"type": "string", "minLength": 1},
        "charge_type": {"enum": ["Usage", "Tax", "Credit", "Refund"]},
    },
}


def _get_dotted(raw: dict, path: str) -> Any:
    """Resolve a dotted source path (e.g. 'sku.id') against a nested provider row."""
    node = raw
    for part in path.split("."):
        if not isinstance(node, dict) or part not in node:
            return None
        node = node[part]
    return node


def _sum_gcp_credits(raw: dict) -> Decimal:
    """GCP's cost field excludes credits; net cost requires summing the credits array."""
    base = Decimal(str(raw.get("cost", 0)))
    credits_total = sum(Decimal(str(c.get("amount", 0))) for c in raw.get("credits", []) or [])
    return base + credits_total  # credits are negative amounts in the export


def _aws_charge_type(raw: dict) -> str:
    return {"Usage": "Usage", "Tax": "Tax", "Credit": "Credit", "Refund": "Refund"}.get(
        raw.get("line_item_line_item_type", "Usage"), "Usage"
    )


def _gcp_charge_type(raw: dict) -> str:
    return {"regular": "Usage", "tax": "Tax", "adjustment": "Credit"}.get(
        raw.get("cost_type", "regular"), "Usage"
    )


# The registry: one entry list per provider/version. Each entry declares where the
# value comes from, what canonical field it feeds, its type, whether it's required,
# and an optional named transform for anything beyond a straight rename/cast.
MAPPING_REGISTRY: dict[str, list[dict]] = {
    "aws_cur_v1": [
        {"source": "line_item_unblended_cost", "target": "cost_amount", "type": "decimal", "required": True},
        {"source": "line_item_usage_amount", "target": "usage_amount", "type": "decimal", "required": False},
        {"source": "bill_billing_period_start_date", "target": "billing_period", "type": "str", "required": True},
        {"source": "line_item_resource_id", "target": "resource_id", "type": "str", "required": False},
        {"source": "product_servicename", "target": "service_name", "type": "str", "required": True},
        {"source": "line_item_usage_type", "target": "sku_id", "type": "str", "required": True},
        {"source": "line_item_usage_account_id", "target": "account_id", "type": "str", "required": True},
        {"source": None, "target": "charge_type", "type": "str", "required": True, "transform": _aws_charge_type},
    ],
    "gcp_export_v1": [
        {"source": None, "target": "cost_amount", "type": "decimal", "required": True, "transform": _sum_gcp_credits},
        {"source": "usage.amount", "target": "usage_amount", "type": "decimal", "required": False},
        {"source": "invoice.month", "target": "billing_period", "type": "str", "required": True},
        {"source": "resource.name", "target": "resource_id", "type": "str", "required": False},
        {"source": "service.description", "target": "service_name", "type": "str", "required": True},
        {"source": "sku.id", "target": "sku_id", "type": "str", "required": True},
        {"source": "project.id", "target": "account_id", "type": "str", "required": True},
        {"source": None, "target": "charge_type", "type": "str", "required": True, "transform": _gcp_charge_type},
    ],
}


def resolve_row(raw: dict, provider_version: str) -> CanonicalCostRecord:
    """Transform one raw provider row into a CanonicalCostRecord using the registry."""
    if provider_version not in MAPPING_REGISTRY:
        raise KeyError(f"No mapping registered for provider_version={provider_version!r}")

    resolved: dict[str, Any] = {}
    for entry in MAPPING_REGISTRY[provider_version]:
        transform: Optional[Callable[[dict], Any]] = entry.get("transform")
        value = transform(raw) if transform else _get_dotted(raw, entry["source"])

        if value is None and entry["required"]:
            raise ValueError(
                f"Required field '{entry['target']}' resolved to None from "
                f"source={entry['source']!r} (provider_version={provider_version})"
            )

        if entry["type"] == "decimal" and value is not None:
            try:
                value = Decimal(str(value))
            except InvalidOperation as exc:
                raise ValueError(f"Non-numeric value for '{entry['target']}': {value!r}") from exc

        resolved[entry["target"]] = value

    record = CanonicalCostRecord(
        cost_amount=resolved.get("cost_amount", Decimal("0")),
        usage_amount=resolved.get("usage_amount") or Decimal("0"),
        billing_period=resolved["billing_period"],
        resource_id=resolved.get("resource_id"),
        service_name=resolved["service_name"],
        sku_id=resolved["sku_id"],
        account_id=resolved["account_id"],
        charge_type=resolved["charge_type"],
    )
    validate_record(record)
    return record


def validate_record(record: CanonicalCostRecord) -> None:
    """Validate the resolved record against the unified jsonschema before it ships downstream."""
    payload = {**asdict(record), "cost_amount": str(record.cost_amount), "usage_amount": str(record.usage_amount)}
    try:
        validate(instance=payload, schema=CANONICAL_SCHEMA)
    except ValidationError as exc:
        logger.error("Schema validation failed for resolved record: %s", exc.message)
        raise


# Golden-record fixtures: known raw input paired with the exact expected canonical
# output. Run in CI on every registry change to catch silent mapping drift.
GOLDEN_RECORDS = [
    (
        "aws_cur_v1",
        {
            "line_item_unblended_cost": "12.500000",
            "line_item_usage_amount": "100",
            "bill_billing_period_start_date": "2026-07-01T00:00:00Z",
            "line_item_resource_id": "i-0abc123",
            "product_servicename": "Amazon Elastic Compute Cloud",
            "line_item_usage_type": "BoxUsage:m5.large",
            "line_item_usage_account_id": "111122223333",
            "line_item_line_item_type": "Usage",
        },
        CanonicalCostRecord(
            cost_amount=Decimal("12.500000"), usage_amount=Decimal("100"),
            billing_period="2026-07-01T00:00:00Z", resource_id="i-0abc123",
            service_name="Amazon Elastic Compute Cloud", sku_id="BoxUsage:m5.large",
            account_id="111122223333", charge_type="Usage",
        ),
    ),
    (
        "gcp_export_v1",
        {
            "cost": "15.00", "credits": [{"amount": "-2.50"}],
            "usage": {"amount": "50"}, "invoice": {"month": "202607"},
            "resource": {"name": "projects/p/instances/vm-1"},
            "service": {"description": "Compute Engine"},
            "sku": {"id": "CP-COMPUTEENGINE-VMIMAGE-N1-STANDARD-4"},
            "project": {"id": "my-project"}, "cost_type": "regular",
        },
        CanonicalCostRecord(
            cost_amount=Decimal("12.50"), usage_amount=Decimal("50"),
            billing_period="202607", resource_id="projects/p/instances/vm-1",
            service_name="Compute Engine", sku_id="CP-COMPUTEENGINE-VMIMAGE-N1-STANDARD-4",
            account_id="my-project", charge_type="Usage",
        ),
    ),
]


def run_golden_record_tests() -> bool:
    all_passed = True
    for provider_version, raw, expected in GOLDEN_RECORDS:
        try:
            actual = resolve_row(raw, provider_version)
        except Exception as exc:
            logger.error("Golden record for %s raised unexpectedly: %s", provider_version, exc)
            all_passed = False
            continue
        if actual != expected:
            logger.error("Golden record mismatch for %s:\n  expected=%s\n  actual=%s",
                          provider_version, expected, actual)
            all_passed = False
        else:
            logger.info("Golden record OK: %s", provider_version)
    return all_passed


if __name__ == "__main__":
    if not run_golden_record_tests():
        raise SystemExit(1)
    logger.info("All golden-record checks passed.")

Verification & Testing

Correctness here means “the resolver’s output never silently diverges from what the mapping spec claims it will produce.” Three checks enforce that:

  • Golden-record regression. The fixture set above pins one real (anonymized) row per provider to an exact expected CanonicalCostRecord. Run it in CI on every pull request that touches MAPPING_REGISTRY or a transform function — a diff in expected output is a reviewable change, not a production incident.
  • Required-field fuzzing. For each mapping entry marked required: True, construct a raw row with that source field deleted and assert resolve_row raises ValueError rather than returning a record with a masked default. This catches the “returns zero instead of failing” failure mode directly.
  • Cross-provider total reconciliation. For a period with known spend in more than one provider, sum cost_amount per provider after resolution and compare against each provider’s own invoice total (AWS Cost Explorer total, GCP’s BigQuery export SUM(cost) + SUM(credits), Azure’s Cost Management amount). A mismatch beyond rounding tolerance means a mapping entry — most often a currency or credit-summation transform — is wrong.

Common Pitfalls Checklist

  • Treating cost as cost_amount for GCP without summing credits. Net cost is cost + SUM(credits[].amount); skipping the sum overstates spend by the credited amount — use the _sum_gcp_credits transform shown above.
  • Silently defaulting a required field to 0 or "". This hides broken exports as legitimate zero-cost rows — raise ValueError on any required field resolving to None, as resolve_row does.
  • Mapping by column position instead of name. A reordered or schema-revised export shifts every downstream field — always resolve by field name (or dotted path), never ordinal index.
  • Skipping the golden-record test on registry edits. A one-line mapping typo (cost_amount pointed at usage_amount) passes type checks but produces wrong totals — the golden-record diff is the only thing that catches it before production.
  • Assuming charge_type values are comparable strings across providers. AWS’s Usage/Tax/Credit/Refund, GCP’s regular/tax/adjustment, and Azure’s own enum need an explicit transform into one shared vocabulary, not a pass-through.

Frequently Asked Questions

Why not just normalize field names with a rename dict instead of a full registry?

A rename dict works only when every canonical field maps to exactly one raw field with no type or unit conversion. In practice at least one field per provider needs a transform — GCP’s credit summation, Azure’s currency field disambiguation, or an enum translation for charge type — and a registry entry can carry that transform alongside the rename, where a flat dict cannot.

How often do provider billing schemas actually change?

AWS adds CUR columns without a version bump several times a year; GCP and Azure revise export schemas less frequently but without much advance notice. Treat every provider mapping as versioned (aws_cur_v1, aws_cur_v2) rather than a single evolving dict, so an old pipeline run against archived data still resolves correctly against the schema it was written for.

Should the mapping spec live in Python dicts or an external YAML file?

Either works with the same resolver — the example above uses Python dicts for readability, but loading MAPPING_REGISTRY from YAML at startup lets a FinOps engineer add a provider field mapping without a code deploy, at the cost of losing static type checking on the entries themselves. Most teams start with dicts and migrate to YAML once non-engineers need to edit mappings.

What happens to fields in the raw export that have no canonical target?

They are dropped during resolution, which is intentional — the canonical schema defines the contract that allocation, tagging, and reporting downstream depend on. If a provider-specific field is genuinely needed later, add it as an optional canonical field and a new registry entry rather than smuggling it through unmapped.

Up: Normalizing Multi-Cloud Billing Schemas · Home: Cloud Cost Optimization & FinOps Automation