Shopify Streaming Import Integration

Shopify is a leading e-commerce platform that powers online stores for businesses of all sizes. With this Streaming Ingress Connector, you can receive real-time webhook events from your Shopify store into Treasure AI, enabling immediate analytics, triggers, and personalization based on customer purchases, cart activity, and product changes.

This integration receives webhook events sent by Shopify, organized into the following categories:

For a complete list of webhook events and their data structures, refer to the Shopify Webhooks documentation.

• Basic knowledge of Treasure AI
• Basic knowledge of Shopify
• A Shopify store with webhook configuration permissions
• A Treasure AI API key (write-only)

• You must create a database and table in Plazma before creating a source to ingest data
• The Shopify webhook client secret is required for HMAC-SHA256 signature validation
• Maximum payload size is subject to Shopify's webhook payload limits
• Webhook events for unsupported topics are silently filtered
• Duplicate webhook deliveries are automatically deduplicated within a 24-hour window

If your security policy requires IP whitelisting, you must add Treasure Data's IP addresses to your allowlist to ensure a successful connection.

Please find the complete list of static IP addresses, organized by region, at the following document

1. Open Treasure AI Console.
2. Select Integrations Hub.
3. Select Catalog.
4. Search for the Shopify Streaming connector in the Catalog; hover your mouse over the icon and select Create Authentication.
5. Ensure that the Credentials tab is selected and then enter credential information for the integration.

1. Open Treasure AI Console.
2. Navigate to Integrations Hub > Authentications.
3. Locate your new authentication and select New Source.

1. Type a source name in the Data Transfer Name field.
2. Select Next.
3. The Create Source page displays with the Source Table tab selected.

1. Configure the destination table where Shopify data will be stored and select Next.

The source ID (UUID v4) is issued upon Source creation.

After creating the Source, you are automatically taken to the Sources listing page.

1. Search for the source you created.
2. Click on "..." in the same row and select Copy Unique ID.

This Unique ID is the Source ID required when configuring the webhook URL in Shopify.

To receive Shopify events in Treasure AI, you must configure a webhook in your Shopify admin.

1. Log in to your Shopify Admin.
2. Navigate to Settings > Notifications > Webhooks.
3. Select Create webhook.
4. Choose the event topic you want to subscribe to (for example, Order creation).
5. Set the Format to JSON.
6. Enter the webhook URL for your Treasure AI region. Replace {source_id} with the Source ID you copied in the previous step.

https://shopify-in-streaming.treasuredata.com/v1/task/{source_id}

1. Select the latest Webhook API version.
2. Select Save.

Shopify sends the following HTTP headers with each webhook delivery. The connector uses these headers for authentication, deduplication, and metadata:

The connector starts ingesting Shopify event data into Treasure AI as soon as the first event is triggered after successful webhook configuration and source creation.

The connector returns the following HTTP status codes to Shopify:

The connector automatically flattens nested JSON payloads from Shopify webhooks. Nested objects are flattened using underscore-separated keys, and arrays are serialized as JSON strings.

• Lowercase letters and underscores only
• Top-level scalar fields: field_name
• Nested fields: parent_child (e.g., shipping_address_country)

Each event record includes the following metadata fields in addition to the flattened webhook payload:

Order Creation Event Example:

{
  "topic": "orders/create",
  "shop_domain": "example-store.myshopify.com",
  "api_version": "2026-01",
  "webhook_id": "b1234567-89ab-cdef-0123-456789abcdef",
  "triggered_at": "2026-01-15T10:30:00.000Z",
  "received_time": 1768473000,
  "id": 5678901234,
  "email": "customer@example.com",
  "created_at": "2026-01-15T10:30:00-05:00",
  "updated_at": "2026-01-15T10:30:00-05:00",
  "total_price": "59.99",
  "currency": "USD",
  "financial_status": "paid",
  "fulfillment_status": null,
  "customer_id": 1234567890,
  "customer_email": "customer@example.com",
  "customer_first_name": "Jane",
  "customer_last_name": "Doe",
  "shipping_address_city": "New York",
  "shipping_address_country": "United States",
  "shipping_address_province": "New York",
  "line_items": "[{\"id\":9876543210,\"title\":\"Example Product\",\"quantity\":1,\"price\":\"59.99\"}]",
  "time": 1768473005
}

Customer Creation Event Example:

{
  "topic": "customers/create",
  "shop_domain": "example-store.myshopify.com",
  "api_version": "2026-01",
  "webhook_id": "c2345678-90ab-cdef-1234-567890abcdef",
  "triggered_at": "2026-01-15T09:15:00.000Z",
  "received_time": 1768468500,
  "id": 1234567890,
  "email": "newcustomer@example.com",
  "first_name": "John",
  "last_name": "Smith",
  "phone": "+1-555-0100",
  "created_at": "2026-01-15T09:15:00-05:00",
  "updated_at": "2026-01-15T09:15:00-05:00",
  "orders_count": 0,
  "state": "enabled",
  "total_spent": "0.00",
  "default_address_city": "San Francisco",
  "default_address_province": "California",
  "default_address_country": "United States",
  "default_address_zip": "94105",
  "time": 1768468505
}

To query the ingested data, use SQL queries. Example:

SELECT
  id AS order_id,
  topic,
  email,
  total_price,
  currency,
  financial_status,
  customer_first_name,
  customer_last_name,
  shipping_address_city,
  shipping_address_country,
  line_items,
  time
FROM
  your_database.your_table
WHERE
  topic = 'orders/create'
  AND TD_INTERVAL(time, '-1d', 'JST')