Building a Freight Data Platform
with Databricks, dbt & Airflow

Reason: A single project config makes model placement and build behavior inspectable.

clean-targets: # directories to be removed by `dbt clean`
  - "target"
  - "dbt_packages"

# Configuring models
# Full documentation: https://docs.getdbt.com/docs/configuring-models

# In this example config, we tell dbt to build all models in the example/
# directory as views. These settings can be overridden in the individual model
# files using the `{{ config(...) }}` macro.
models:
  freight:
    +persist_docs:
      # Push model description as a database table comment
      relation: true
      # Push column descriptions as database column comments
      columns: true
    staging:
      +schema: staging
      +materialized: view
    observability:
      +schema: observability
    intermediate:
      +schema: intermediate
      +group: core
      +access: protected
    marts:
      +schema: marts
      +group: core

Open complete file

Reason: source() references a declared external relation.

version: 2

sources:
  - name: bronze
    database: "{{ target.catalog }}"
    schema: bronze
    tables:
      - name: customers
        config:
          loaded_at_field: _ingested_at
      - name: trucks
        config:
          loaded_at_field: _ingested_at
      - name: trailers
        config:
          loaded_at_field: _ingested_at
      - name: facilities
        config:
          loaded_at_field: _ingested_at
      - name: routes
        config:
          loaded_at_field: _ingested_at
      - name: drivers
        config:
          loaded_at_field: _ingested_at
      - name: loads
        config:

Open complete file

Reason: dbt resolves dependencies and model locations through ref().

{{ config(materialized="table", tags=["intermediate"], event_time="actual_datetime") }}

select
    event_id,
    load_id,
    trip_id,
    event_type,
    facility_id,
    scheduled_datetime,
    actual_datetime,
    detention_minutes,
    on_time_flag as on_time_flag_raw,
    {{ recompute_on_time("actual_datetime", "scheduled_datetime") }}
        as is_on_time_computed,
    on_time_flag
    <> {{ recompute_on_time("actual_datetime", "scheduled_datetime") }}
        as flag_mismatch,
    location_city,
    location_state
from {{ ref("stg_freight__delivery_events") }}

Open complete file

Reason: Selection changes compute timing, storage and update behavior.

{{
    config(
        materialized="incremental",
        incremental_strategy="microbatch",
        event_time="actual_datetime",
        batch_size="month",
        begin="2022-01-01",
        on_schema_change="append_new_columns",
        tags=["marts"],
    )
}}

select

Open complete file

Reason: Reused SQL expressions can be tested without copying logic to every model.

{% macro recompute_on_time(actual_col, scheduled_col) %}
    case
        when {{ actual_col }} is null or {{ scheduled_col }} is null
        then null
        when {{ actual_col }} <= {{ scheduled_col }}
        then true
        else false
    end
{% endmacro %}

Open complete file

Reason: Generic tests parameterize checks across model columns.

{% test not_orphaned(model, column_name, to, field, threshold_pct=0) %}
    with
        orphans as (
            select {{ column_name }} as key_value
            from {{ model }}
            where
                {{ column_name }} is not null
                and {{ column_name }} not in (select {{ field }} from {{ to }})
        ),
        counts as (
            select
                (select count(*) from orphans) as orphan_count,
                (select count(*) from {{ model }}) as total_count
        )
    select o.*
    from orphans o
    cross join counts c
    where c.orphan_count > c.total_count * {{ threshold_pct }} / 100.0
{% endtest %}

Open complete file

Reason: Keeps checks for specific SQL behavior in version control.

-- depends_on: {{ ref('qtn_flag_mismatch') }}
select c.event_id
from {{ ref("int_delivery_events_cleaned") }} c
left join {{ ref("qtn_flag_mismatch") }} q on c.event_id = q.event_id
where c.flag_mismatch = true and q.event_id is null

Open complete file

Reason: Tests a model transformation independently of the complete dataset.

unit_tests:
  - name: test_recompute_on_time_logic
    model: int_delivery_events_cleaned
    given:
      - input: ref('stg_freight__delivery_events')
        rows:
          - {
              event_id: E1,
              load_id: L1,
              trip_id: T1,
              event_type: Pickup,
              facility_id: F1,
              scheduled_datetime: "2024-01-01 10:00:00",
              actual_datetime: "2024-01-01 09:00:00",
              detention_minutes: 0,
              on_time_flag: false,
              location_city: Dallas,
              location_state: TX,
            }
          - {
              event_id: E2,
              load_id: L2,
              trip_id: T2,
              event_type: Delivery,
              facility_id: F2,

Open complete file

Reason: Treats rule definitions as versioned project inputs.

rule_id,model_name,column_name,rule_type,rule_expression,threshold_pct,severity,owner,description
DQ001,stg_freight__loads,load_id,not_null,,0,error,data-team,Primary key must not be null
DQ002,stg_freight__trips,driver_id,not_orphaned,stg_freight__drivers.driver_id,2,warn,data-team,Trip references a driver_id not present in drivers
DQ003,stg_freight__trips,truck_id,not_orphaned,stg_freight__trucks.truck_id,2,warn,data-team,Trip references a truck_id not present in trucks
DQ004,stg_freight__fuel_purchases,gallons,in_range,0:300,0.1,warn,finance,Fuel volume outside plausible range

Open complete file

Reason: Snapshot history is separate from current-state driver dimension.

{% snapshot snap_drivers %}
    {{
        config(
            target_schema="marts",
            unique_key="driver_id",
            strategy="check",
            check_cols=["employment_status", "termination_date", "home_terminal"],
            hard_deletes="new_record",
        )
    }}
    select *
    from {{ ref("stg_freight__drivers") }}
{% endsnapshot %}

Open complete file

Reason: Profile and rule results support inspection; not automatically a Gold filter.

-- depends_on: {{ ref('stg_freight__loads') }}
-- depends_on: {{ ref('stg_freight__trips') }}
-- depends_on: {{ ref('stg_freight__drivers') }}
-- depends_on: {{ ref('stg_freight__trucks') }}
-- depends_on: {{ ref('stg_freight__fuel_purchases') }}
-- depends_on: {{ ref('stg_freight__delivery_events') }}
{{ config(materialized="incremental", tags=["dq"]) }}

with rule_checks as ({{ build_dq_sql() }})

select
    *,
    round(100.0 * failed_rows / nullif(total_rows, 0), 2) as failed_pct,
    (100.0 * failed_rows / nullif(total_rows, 0)) <= threshold_pct as is_passed
from rule_checks

Open complete file

Reason: A hook after execution cannot retroactively prevent earlier table writes.

{% macro dq_severity_gate() %}
    -- depends_on: {{ ref('dq_rule_results') }}
    {% if execute %}
        {# Check if dq_rule_results exists before querying to prevent error on initial tasks #}
        {%- set relation = adapter.get_relation(
            database=(
                target.catalog
                if target.type == "databricks"
                else target.database
            ),
            schema=generate_schema_name("observability"),
            identifier="dq_rule_results",
        ) -%}

        {% if relation is not none %}
            {%- set check_query -%}
      select count(*) as n
      from {{ ref('dq_rule_results') }}
      where severity = 'error' and not is_passed
        and dbt_invocation_id = '{{ invocation_id }}'
            {%- endset -%}

            {%- set results = run_query(check_query) -%}
            {%- set failed_count = results.columns[0].values()[0] if results else 0 -%}

Open complete file

Reason: Schema/type contracts and row-level tests solve different problems.

version: 2

models:
  - name: fct_shipment_lifecycle
    config:
      contract:
        enforced: true
    columns:
      - name: load_id
        data_type: string
        constraints:
          - type: not_null
          - type: primary_key
      - name: booked_at
        data_type: timestamp
      - name: dispatched_at
        data_type: timestamp
      - name: picked_up_at
        data_type: timestamp
      - name: delivered_at
        data_type: timestamp
      - name: hrs_book_to_dispatch
        data_type: double
      - name: hrs_dispatch_to_pickup

Open complete file

Reason: Exposure metadata represents a declared relationship, not live proof of integration.

version: 2

exposures:
  - name: freight_sla_dashboard
    type: dashboard
    maturity: high
    url: "{{ var('sla_dashboard_url', 'https://replace-with-real-url') }}"
    description: >
      SLA dashboard used by the operations team — on-time rate by route/customer/driver,
      cost-per-mile, data quality scorecard.
    depends_on:
      - ref('fct_shipment_lifecycle')
      - ref('fct_load')
      - ref('dim_customer')
      - ref('dim_route')
    owner:
      name: Huu Toan
      email: toandata.work@gmail.com

Open complete file

Reason: Changes the physical destination of dbt models.

{% macro generate_schema_name(custom_schema_name, node) -%}
    {%- set default_schema = target.schema -%}
    {%- if custom_schema_name is none -%}
        {{ default_schema }}
    {%- else -%}
        {{ custom_schema_name | trim }}
    {%- endif -%}
{%- endmacro %}

Open complete file

Reason: dbt model access does not establish cloud IAM enforcement.

groups:
  - name: core
    owner:
      name: core

Open complete file