Bruin CLI Step 3 of 6

Shopify Data Pipeline

Guide overview
  1. Set Up Your Project & Warehouse8 min
  2. Ingest Your Ecommerce Data15 min
  3. 3Clean & Join Your Data15 min
  4. 4Build Ecommerce Reports15 min
  5. 5Set Up Your AI Data Analyst10 min
  6. 6Build Ecommerce Dashboards5 min
PreviousIngest Your Ecommerce DataNextBuild Ecommerce Reports

Build staging models

Create a staging/ subfolder to keep your staging assets organized:

mkdir -p ecommerce/assets/staging

Using Claude Code to write staging SQL

This is where Claude Code really helps. Instead of writing every SQL file from scratch, you can ask Claude Code to look at your raw tables and generate the staging queries for you:

Look at the schemas for raw.shopify_orders and raw.stripe_charges in my warehouse. Write a Bruin SQL asset that joins them into a staging.stg_orders table - deduplicate, convert Stripe cents to dollars, and add not_null and unique checks on order_id. Put it in ecommerce/assets/staging/stg_orders.sql.

Claude Code will use the Bruin MCP to inspect the actual column names and types in your warehouse, then write SQL that matches your real data. You can do this for each staging table below.

The SQL examples below are what the files should look like. You can create them manually, or have Claude Code generate them and just review the output.

Tip

Production tips from a real Shopify pipeline: Shopify's raw data has some quirks to handle in staging:

  • Test orders: Filter with WHERE test IS NOT TRUE to exclude test transactions
  • Deduplication: Shopify can send the same order multiple times. Add QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 to keep only the latest version
  • JSON fields: Fields like customer, shipping_address, and line_items are nested JSON. Extract them with JSON_EXTRACT_SCALAR(customer, '$.id') (BigQuery) or equivalent
  • Product IDs: Shopify product IDs come in GID format (gid://shopify/Product/123456789). Parse with REPLACE(id, 'gid://shopify/Product/', '') in your products staging table
  • Line items: For product-level analytics, create a staging.stg_order_line_items table that flattens the line_items JSON array using UNNEST(JSON_EXTRACT_ARRAY(line_items))

1) Staged orders - stg_orders.sql

This is the core table: every order, enriched with Stripe payment data.

Warning

The SQL below includes a LEFT JOIN to raw.stripe_charges and lists it in depends. This only applies if you set up Stripe in Step 2. If you're using Shopify Payments, remove raw.stripe_charges from the depends list and remove the LEFT JOIN and the three stripe_* columns from the SELECT. The payment data is already in raw.shopify_orders (the financial_status and total_price fields). Ask Claude Code to adjust the query for you if needed.

Create ecommerce/assets/staging/stg_orders.sql:

/* @bruin
name: staging.stg_orders
type: sql
materialization:
  type: table
depends:
  - raw.shopify_orders
  - raw.stripe_charges
columns:
  - name: order_id
    type: varchar
    checks:
      - name: not_null
      - name: unique
  - name: order_date
    type: timestamp
    checks:
      - name: not_null
custom_checks:
  - name: has_rows
    query: "SELECT count(*) > 0 FROM staging.stg_orders"
    value: 1
@bruin */

SELECT
    o.id AS order_id,
    o.order_number,
    o.email AS customer_email,
    o.created_at AS order_date,
    o.financial_status AS payment_status,
    o.fulfillment_status,
    CAST(o.total_price AS DECIMAL(12,2)) AS order_total,
    CAST(o.subtotal_price AS DECIMAL(12,2)) AS subtotal,
    CAST(o.total_tax AS DECIMAL(12,2)) AS tax_amount,
    CAST(o.total_discounts AS DECIMAL(12,2)) AS discount_amount,
    o.currency,
    o.cancel_reason,
    o.cancelled_at,
    c.amount / 100.0 AS stripe_charge_amount,
    c.status AS stripe_status,
    c.paid AS stripe_paid
FROM raw.shopify_orders o
LEFT JOIN raw.stripe_charges c
    ON o.email = c.receipt_email
    AND toDate(o.created_at) = toDate(c.created)

The LEFT JOIN keeps all Shopify orders even if the Stripe charge hasn't synced yet. The / 100.0 converts Stripe's cent-based amounts to dollars.

2) Staged customers - stg_customers.sql

A unified customer table that merges Shopify and Stripe profiles. This query uses COALESCE, LEAST, FULL OUTER JOIN, and lower() - all of which work the same across warehouses, so there is no warehouse-specific variant needed.

Warning

The SQL below joins raw.shopify_customers with raw.stripe_customers. This only applies if you set up Stripe in Step 2. If you're using Shopify Payments, remove raw.stripe_customers from the depends list, replace the FULL OUTER JOIN with a plain SELECT from raw.shopify_customers, and drop the stripe_* columns. Ask Claude Code to adjust the query for you if needed.

Create ecommerce/assets/staging/stg_customers.sql:

/* @bruin
name: staging.stg_customers
type: sql
materialization:
  type: table
depends:
  - raw.shopify_customers
  - raw.stripe_customers
columns:
  - name: customer_email
    type: varchar
    checks:
      - name: not_null
      - name: unique
@bruin */

SELECT
    COALESCE(sc.email, st.email) AS customer_email,
    sc.id AS shopify_customer_id,
    st.id AS stripe_customer_id,
    sc.first_name,
    sc.last_name,
    sc.created_at AS shopify_created_at,
    st.created AS stripe_created_at,
    LEAST(sc.created_at, st.created) AS first_seen_at,
    sc.orders_count,
    CAST(sc.total_spent AS DECIMAL(12,2)) AS shopify_total_spent,
    sc.tags AS customer_tags,
    sc.state AS customer_state
FROM raw.shopify_customers sc
FULL OUTER JOIN raw.stripe_customers st
    ON lower(sc.email) = lower(st.email)
WHERE COALESCE(sc.email, st.email) IS NOT NULL

The FULL OUTER JOIN catches customers who only exist in one system. COALESCE gives priority to Shopify's email since it's the primary ecommerce platform.

3) Staged products - stg_products.sql

A clean product catalog. This query uses only standard SQL (CAST, WHERE), so it works identically on every warehouse.

Note

Shopify product IDs are in GID format (e.g. gid://shopify/Product/123456789). Depending on how your Shopify data lands in the warehouse, the raw id column may need parsing with REPLACE(id, 'gid://shopify/Product/', '') to extract the numeric ID.

Create ecommerce/assets/staging/stg_products.sql:

/* @bruin
name: staging.stg_products
type: sql
materialization:
  type: table
depends:
  - raw.shopify_products
columns:
  - name: product_id
    type: varchar
    checks:
      - name: not_null
      - name: unique
@bruin */

SELECT
    id AS product_id,
    title AS product_name,
    product_type AS category,
    vendor,
    status AS product_status,
    CAST(price AS DECIMAL(12,2)) AS price,  -- column name depends on how your Shopify data is structured; ask Claude Code to check raw.shopify_products
    tags,
    created_at,
    updated_at
FROM raw.shopify_products
WHERE status = 'active'

4) Staged marketing spend - stg_marketing_spend.sql

A unified marketing performance table that combines your ad platform and email marketing data into a common format. This query varies by warehouse (date functions), by marketing tool (email side), and by ads platform (ads side).

Warning

Skipped a data source? If you skipped advertising or email marketing in the stack picker, the irrelevant portions below are already hidden. If you skipped both, this entire section is hidden automatically. For the remaining parts, adjust the SQL to remove references to skipped sources. Ask Claude Code to help.

Create ecommerce/assets/staging/stg_marketing_spend.sql:

The ads portion of this query depends on which ad platform you connected in Step 2. Each platform uses its own table name and date column, so select the tab that matches your ad platform. The date casting also differs by warehouse - adjust the toDate() / DATE() / ::DATE wrapper to match yours (see the warehouse tabs in the web sessions section below for examples).

Ads portion

/* @bruin
name: staging.stg_marketing_spend
type: sql
materialization:
  type: table
depends:
  - raw.facebook_ad_insights
columns:
  - name: spend_date
    type: date
    checks:
      - name: not_null
@bruin */

-- Facebook Ads spend
SELECT
    DATE(date_start) AS spend_date,       -- use toDate() for ClickHouse, ::DATE for Snowflake
    'paid_ads' AS channel,
    campaign_name,
    CAST(spend AS DECIMAL(12,2)) AS spend,
    CAST(impressions AS INTEGER) AS impressions,
    CAST(clicks AS INTEGER) AS clicks,
    CAST(conversions AS INTEGER) AS conversions
FROM raw.facebook_ad_insights

Email marketing portion

The email side of the query uses UNION ALL to append email campaign data below the ads data. The columns and table names differ by email platform:

Append this to the ads query above with UNION ALL:

UNION ALL

-- Klaviyo email campaigns (no direct spend, but track engagement)
SELECT
    send_time::date AS spend_date,   -- adjust cast for your warehouse
    'email' AS channel,
    name AS campaign_name,
    0.00 AS spend,
    num_recipients AS impressions,
    CAST(click_count AS INTEGER) AS clicks,
    CAST(conversion_count AS INTEGER) AS conversions
FROM raw.klaviyo_campaigns kc
LEFT JOIN raw.klaviyo_metrics km
    ON kc.id = km.campaign_id
WHERE send_time IS NOT NULL

Add raw.klaviyo_campaigns and raw.klaviyo_metrics to the depends list in the Bruin asset header.

Note

The email portion needs both the right source columns (Klaviyo vs HubSpot) and the right date cast for your warehouse. The ::date cast above is Snowflake syntax - ClickHouse uses toDate(), BigQuery uses DATE(). Ask Claude Code to generate the combined SQL for your specific stack.

5) Staged web sessions - stg_web_sessions.sql

Clean session data with traffic source mapping. The column names differ depending on whether you use GA4 or Mixpanel, and the date functions differ by warehouse.

Create ecommerce/assets/staging/stg_web_sessions.sql:

Source-specific columns

The raw table names differ by analytics tool. GA4 creates raw.ga4_sessions and raw.ga4_events; Mixpanel creates raw.mixpanel_events (no separate sessions table). Select the tab that matches your analytics tool:

GA4 gives you sessions and events as separate tables. The CASE statement maps GA4 traffic sources to the same channel names used in stg_marketing_spend, so you can join them later for attribution.

SELECT
    s.date AS session_raw_date,       -- cast with your warehouse's date function
    s.sessions AS total_sessions,
    s.new_users,
    s.engaged_sessions,
    e.event_count AS purchase_events,
    CASE
        WHEN s.source = 'facebook' THEN 'paid_ads'
        WHEN s.medium = 'email' THEN 'email'
        WHEN s.medium = 'organic' THEN 'organic_search'
        WHEN s.medium = 'cpc' THEN 'paid_search'
        WHEN s.source = '(direct)' THEN 'direct'
        ELSE 'other'
    END AS channel
FROM raw.ga4_sessions s
LEFT JOIN raw.ga4_events e
    ON s.date = e.date
    AND e.event_name = 'purchase'

Warehouse date casting

Wrap the source query above in the Bruin asset header and apply the correct date function for your warehouse. Update the depends: list to match the analytics tool you chose above (GA4 uses raw.ga4_sessions and raw.ga4_events; Mixpanel uses raw.mixpanel_events):

Use toDate() for the date column:

/* @bruin
name: staging.stg_web_sessions
type: sql
materialization:
  type: table
depends:
  - raw.ga4_sessions       -- GA4: use raw.ga4_sessions and raw.ga4_events
  - raw.ga4_events         -- Mixpanel: replace both with raw.mixpanel_events
columns:
  - name: session_date
    type: date
    checks:
      - name: not_null
@bruin */

SELECT
    toDate(session_raw_date) AS session_date,
    -- ... rest of columns from source query above

Note

The web sessions query needs both the right source columns (GA4 vs Mixpanel) and the right date cast for your warehouse. Ask Claude Code to generate the combined SQL for your specific stack - it will inspect your raw tables and write the correct query.

Validate and run the staging layer

bruin validate .

Fix any errors, then run with the same small date range you used in Step 2:

bruin run --start-date 2025-01-01 --end-date 2025-02-01 .

Bruin executes assets in dependency order - ingestors first, then staging. You can also run just one asset to test it:

bruin run --start-date 2025-01-01 --end-date 2025-02-01 ecommerce/assets/staging/stg_orders.sql

Ask Claude Code to validate and troubleshoot

After creating the staging assets, ask Claude Code to check everything:

Run bruin validate on the project. If there are errors in the staging SQL files, look at the raw table schemas and fix the column names.

If a query fails at runtime, Claude Code can inspect the error and the actual table data:

The stg_customers asset failed. Query the raw.shopify_customers table to check what columns are actually available, and fix the SQL.

Sign up to our newsletter

Practical updates on open-source data pipelines, AI analysts, governance, and what we are shipping at Bruin.