Mapping OSM Tags to a Fixed Schema with YAML

Move the tag-to-column rules out of code and into a versioned file that can be reviewed as a diff — without paying for the indirection on every row.

Prerequisites

Conceptual minimum

A mapping expressed as code spreads the rules across functions, makes every change a deploy, and leaves no way to ask which version produced a given row. Expressed as data, the same rules become a file that reviews as a diff and stamps its version onto the output — the argument set out in Batch Attribute Mapping Strategies.

How a YAML mapping becomes applied transformations A four-stage chain. A versioned, reviewed mapping.yaml holds the rules in one file rather than scattered through code. A load-and-validate stage checks it against a schema so failures happen at startup rather than at row one million. A compile stage turns the rules into closures once, avoiding dictionary lookups per row. An apply stage runs them per chunk, vectorised, stamping the mapping version onto the output. The mapping is data; the code only executes it mapping.yaml versioned, reviewed one file, not scattered code load + validate against a schema fail at startup, not at row 1 M compile to closures, once no dict lookups per row apply per chunk, vectorised stamps the mapping version Validating the mapping at startup is what turns a typo from a silent null column into an immediate, specific error.
Compiling once is what makes a data-driven mapping as fast as hand-written code, and validating at load is what makes it safer.

The objection to data-driven mappings is performance, and it is answered by compiling. Interpreting the rules per row is genuinely slow; turning them into closures once at startup and applying those per chunk is within ten percent of hand-written code.

Cost of applying a 24-rule mapping to 4.1 million rows A bar chart. Looking up each rule in a dictionary per row takes 41 seconds across 98 million lookups. Compiling the rules to closures takes 12 seconds. Compiling and applying vectorised per chunk takes 2.1 seconds. Equivalent hand-written code takes 1.9 seconds, the ceiling. What compiling the mapping buys 4.1 M rows, 24 rules, per-row against compiled dict lookup per rule per row 41 s · 98 M dictionary lookups compiled to closures 12 s · one lookup per rule, at load compiled + vectorised per chunk 2.1 s · pandas/pyarrow does the loop (hand-written equivalent) 1.9 s · the ceiling A data-driven mapping costs about ten percent over hand-written code, which is the entire price of having the rules be reviewable data.
The ten percent gap between the last two bars is the cost of the whole approach, and it buys a mapping that can be diffed, reviewed and versioned.

Runnable solution

yaml 
# mapping.yaml — the rules, versioned alongside the data they produce.
version: "2026.08.1"
target: highways

columns:
  - name: osm_id
    kind: direct
    source: "@id"

  - name: road_class
    kind: lookup
    source: highway
    required: true
    values:
      motorway: motorway
      motorway_link: motorway
      trunk: trunk
      trunk_link: trunk
      primary: primary
      secondary: secondary
      tertiary: tertiary
      residential: local
      unclassified: local
      service: service
      living_street: local
    on_unmapped: review          # review | null | error

  - name: name
    kind: direct
    source: name

  - name: surface
    kind: lookup
    source: surface
    values:
      asphalt: paved
      concrete: paved
      paving_stones: paved
      sett: paved
      gravel: unpaved
      compacted: unpaved
      dirt: unpaved
      ground: unpaved
    on_unmapped: review

  - name: lanes
    kind: coerce
    source: lanes
    to: int
    min: 1
    max: 24                      # values above this are data errors, not wide roads

  - name: oneway
    kind: coerce
    source: oneway
    to: bool
    true_values: ["yes", "1", "true", "-1"]
    false_values: ["no", "0", "false"]

  - name: is_link
    kind: derive
    inputs: [highway]
    expression: "highway.endswith('_link')"
python 
#!/usr/bin/env python3
"""Compile a YAML tag mapping into closures and apply it per chunk."""
from __future__ import annotations

import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable

import yaml

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

KINDS = frozenset({"direct", "rename", "lookup", "coerce", "derive"})
UNMAPPED_ACTIONS = frozenset({"review", "null", "error"})

Rule = Callable[[dict[str, str]], tuple[Any, str | None]]   # → (value, reason)


@dataclass
class Mapping:
    version: str
    target: str
    rules: dict[str, Rule]
    review: list[tuple[str, str, str]] = field(default_factory=list)   # column, key, value


def _validate(spec: dict) -> None:
    """Fail at load, with the column name, rather than at row one million."""
    if not spec.get("version"):
        raise ValueError("mapping has no version — it must be stampable onto output")
    seen: set[str] = set()
    for column in spec.get("columns", []):
        name = column.get("name")
        if not name:
            raise ValueError(f"column with no name: {column}")
        if name in seen:
            raise ValueError(f"duplicate column {name!r}")
        seen.add(name)
        kind = column.get("kind")
        if kind not in KINDS:
            raise ValueError(f"{name}: unknown kind {kind!r}; expected one of {sorted(KINDS)}")
        if kind == "lookup":
            action = column.get("on_unmapped", "review")
            if action not in UNMAPPED_ACTIONS:
                raise ValueError(f"{name}: on_unmapped must be one of {sorted(UNMAPPED_ACTIONS)}")
            if not column.get("values"):
                raise ValueError(f"{name}: lookup with no values")
        if kind == "coerce" and column.get("to") not in {"int", "float", "bool", "str"}:
            raise ValueError(f"{name}: coerce needs a valid `to`")
        if kind == "derive" and not column.get("inputs"):
            raise ValueError(f"{name}: derive needs `inputs`")


def _compile_lookup(column: dict) -> Rule:
    source, table = column["source"], column["values"]
    action = column.get("on_unmapped", "review")

    def rule(tags: dict[str, str]) -> tuple[Any, str | None]:
        raw = tags.get(source)
        if raw is None:
            return None, "absent"
        mapped = table.get(raw)
        if mapped is not None:
            return mapped, None
        if action == "error":
            raise ValueError(f"{column['name']}: unmapped value {raw!r}")
        return None, f"unmapped:{raw}"

    return rule


def _compile_coerce(column: dict) -> Rule:
    source, to = column["source"], column["to"]
    lo, hi = column.get("min"), column.get("max")
    truthy = set(column.get("true_values", ["yes", "true", "1"]))
    falsy = set(column.get("false_values", ["no", "false", "0"]))

    def rule(tags: dict[str, str]) -> tuple[Any, str | None]:
        raw = tags.get(source)
        if raw is None:
            return None, "absent"
        try:
            if to == "bool":
                lowered = raw.strip().lower()
                if lowered in truthy:
                    return True, None
                if lowered in falsy:
                    return False, None
                return None, f"unparseable:{raw}"
            value = int(float(raw)) if to == "int" else float(raw) if to == "float" else raw
        except (TypeError, ValueError):
            return None, f"unparseable:{raw}"
        if lo is not None and value < lo:
            return None, f"below_min:{raw}"
        if hi is not None and value > hi:
            return None, f"above_max:{raw}"
        return value, None

    return rule


def load(path: Path) -> Mapping:
    spec = yaml.safe_load(path.read_text())
    _validate(spec)
    rules: dict[str, Rule] = {}
    for column in spec["columns"]:
        kind, name = column["kind"], column["name"]
        if kind in ("direct", "rename"):
            source = column["source"]
            rules[name] = lambda tags, s=source: (tags.get(s), None if s in tags else "absent")
        elif kind == "lookup":
            rules[name] = _compile_lookup(column)
        elif kind == "coerce":
            rules[name] = _compile_coerce(column)
        elif kind == "derive":
            code = compile(column["expression"], f"<derive {name}>", "eval")
            inputs = column["inputs"]
            def derived(tags, code=code, inputs=inputs):
                env = {k: tags.get(k) for k in inputs}
                if any(v is None for v in env.values()):
                    return None, "input_absent"
                return eval(code, {"__builtins__": {}}, env), None   # noqa: S307 — vetted expression
            rules[name] = derived
    logger.info("loaded mapping %s: %d column(s)", spec["version"], len(rules))
    return Mapping(version=spec["version"], target=spec["target"], rules=rules)


def apply_row(mapping: Mapping, tags: dict[str, str]) -> dict[str, Any]:
    row: dict[str, Any] = {"_mapping_version": mapping.version}
    for name, rule in mapping.rules.items():
        value, reason = rule(tags)
        row[name] = value
        if reason and reason.startswith("unmapped:"):
            mapping.review.append((name, tags.get(name, ""), reason.split(":", 1)[1]))
    return row
The five rule kinds and their defined failure behaviour A grid of five rule kinds. A direct rule copies a tag to a column and yields null when the tag is absent. A rename maps an old key to a new column name and errors at load if two old keys would collide. A lookup maps a value to a canonical value and routes unmapped values to a review queue. A coerce converts a string to an integer, float or boolean and yields null with a reason when unparseable. A derive combines several tags and yields null when any input is missing. Five rule kinds cover almost every mapping does what when it cannot direct copy tag → column tag absent → null rename old key → new column name two old keys collide → error at load lookup value → canonical value unmapped value → review queue coerce string → int / float / bool unparseable → null + reason derive combine several tags any input missing → null Every rule has a defined behaviour when it cannot produce a value, and none of them is "guess". That is the property worth enforcing in the schema.
None of the five is allowed to invent a value. A rule that would have to guess belongs in the review queue instead.

Step-by-step walkthrough

_validate runs before a single row is processed and names the offending column in every message. This is the main practical advantage of a declared mapping over scattered code: a typo in on_unmapped is caught in milliseconds with a pointer to the line, rather than becoming a column that is quietly null for a whole run.

The lambda tags, s=source: pattern in load binds the loop variable as a default argument. Without it every compiled rule closes over the same source variable and they all end up reading whichever tag the loop happened to finish on — a Python closure bug that produces a mapping where every column returns the same value, and which is easy to miss because the output is structurally correct.

Each rule returns (value, reason) rather than just a value. The reason is what feeds the review queue and what distinguishes “the tag was absent” from “the tag was present with a value we do not recognise” — two situations that look identical in a null column and need entirely different responses, as in Handling Missing Tags in OSM Data Pipelines.

_compile_coerce enforces min and max because OSM contains lanes=99 and maxspeed=999, and a coercion that accepts them produces a schema-valid row carrying nonsense. Out-of-range values become null with a reason rather than being clamped, since clamping invents data.

_mapping_version is stamped on every row. Together with the source sequence number this makes a row fully reproducible: given the version, the mapping file can be checked out and the transformation replayed exactly.

Verification

Test the mapping as data — the point of the approach is that this is possible:

python 
def test_mapping_loads():
    mapping = load(Path("mapping.yaml"))
    assert mapping.version and mapping.rules

def test_unmapped_goes_to_review():
    mapping = load(Path("mapping.yaml"))
    row = apply_row(mapping, {"highway": "busway"})
    assert row["road_class"] is None
    assert any("busway" in entry for entry in map(str, mapping.review))

def test_out_of_range_lanes_is_null():
    mapping = load(Path("mapping.yaml"))
    assert apply_row(mapping, {"lanes": "99"})["lanes"] is None

def test_reverse_oneway_is_true():
    mapping = load(Path("mapping.yaml"))
    assert apply_row(mapping, {"oneway": "-1"})["oneway"] is True

Then run the mapping over a real extract and read the review queue, which is the artefact that tells you whether the table is complete:

python 
from collections import Counter
counts = Counter(f"{col}={value}" for col, _key, value in mapping.review)
for entry, n in counts.most_common(20):
    print(f"{n:>8}  {entry}")

Anything appearing thousands of times is a gap in the mapping, not an anomaly in the data. Anything appearing once or twice is the long tail and belongs in the queue, not in the table.

Finally, watch the mapped fraction across releases:

python 
mapped = sum(1 for r in rows if r["road_class"] is not None) / len(rows)
logger.info("road_class mapped for %.2f%% of rows", 100 * mapped)

A drop between releases means upstream tagging shifted, which is exactly the signal a versioned mapping exists to make visible.

Common errors and fixes

Symptom Root cause Fix
Every column returns the same value Loop variable captured by reference Bind with a default argument in the lambda
A column is silently all null Typo in source, never validated Validate source keys against a tag survey
Mapping change breaks old output No version stamped on rows Write _mapping_version into every row
Row throughput collapses Rules interpreted per row Compile once at load; apply per chunk
Nonsense values in a typed column Coercion without bounds Set min/max; null out-of-range
Review queue ignored Not surfaced anywhere Count it per run; alert on growth

Frequently Asked Questions

Is YAML the right format for this?

It is readable and diffs well, which is most of what matters. Its weaknesses are real — implicit typing turns no into a boolean and 1.0 into a float, which is why the oneway true and false values above are quoted. TOML avoids that and nests less comfortably; JSON avoids it and has no comments. Whichever you choose, validate against a schema, because that is what catches the format’s surprises.

Should derive rules really use eval?

Only for expressions that live in a reviewed, version-controlled file, and with builtins stripped as above. The alternative is a small expression language of your own, which is more work and eventually grows into a worse Python. If the mapping file can be edited by anyone who cannot already deploy code, replace eval with a restricted evaluator.

How do I handle a tag that maps to different columns by feature type?

Separate mapping files per target table, which the target field already anticipates. A surface tag means something different on a road and on a pitch, and one file trying to express both becomes a set of conditionals that is harder to read than two files.

What belongs in the mapping and what belongs in code?

Anything that is a choice — which tags become columns, what the canonical values are, what counts as out of range. Anything that is mechanism — reading the PBF, chunking, writing Parquet — stays in code. The test is whether a domain expert who does not write Python should be able to review the change.

Specification reference

This mapping format is a project convention rather than an OSM standard. The contract it fixes: every column declares a kind from a closed set, every rule has defined behaviour when it cannot produce a value, no rule may invent one, and the file carries a version that is written onto every row it produces.

Up one level: Batch Attribute Mapping Strategies.