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

# 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 (`orde