SUBMISSION: Project 2 — Build a Production Staging Layer for a Shopify Source

🎯 Hands-on Project Deliverable: Build the critical staging layer for an enterprise Shopify data pipeline. You will handle CDC sync duplication, configure automated source freshness alerts, and write schema assertion tests.

---

🏢 Business Scenario & Executive Prompt

Your e-commerce client ingests Shopify sales data through a Fivetran connector into Snowflake. However, during network retries and connector re-syncs, Fivetran occasionally loads the same order record twice with different sync timestamps.

Downstream executive dashboards are overcounting gross revenue by 4.2% because nobody deduplicated the raw staging records. Furthermore, currency values are stored as raw integers (cents), test orders pollute the customer metrics, and no automated alerts exist when the data pipeline stalls.

Your goal is to build a rock-solid, production-certified staging layer for Shopify in dbt that guarantees deduplication, converts units, filters test records, and flags pipeline freshness failures.

---

📋 Upstream Raw Schemas (raw_shopify)

Fivetran delivers data into the raw_shopify warehouse schema with the following fields:

1. raw_shopify.orders:

   • id (varchar / string): Unique Shopify order ID.
   • customer_id (varchar / string): FK to customers.
   • order_number (int): Sequential human-friendly order number.
   • financial_status (varchar): e.g., 'paid', 'pending', 'refunded', 'voided'.
   • fulfillment_status (varchar): e.g., 'fulfilled', 'unfulfilled', 'partial'.
   • total_price (numeric): Total order amount in dollars (or cents depending on store).
   • subtotal_price (numeric): Pre-tax order amount.
   • total_tax (numeric): Tax amount.
   • currency (varchar): 3-letter currency code (e.g. 'USD').
   • test (boolean): TRUE if test order generated by QA.
   • created_at (timestamp): Timestamp checkout completed.
   • updated_at (timestamp): Last record update timestamp.
   • _fivetran_synced (timestamp): Ingestion sync timestamp.

2. raw_shopify.customers:

   • id (varchar / string): Customer ID.
   • email (varchar): Customer email address (un-normalized).
   • first_name (varchar), last_name (varchar).
   • orders_count (int): Lifetime order count.
   • total_spent (numeric): Lifetime spend.
   • country (varchar): ISO country code.
   • created_at (timestamp), updated_at (timestamp).
   • _fivetran_synced (timestamp).

3. raw_shopify.products:

   • id (varchar / string): Product ID.
   • title (varchar): Product name.
   • product_type (varchar): Category classification.
   • vendor (varchar): Brand/vendor name.
   • status (varchar): 'active', 'draft', 'archived'.
   • _fivetran_synced (timestamp).

---

🎯 Required Deliverables

Submit a .zip archive or GitHub repository link containing:

1. models/staging/shopify/_shopify__sources.yml: Complete source declarations with loaded_at_field and freshness SLA rules.
2. models/staging/shopify/stg_shopify__orders.sql: Deduped and cleaned order staging model.
3. models/staging/shopify/stg_shopify__customers.sql: Deduped customer staging model with normalized email addresses and combined name.
4. models/staging/shopify/stg_shopify__products.sql: Staging model filtering out archived inventory items.
5. models/staging/shopify/_shopify__models.yml: Schema test file asserting uniqueness, non-null, and referential integrity.
6. dbt_project.yml: Folder-level materialization configuration declaring all staging models as views.

---

🛠️ Step-by-Step Implementation Guide

Step 1 — Declare Sources and Automated Freshness

Create models/staging/shopify/_shopify__sources.yml.

• Set schema: raw_shopify and loaded_at_field: _fivetran_synced.
• Define freshness thresholds: warn after 24 hours, error after 48 hours.
• Document every raw table.

Step 2 — Deduplicate and Clean Orders (stg_shopify__orders.sql)

• Use a CTE with ROW_NUMBER() OVER (PARTITION BY id ORDER BY _fivetran_synced DESC) AS row_num to select only the most recent version of each record (WHERE row_num = 1).
• Filter out QA test records: WHERE NOT COALESCE(test, FALSE).
• Cast string IDs to bigint for fast database joins.
• Rename fields to standard analytics conventions (financial_status → payment_status).

Step 3 — Clean and Standardize Customers (stg_shopify__customers.sql)

• Deduplicate on id using _fivetran_synced.
• Cleanse email strings: LOWER(TRIM(email)) AS email_address.
• Create a composite column: TRIM(CONCAT(first_name, ' ', last_name)) AS full_name.

Step 4 — Filter Products (stg_shopify__products.sql)

• Deduplicate on product id.
• Exclude obsolete catalog items: WHERE status != 'archived'.

Step 5 — Write Comprehensive Schema Tests (_shopify__models.yml)

• Declare unique and not_null tests on primary keys (order_id, customer_id, product_id).
• Declare a relationships test verifying that stg_shopify__orders.customer_id links to stg_shopify__customers.customer_id.
• Declare an accepted_values test on payment_status ('paid', 'pending', 'refunded', 'voided', 'partially_paid').

Step 6 — Validate Locally

Execute the dbt verification commands:

dbt source freshness                  # confirm upstream ingestion SLAs
dbt build --select staging.shopify    # compile, run, and test staging models

---

💻 Starter Scaffold

1. models/staging/shopify/_shopify__sources.yml

version: 2

sources:
  - name: shopify
    description: "Shopify e-commerce data ingested via Fivetran CDC."
    database: raw
    schema: raw_shopify
    loaded_at_field: _fivetran_synced
    freshness:
      warn_after: {count: 24, period: hour}
      error_after: {count: 48, period: hour}
    tables:
      - name: orders
        description: "Raw checkout order transactions."
        columns:
          - name: id
            tests: [unique, not_null]
      - name: customers
        description: "Raw customer profile records."
      - name: products
        description: "Raw catalog inventory items."

2. models/staging/shopify/stg_shopify__orders.sql

WITH source AS (
    SELECT * FROM {{ source('shopify', 'orders') }}
),

deduped AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY id 
            ORDER BY _fivetran_synced DESC
        ) AS dedup_rank
    FROM source
),

renamed AS (
    SELECT
        id::bigint AS order_id,
        customer_id::bigint AS customer_id,
        order_number,
        financial_status AS payment_status,
        fulfillment_status,
        total_price AS order_total_usd,
        subtotal_price AS order_subtotal_usd,
        total_tax AS tax_amount_usd,
        currency,
        created_at AS order_placed_at,
        updated_at AS order_updated_at,
        _fivetran_synced AS synced_at
    FROM deduped
    WHERE dedup_rank = 1
      AND NOT COALESCE(test, FALSE)
)

SELECT * FROM renamed

3. models/staging/shopify/stg_shopify__customers.sql

WITH source AS (
    SELECT * FROM {{ source('shopify', 'customers') }}
),

deduped AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY id 
            ORDER BY _fivetran_synced DESC
        ) AS dedup_rank
    FROM source
),

renamed AS (
    SELECT
        id::bigint AS customer_id,
        LOWER(TRIM(email)) AS email_address,
        first_name,
        last_name,
        TRIM(CONCAT(first_name, ' ', last_name)) AS full_name,
        country,
        orders_count,
        total_spent::numeric(12,2) AS lifetime_spend_usd,
        created_at AS customer_created_at,
        _fivetran_synced AS synced_at
    FROM deduped
    WHERE dedup_rank = 1
)

SELECT * FROM renamed

4. models/staging/shopify/_shopify__models.yml

version: 2

models:
  - name: stg_shopify__orders
    description: "One row per Shopify order. Deduped on order_id. Test orders filtered out."
    columns:
      - name: order_id
        description: "Primary key for Shopify orders."
        tests:
          - unique
          - not_null
      - name: customer_id
        description: "Foreign key to customer profile."
        tests:
          - not_null
          - relationships:
              to: ref('stg_shopify__customers')
              field: customer_id
      - name: payment_status
        tests:
          - accepted_values:
              values: ['paid', 'pending', 'refunded', 'voided', 'partially_paid']

  - name: stg_shopify__customers
    description: "One row per customer profile."
    columns:
      - name: customer_id
        tests:
          - unique
          - not_null
      - name: email_address
        tests:
          - not_null

---

📊 100-Point Evaluation Rubric

Evaluation Dimension	Weight	Target Standard (Pass ≥ 80%)
1. Source Freshness & Declarations	25 pts	loaded_at_field configured; reasonable freshness warning/error thresholds set; tables described in _shopify__sources.yml.
2. Fivetran Deduplication Pattern	25 pts	Robust ROW_NUMBER() OVER (PARTITION BY id ORDER BY _fivetran_synced DESC) implemented to guard against sync duplicate records.
3. Cleaning & Type Normalization	25 pts	Test orders filtered out (NOT COALESCE(test, FALSE)); IDs cast to bigints; clean column aliases adhering to dbt conventions.
4. Schema Test Coverage	25 pts	unique and not_null assertions declared on primary keys; relationships test referencing parent customer IDs; accepted_values test on statuses.

---

📤 Submission Instructions & Guidelines

1. Deliverable Formats: Upload your completed models/staging/shopify/ folder as a .zip archive, or upload stg_shopify__orders.sql directly and provide your GitHub repository/PR URL in the Submission Notes.
2. Dialect Compatibility: Ensure all DDL and queries execute cleanly on standard ANSI SQL, Snowflake, BigQuery, Databricks, or PostgreSQL.
3. Self-Review: Cross-check your solution against the 100-Point Evaluation Rubric above before submitting (passing threshold is 80/100 points).
4. Upload: Use the drag-and-drop file upload zone below. You can optionally detail architectural trade-offs in the submission notes box.
5. Confirmation: Click Submit Assignment. Your submission will be recorded, your lesson and module progress will be marked complete, and you can download your submitted work at any time.