Ingestion Design

The ingestion layer extracts raw data from OEM-specific external APIs, validates it, and progressively refines it into consolidated entity records partitioned by day. Each OEM runs its own independent Dagster deployment; shared infrastructure lives in a common library so OEM pipelines don't share state or affect each other.

Pipeline Structure

Every ingestion pipeline follows the same three-stage structure:

Stage	Responsibility
Raw	Fetch data from the source API and store the response exactly as returned
Transformed	Extract entity records from the raw response, deduplicate by key group, and validate
Consolidated	Merge transformed records across all sources for an entity type

Each stage has asset checks that run automatically after materialization.

Dependency Ordering

The asset DAG has two independent ordering dimensions that do not form cycles:

• Source dependencies operate in the raw and transformed tiers. One raw source may depend on another (e.g. fetching search results requires dealer IDs from a prior fetch). These dependencies are fully resolved by the end of the transformed tier.

• Entity dependencies begin at the consolidated tier, where foreign_keys allow one entity's consolidation to reference another consolidated entity's same-partition data. For example, consolidated inventory can reference consolidated models to attach a resolved model ID to each row. Since all source-level dependencies are already resolved, entity ordering is independent of source ordering.

This separation guarantees no circular dependencies: sources never depend on entities, and entities never depend on sources.

Asset Key Conventions

Dagster asset keys form a path hierarchy using / as the separator:

Stage	Pattern	Example
Source	<oem>/<market>/sources/<source>_<resource>	audi/us/sources/scs_search
Raw	<oem>/<market>/raw/<source>_<resource>	audi/us/raw/scs_search
Transformed	<oem>/<market>/transformed/<name>_<entity>	audi/us/transformed/onegraph_inventory
Consolidated	<oem>/<market>/consolidated/<entity>	audi/us/consolidated/inventory

Asset Checks

Stage	Check name	What it verifies	Blocking
Raw	fetch_attempts	All HTTP requests made by the fetcher returned 2xx; per-status failure patterns surfaced in metadata	No
Transformed	discrepancies	No conflicting rows for the same entity ID; same-source field disagreements grouped into discrepancy patterns	No
Consolidated	integrity	Every row has a non-null entity ID; records row count	No
Consolidated	unresolved_refs	Cross-entity reference links resolved successfully	No
Transformed (highest-priority source)	cross_source_discrepancies	Cross-source merges produce no conflicting field values for the same entity; field-level disagreements grouped into discrepancy patterns	No
Transformed (paired inventory sources)	cross_source_overlap	Membership overlap between two inventory feeds stays within an expected range; per-side coverage and a breakdown of keys missing from the other feed surfaced in metadata	Configurable

Most checks are non-blocking (severity WARN) — they surface issues in the Dagster UI without aborting the run. The fetch_attempts check additionally reports tracking=instrumented or tracking=uninstrumented in its metadata so operators can tell whether the source module is recording per-request outcomes (see Component Reference).

The one exception is cross_source_overlap: when its blocking flag is set it emits at severity ERROR and does short-circuit the run. See ConsolidatedComponent → Cross-source overlap for configuration and the run-topology caveat.

Failing checks are triaged per the rules in Verification → Alert triage — potential raw-collection issues take priority over everything else, because raw gaps are permanent.

Asset Groups and Tags

Source observability assets are grouped sources. Raw fetch assets are grouped raw. Transformed assets are grouped transformed. Consolidated assets are grouped consolidated. Groups are visible in the Dagster UI's asset catalog and can be used to filter and select assets in bulk.

Every ingestion asset also carries an oem tag for per-OEM filtering:

oem: audi

Partitioning

All ingestion assets use a DailyPartitionsDefinition with end_offset=1. The end_offset=1 means today's date is always a valid partition, so assets can be materialized on the day they are due without waiting for the day to roll over.

Request Throttling

Fan-out HTTP fetches are paced by ai_core.http.Throttle, a pool-wide request pacer shared across every worker thread and retry in a fan-out. It enforces steady-state spacing (min_interval seconds between requests across the pool) and a shared 429 penalty window (penalize() pauses the whole pool, not just the worker that hit the limit).

By default the throttle keeps its state in-process, so it only coordinates threads within a single Dagster run. When the same origin (or a shared API budget) is hit by multiple concurrent runs — backfills, overlapping partitions, or several sources behind one rate limit — a per-run throttle cannot see the aggregate load. Throttle therefore accepts a pluggable backend: the default LocalBackend (in-process) or a DynamoThrottleBackend that stores the pacing state in a shared DynamoDB item, so every process sharing a key forms one pool.

Distributed throttle

Construct a cross-process throttle with ai_core.http.distributed_throttle(key, min_interval, *, local_min_interval=None, meter=None). The key is a stable string identifying the shared budget:

• Per-credential limits → one key per credential. Zyte's rate limit is per API key, so zyte_throttle(prefix) builds the key from a prefix (e.g. "BMW_STOLO" → "zyte/bmw_stolo"). ZyteSession/ZyteSessionFactory take that same key_prefix and use it for both the throttle and the <PREFIX>_ZYTE_API_KEY credential lookup, so a call site's pacing and credential stay in lockstep. Call sites sharing a prefix share one pool; distinct keys are paced independently. Only the prefix is used — the secret value never reaches the throttle store.
• Per-origin limits where only concurrent runs of the same source contend → "<oem>/<source>", e.g. "gm/inventory".

Coordination is a single conditional UpdateItem (compare-and-set) per granted slot — no lock is held, so a crashed process cannot wedge the pool. Timestamps are wall-clock epoch seconds; NTP skew between tasks is far below any min_interval in use and the CAS is on stored state, so skew can only blur spacing, never corrupt it. Items carry an expires_at TTL so idle keys clean themselves up.

Enabling it. The backend is only used when DISTRIBUTED_THROTTLE_TABLE is set (to the DynamoDB table name, distributed-throttle in prod); unset, distributed_throttle() returns a plain in-process Throttle at local_min_interval. This env var is the per-project kill-switch — local dev, unit tests, and un-opted-in deployments are safe by default. DISTRIBUTED_THROTTLE_REGION (optional — unset lets boto3 resolve the region normally, e.g. from AWS_REGION or task metadata) and DYNAMODB_LOCAL_ENDPOINT (local override) are also honored. The run task's IAM role needs dynamodb:UpdateItem and dynamodb:GetItem on the table.

Contention vs. outage. Every grant and every failed compare-and-set is a write to one hot partition key, so a large fan-out can make DynamoDB throttle the writes. That is contention, not an outage — the backend treats a throttling response as a back-off-and-retry signal (exponential backoff keyed on DynamoDB's own throttling) so the write storm drains itself, rather than counting it toward the outage fallback.

Fail-open. DynamoDB availability is not a correctness dependency: on genuine connectivity errors (three consecutive non-throttling, non-conditional failures) the backend logs once and paces in-process at max(min_interval, local_min_interval) for a bounded window, then re-probes DynamoDB. A DynamoDB outage costs cross-run coordination for that window but never a run — the same behavior as before the distributed backend existed.

Metering. An injected RequestMeter records this process's share of the offered load, not the pool-wide aggregate; read each run's meter summary to reconstruct per-process contribution.