Bill of Lading Schema Mapping #

Bill of Lading Schema Mapping is the deterministic translation layer that normalizes heterogeneous B/L payloads — UN/EDIFACT, ANSI X12, and carrier JSON/XML — into a single version-controlled record that downstream port automation can query without ambiguity. Shipping operators, port authorities, and terminal systems cannot afford silent field degradation: a mistyped weight, a dropped consignee tax ID, or a misresolved UN/LOCODE cascades into customs holds, demurrage disputes, and stowage errors. This mapping discipline anchors into the Core Maritime Architecture & Taxonomy framework, enforcing strict naming conventions, semantic relationships, and lifecycle state tracking so every normalized field carries a declared type, a source lineage, and a validation verdict.

Ingestion Boundary & Protocol Handling #

Production ingestion begins at the protocol edge. Raw payloads arrive over AS2, SFTP, or HTTPS endpoints, often with inconsistent character encodings, truncated segments, or carrier-specific escape sequences. A resilient parser must strip transport wrappers, normalize line endings, and split payloads on deterministic delimiters before any business logic executes. This is the same syntax-tolerant, semantics-strict posture used across Document Ingestion & EDI Parsing Workflows: accept messy-but-parseable input at the wire, then apply uncompromising checks downstream.

Each transport family has its own boundary rules. For EDIFACT, segment boundaries (UNB, BGM, NAD, LOC, GID, MEA) require positional parsing driven by the service string in the optional UNA segment. ANSI X12 relies on ISA/GS/ST envelopes with element and segment separators declared in the ISA header itself (* and ~ are conventional, never assumed). Carrier APIs deliver nested JSON/XML with inconsistent casing and optional fields. The ingestion layer must therefore:

1. Detect format via magic bytes or envelope headers (UNB/UNA → EDIFACT, ISA → X12, {/< → API).
2. Read delimiters from the payload rather than hard-coding them — carrier delimiter drift is the single most common parse failure.
3. Apply codec fallbacks in order (utf-8 → iso-8859-1 → windows-1252) and record which codec succeeded.
4. Emit structured JSON logs with correlation_id, source_system, and raw_payload_hash before transformation begins.

The canonical inbound segments this layer maps are stable across most carriers; the table below is the minimum contract the parser resolves before typing:

If the parser encounters unrecoverable structural corruption — a broken envelope, an unreadable encoding, a missing mandatory service string — the payload routes immediately to a dead-letter queue (DLQ) with a PARSE_FAILURE status. No partial records proceed downstream.

Python Data Structure Mapping #

Once parsed, maritime fields map to explicit Python data structures. Implicit dictionaries are unacceptable in production; use pydantic.BaseModel (or typing.TypedDict for hot paths) with strict type annotations so malformed payloads are rejected at construction. The mapping layer enforces deterministic coercion, unit normalization, and null handling before a record is ever persisted.

from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional

import structlog
from pydantic import BaseModel, Field, field_validator

log = structlog.get_logger()


class BillOfLading(BaseModel):
    bl_number: str = Field(..., min_length=1)
    vessel_imo: int = Field(..., ge=1_000_000, le=9_999_999)
    consignee_id: str
    pol: str = Field(..., pattern=r"^[A-Z]{5}$")   # UN/LOCODE
    pod: str = Field(..., pattern=r"^[A-Z]{5}$")
    gross_weight_kg: Decimal = Field(..., ge=0)
    laden_at: datetime
    data_quality: str = "AUTHORITATIVE"

    @field_validator("bl_number")
    @classmethod
    def upper(cls, v: str) -> str:
        return v.strip().upper()

    @field_validator("laden_at")
    @classmethod
    def as_utc(cls, v: datetime) -> datetime:
        return v.astimezone(timezone.utc)

Field resolution follows a strict precedence chain: primary carrier field → fallback alias → computed default → explicit None. Standard coercion patterns include:

• Dates/times: EDIFACT DTM 2403151030 (CCYYMMDDHHMM, format 203) → datetime(2024, 3, 15, 10, 30, tzinfo=UTC). Reject ambiguous two-digit years.
• Weights/measures: MEA+WT+G+15000:KGM → Decimal("15000") kg, then normalise to metric tonnes only at the presentation boundary — never lose precision mid-pipeline.
• Locations: LOC+9+USNYC → {"loc_function": "PORT_OF_LOADING", "un_locode": "USNYC", "verified": True}.
• Container references: EQD+CN+MSKU1234567+45G1 → {"iso_code": "MSKU1234567", "size_type": "45G1", "teu": 2.0}.

The regex extraction, positional segment indexing, and memory-safe serialization behind these rules are documented in full in How to map UN/EDIFACT B/L fields to Python dicts, which covers high-throughput streaming without materialising whole payloads in memory. The parallel task of turning a forwarding instruction into this same typed shape — resolving its transport legs, parties, and equipment — is walked through in Mapping IFTMIN transport instructions to typed models.

Validation, Quarantine & Compliance Auditing #

Maritime documentation is inherently inconsistent. Missing consignee tax IDs, truncated cargo descriptions, and malformed container seals routinely trigger downstream failures. Production systems implement a multi-tier validation boundary — structural, then semantic, then regulatory — before any record enters the operational data lake.

1. Structural validation: Apply Pydantic strict mode (or JSON Schema) at the ingestion boundary. Reject records missing mandatory fields (bl_number, vessel_imo, gross_weight_kg, consignee_id). These are format defects, so they fail fast to the DLQ.
2. Semantic validation: Cross-reference against authoritative registries. Validate container codes against the ISO 6346 check-digit algorithm, verify UN/LOCODEs against the official port registry, and enforce TEU/weight limits per vessel class and SOLAS Verified Gross Mass (VGM) thresholds.
3. Quarantine routing: Records that parse cleanly but fail a semantic or regulatory check route to a QUARANTINE topic, not the DLQ. This preserves pipeline throughput while shipping-ops teams triage exceptions a human can plausibly fix — an unknown LOCODE, a missing VGM — without halting the line.
4. Immutable audit trail: Every validation event logs record_id, rule_applied, original_value, transformed_value, and compliance_status. This chain satisfies customs audit requirements and supports rapid root-cause analysis during port state control inspections.

The distinction between the DLQ and the quarantine topic is load-bearing: keeping unrecoverable corruption separate from recoverable business exceptions is what lets the quarantine path reconcile automatically once an authoritative source recovers.

Downstream Integration #

Mapped B/L data does not exist in isolation. It drives terminal operating system (TOS) updates, customs declarations, stowage planning, and equipment interchange receipts. The schema explicitly tags regulatory attributes and operational state flags so downstream consumers never re-infer meaning from raw text.

Container relationships require strict parent-child resolution. A single B/L may reference multiple containers, each with distinct seals, cargo descriptions, and hazardous-material codes. Resolving these into the Container Hierarchy Data Models — the Vessel → Bay → Tier → Row → Container topology — ensures that equipment tracking, reefer monitoring, and dangerous-goods segregation rules execute deterministically off a shared structure rather than a flattened copy.

Port operations consume B/L status transitions to trigger milestone events. When a mapped record reaches GATE_IN or LOADED_ONBOARD, it must propagate to the Port Call Workflow Design state machine. Idempotent event publishing — keyed on a deterministic (bl_number, milestone, event_hash) tuple — guarantees that duplicate AS2 retransmissions or API retries do not corrupt stowage plans or customs manifests. Where the same cargo is also described by an aggregated manifest, the mapped record is reconciled against IFCSUM EDI Message Parsing output, and its physical movements are correlated with the real-time position feed from AIS Data Stream Integration. Every cross-boundary publish is authenticated under the Maritime Security Boundary Setup zero-trust controls.

Fallback Chains & Uptime Guarantees #

Uptime in maritime automation depends on graceful degradation, not brittle failure. External enrichment services — customs APIs, port authority gateways, carrier tracking endpoints — experience latency spikes, rate limits, and scheduled maintenance. Production pipelines implement explicit fallback chains:

1. Circuit breakers: Track failure rates per downstream endpoint. Open the circuit after five consecutive 5xx or timeout errors and route requests to cached schema defaults or local registry lookups instead of hammering a degraded service.
2. Exponential backoff + jitter: Retry transient 429/503 responses using base_delay * 2 ** attempt + random_jitter. Cap at three retries before routing the record to the quarantine topic.
3. Schema versioning: Keep every carrier schema version deployed simultaneously and select the parser at runtime from the directory version carried in the UNB and UNH segments (for example D95B versus D16A). Deprecate an old schema only after telemetry shows zero inbound traffic against it.
4. Structured observability: Emit OpenTelemetry-compliant logs with trace_id, span_id, and service_name. Track pipeline latency percentiles (p50, p95, p99), validation-failure rates, and DLQ depth, and alert on sustained error thresholds.

Fallback logic must never compromise data integrity. If a fallback enriches a record with estimated values, tag it explicitly (data_quality="ESTIMATED") and trigger a reconciliation job once the authoritative source recovers — so port authorities and ops teams always operate on auditable, traceable data while the pipeline keeps moving.

Step-by-step Implementation Guide #

The reference pipeline below maps a raw EDIFACT B/L payload to a validated, downstream-ready record. Each step is runnable in isolation and uses type annotations with structlog for structured JSON logging.

Step 1 — Detect format and read delimiters from the payload #

Never assume delimiters. Read the EDIFACT service string from UNA, or the element/segment separators from the X12 ISA header, before splitting.

import structlog

log = structlog.get_logger()


def detect_delimiters(raw: str) -> dict[str, str]:
    if raw.startswith("UNA"):
        # UNA<comp><data><decimal><release><space><segment>
        una = raw[3:9]
        d = {"component": una[0], "data": una[1], "release": una[3], "segment": una[5]}
    elif raw.startswith("ISA"):
        d = {"data": raw[3], "component": raw[104], "segment": raw[105]}
    else:
        raise ValueError("unrecognised envelope")
    log.info("delimiters_detected", **d)
    return d

Step 2 — Split into segments and index by tag #

def to_segments(raw: str, sep: str) -> list[list[str]]:
    return [seg.split("+") for seg in raw.split(sep) if seg.strip()]


def index_by_tag(segments: list[list[str]]) -> dict[str, list[list[str]]]:
    idx: dict[str, list[list[str]]] = {}
    for seg in segments:
        idx.setdefault(seg[0], []).append(seg)
    return idx

Step 3 — Coerce raw segment values into typed fields #

from datetime import datetime, timezone
from decimal import Decimal


def coerce_weight(mea: list[str]) -> Decimal:
    # MEA+WT+G+15000:KGM  -> Decimal("15000")
    value, _, _unit = mea[3].partition(":")
    return Decimal(value)


def coerce_dtm(dtm_value: str) -> datetime:
    # 2403151030 with format qualifier 203 (CCYYMMDDHHMM)
    return datetime.strptime(dtm_value, "%y%m%d%H%M").replace(tzinfo=timezone.utc)

Step 4 — Construct and structurally validate the model #

def build_bl(idx: dict[str, list[list[str]]]) -> BillOfLading:
    bl = BillOfLading(
        bl_number=idx["BGM"][0][2],
        vessel_imo=int(idx["TDT"][0][8]),
        consignee_id=idx["NAD"][0][2],
        pol=idx["LOC"][0][2],
        pod=idx["LOC"][1][2],
        gross_weight_kg=coerce_weight(idx["MEA"][0]),
        laden_at=coerce_dtm(idx["DTM"][0][1].split(":")[1]),
    )
    log.info("bl_structurally_valid", bl_number=bl.bl_number)
    return bl

Step 5 — Apply semantic checks, then route #

def route(bl: BillOfLading, *, locode_ok: bool, vgm_ok: bool) -> str:
    if not locode_ok:
        log.warning("quarantine", bl_number=bl.bl_number, reason="LOCODE_UNKNOWN")
        return "QUARANTINE"
    if not vgm_ok:
        log.warning("quarantine", bl_number=bl.bl_number, reason="VGM_MISSING")
        return "QUARANTINE"
    log.info("bl_accepted", bl_number=bl.bl_number)
    return "ACCEPTED"

A pydantic.ValidationError raised in Step 4 is a structural defect and belongs in the DLQ; a False result in Step 5 is a recoverable business exception and belongs in quarantine.

Troubleshooting Common Failures #

Frequently Asked Questions #

Related #

• Mapping IFTMIN transport instructions to typed models — resolving a forwarding instruction’s legs, parties, and equipment into the same typed record
• How to map UN/EDIFACT B/L fields to Python dicts — the segment-level parsing detail beneath this mapping layer
• Container Hierarchy Data Models — resolving multi-container B/L references into the equipment topology
• Port Call Workflow Design — the idempotent state machine that consumes B/L milestones
• IFCSUM EDI Message Parsing — aggregated manifest data reconciled against mapped B/L records
• Maritime Security Boundary Setup — zero-trust authentication for every cross-boundary publish

Up: Core Maritime Architecture & Taxonomy — the parent framework governing schema, state, and boundary controls.