Amazon Data Warehouse Architecture

Why Three Layers?

Each layer has a specific job. Staging handles source system quirks. Intermediate handles data quality. Marts handle business logic. When something breaks, you know exactly where to look. When requirements change, you know exactly what to modify.

dbt Staging Model Example

-- models/staging/amazon/stg_amazon__orders.sql

with source as (
    select * from {{ source('amazon_raw', 'orders') }}
),

renamed as (
    select
        -- Primary key
        amazon_order_id,
        
        -- Foreign keys
        marketplace_id,
        
        -- Dates (standardize to timestamp)
        cast(purchase_date as timestamp) as purchase_date,
        cast(last_update_date as timestamp) as last_update_date,
        
        -- Order details
        order_status,
        fulfillment_channel,
        sales_channel,
        ship_service_level,
        
        -- Financials (cast to numeric)
        cast(order_total_amount as numeric(18,2)) as order_total,
        order_total_currency_code as currency,
        
        -- Metadata
        _loaded_at,
        _source_file
        
    from source
)

select * from renamed

Product Master Model

-- models/intermediate/int_product_master.sql

with catalog as (
    select * from {{ ref('stg_amazon__catalog') }}
),

inventory as (
    select distinct sku, fnsku, asin
    from {{ ref('stg_amazon__fba_inventory') }}
),

orders as (
    select distinct sku, asin
    from {{ ref('stg_amazon__order_items') }}
),

-- Build comprehensive mapping
product_mappings as (
    select
        coalesce(c.asin, i.asin, o.asin) as asin,
        coalesce(i.sku, o.sku) as sku,
        i.fnsku,
        c.item_name as product_name,
        c.brand,
        c.item_type as category,
        c.marketplace_id
    from catalog c
    full outer join inventory i on c.asin = i.asin
    full outer join orders o on c.asin = o.asin
),

-- Deduplicate and pick best values
final as (
    select
        asin,
        sku,
        fnsku,
        first_value(product_name ignore nulls) over (
            partition by asin order by product_name
        ) as product_name,
        first_value(brand ignore nulls) over (
            partition by asin order by brand
        ) as brand,
        marketplace_id
    from product_mappings
    qualify row_number() over (
        partition by asin, sku, marketplace_id order by fnsku
    ) = 1
)

select * from final

Daily P&L Model

-- models/marts/finance/fct_daily_pnl.sql

with orders as (
    select * from {{ ref('int_orders_enriched') }}
),

fees as (
    select * from {{ ref('int_financial_transactions') }}
),

ads as (
    select * from {{ ref('int_ad_performance') }}
),

daily_revenue as (
    select
        date_trunc('day', purchase_date) as date,
        sku,
        marketplace_id,
        sum(item_price) as gross_revenue,
        sum(item_tax) as tax_collected,
        count(distinct amazon_order_id) as orders,
        sum(quantity) as units_sold
    from orders
    where order_status not in ('Cancelled', 'Pending')
    group by 1, 2, 3
),

daily_fees as (
    select
        date_trunc('day', posted_date) as date,
        sku,
        marketplace_id,
        sum(case when fee_category = 'referral' then amount else 0 end) as referral_fees,
        sum(case when fee_category = 'fba_fulfillment' then amount else 0 end) as fba_fees,
        sum(case when fee_category = 'fba_storage' then amount else 0 end) as storage_fees,
        sum(case when fee_category = 'refund' then amount else 0 end) as refunds
    from fees
    group by 1, 2, 3
),

daily_ads as (
    select
        date,
        sku,
        marketplace_id,
        sum(spend) as ad_spend,
        sum(sales) as attributed_sales,
        sum(impressions) as impressions,
        sum(clicks) as clicks
    from ads
    group by 1, 2, 3
),

final as (
    select
        coalesce(r.date, f.date, a.date) as date,
        coalesce(r.sku, f.sku, a.sku) as sku,
        coalesce(r.marketplace_id, f.marketplace_id, a.marketplace_id) as marketplace_id,
        
        -- Revenue
        coalesce(r.gross_revenue, 0) as gross_revenue,
        coalesce(r.units_sold, 0) as units_sold,
        coalesce(r.orders, 0) as orders,
        
        -- Fees
        coalesce(f.referral_fees, 0) as referral_fees,
        coalesce(f.fba_fees, 0) as fba_fees,
        coalesce(f.storage_fees, 0) as storage_fees,
        coalesce(f.refunds, 0) as refunds,
        
        -- Ads
        coalesce(a.ad_spend, 0) as ad_spend,
        coalesce(a.attributed_sales, 0) as attributed_sales,
        
        -- Calculated metrics
        coalesce(r.gross_revenue, 0) 
            - coalesce(f.referral_fees, 0) 
            - coalesce(f.fba_fees, 0) 
            - coalesce(f.storage_fees, 0)
            - coalesce(f.refunds, 0)
            - coalesce(a.ad_spend, 0) as contribution_margin
            
    from daily_revenue r
    full outer join daily_fees f 
        on r.date = f.date and r.sku = f.sku and r.marketplace_id = f.marketplace_id
    full outer join daily_ads a 
        on r.date = a.date and r.sku = a.sku and r.marketplace_id = a.marketplace_id
)

select * from final

Recommended Directory Structure

dbt_project/
├── models/
│   ├── staging/
│   │   └── amazon/
│   │       ├── _amazon__sources.yml
│   │       ├── _amazon__models.yml
│   │       ├── stg_amazon__orders.sql
│   │       ├── stg_amazon__order_items.sql
│   │       ├── stg_amazon__settlements.sql
│   │       ├── stg_amazon__fba_inventory.sql
│   │       ├── stg_amazon__sp_campaigns.sql
│   │       └── stg_amazon__catalog.sql
│   │
│   ├── intermediate/
│   │   ├── _int__models.yml
│   │   ├── int_orders_enriched.sql
│   │   ├── int_financial_transactions.sql
│   │   ├── int_product_master.sql
│   │   ├── int_inventory_snapshots.sql
│   │   └── int_ad_performance.sql
│   │
│   └── marts/
│       ├── finance/
│       │   ├── _finance__models.yml
│       │   ├── fct_daily_pnl.sql
│       │   ├── fct_monthly_pnl.sql
│       │   └── dim_fee_types.sql
│       │
│       ├── sales/
│       │   ├── _sales__models.yml
│       │   ├── fct_orders.sql
│       │   ├── fct_order_items.sql
│       │   └── dim_products.sql
│       │
│       └── marketing/
│           ├── _marketing__models.yml
│           ├── fct_ad_performance.sql
│           └── dim_campaigns.sql
│
├── tests/
│   └── generic/
│       └── test_positive_revenue.sql
│
├── macros/
│   ├── fee_categorization.sql
│   └── currency_conversion.sql
│
└── seeds/
    └── fee_type_mapping.csv

Incremental Order Model

-- models/staging/amazon/stg_amazon__orders.sql

{{
    config(
        materialized='incremental',
        unique_key='amazon_order_id',
        incremental_strategy='merge'
    )
}}

with source as (
    select * from {{ source('amazon_raw', 'orders') }}
    {% if is_incremental() %}
    where _loaded_at > (select max(_loaded_at) from {{ this }})
    {% endif %}
),

-- rest of transformation...

dbt Test Configuration

# models/staging/amazon/_amazon__models.yml

version: 2

models:
  - name: stg_amazon__orders
    description: Cleaned Amazon orders from SP-API
    columns:
      - name: amazon_order_id
        description: Unique order identifier
        tests:
          - unique
          - not_null
      
      - name: purchase_date
        description: When the order was placed
        tests:
          - not_null
          - dbt_utils.recency:
              datepart: day
              field: purchase_date
              interval: 3
      
      - name: order_status
        description: Current order status
        tests:
          - accepted_values:
              values: ['Pending', 'Shipped', 'Cancelled', 'Unshipped']
      
      - name: order_total
        description: Total order value
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= 0"