How to Load Adobe Commerce (Magento 2) Orders into Snowflake
Copy Adobe Commerce (Magento 2) orders and order lines into Snowflake every hour through the Magento REST API and the Snowflake SQL API, with merges so changed orders update in place.
What This Integration Does
Getting Magento data into a warehouse usually means a database replica, an extension, or an ETL subscription. This Spojit workflow uses the REST APIs on both sides instead. Every hour it asks Adobe Commerce for orders updated since the last run, flattens them into an orders table and an order lines table, and merges them into Snowflake in two SQL statements. Because it reads by updated_at and merges on the order ID, status changes, invoices, shipments and cancellations show up in Snowflake on the next run, not only new orders.
Re-running is safe: the merge updates rows that exist and inserts the rest, and order lines are replaced per order, so an overlapping window or a retried run leaves the same data in Snowflake.
Prerequisites
- An Adobe Commerce REST connection with an integration token that can read Sales.
- A Snowflake HTTP Requests connection (Base URL
https://YOUR_ACCOUNT.snowflakecomputing.com, Bearer token set to a programmatic access token for a role that can write the target schema). The "How to Generate a PDF Report from Snowflake Query Results" tutorial in this help center covers creating the token. - Target tables:
CREATE TABLE IF NOT EXISTS commerce.magento.orders ( entity_id NUMBER PRIMARY KEY, increment_id STRING, created_at TIMESTAMP_NTZ, updated_at TIMESTAMP_NTZ, status STRING, customer_email STRING, grand_total NUMBER(12,2), tax_amount NUMBER(12,2), shipping_amount NUMBER(12,2), currency STRING, country STRING, raw VARIANT); CREATE TABLE IF NOT EXISTS commerce.magento.order_lines ( order_id NUMBER, item_id NUMBER, sku STRING, name STRING, qty_ordered NUMBER(12,4), price NUMBER(12,2), row_total NUMBER(12,2));
Step 1: Schedule and Window
Add a Trigger node with type Schedule and cron 5 * * * *. Build the window start as the trigger time minus 75 minutes in YYYY-MM-DD HH:mm:ss (Magento's format, UTC): a Date & Time subtract step, then a format step named since, whose text is in {{ since.result }}.
Step 2: Read Updated Orders
The list-orders tool filters on creation date, so use raw-rest-request on Adobe Commerce REST to filter on updated_at:
method:GETpath:ordersqueryParams:searchCriteria[filter_groups][0][filters][0][field]=updated_at&searchCriteria[filter_groups][0][filters][0][value]={{ since.result }}&searchCriteria[filter_groups][0][filters][0][condition_type]=gteq&searchCriteria[pageSize]=100&searchCriteria[currentPage]={{ page }}
Loop with page starting at 1 until the collected count reaches the response's total_count. Name the collected orders orders.
Step 3: Flatten Orders and Lines
Add a Code Runner step with execute-javascript and input {{ orders }}:
const orders = [], lines = [];
for (const o of data) {
const ship = o.extension_attributes?.shipping_assignments?.[0]?.shipping?.address;
orders.push({
entity_id: o.entity_id, increment_id: o.increment_id,
created_at: o.created_at, updated_at: o.updated_at, status: o.status,
customer_email: o.customer_email, grand_total: o.grand_total,
tax_amount: o.tax_amount, shipping_amount: o.shipping_amount,
currency: o.order_currency_code, country: ship?.country_id ?? null, raw: o
});
for (const i of o.items.filter(i => !i.parent_item_id)) {
lines.push({ order_id: o.entity_id, item_id: i.item_id, sku: i.sku, name: i.name,
qty_ordered: i.qty_ordered, price: i.price, row_total: i.row_total });
}
}
return { orders: JSON.stringify(orders), lines: JSON.stringify(lines),
ids: orders.map(o => o.entity_id).join(",") };Name it flat. Skipping items with a parent_item_id avoids counting configurable products twice (Magento lists the parent and the child).
Step 4: Merge Orders
Add an HTTP Requests http-post step on the Snowflake connection to /api/v2/statements, with header X-Snowflake-Authorization-Token-Type: PROGRAMMATIC_ACCESS_TOKEN and body:
{
"statement": "MERGE INTO commerce.magento.orders t USING (SELECT f.value v FROM TABLE(FLATTEN(INPUT => PARSE_JSON(?))) f) s ON t.entity_id = s.v:entity_id::NUMBER WHEN MATCHED THEN UPDATE SET status = s.v:status::STRING, updated_at = s.v:updated_at::TIMESTAMP_NTZ, grand_total = s.v:grand_total::NUMBER(12,2), tax_amount = s.v:tax_amount::NUMBER(12,2), shipping_amount = s.v:shipping_amount::NUMBER(12,2), raw = s.v:raw WHEN NOT MATCHED THEN INSERT (entity_id, increment_id, created_at, updated_at, status, customer_email, grand_total, tax_amount, shipping_amount, currency, country, raw) VALUES (s.v:entity_id::NUMBER, s.v:increment_id::STRING, s.v:created_at::TIMESTAMP_NTZ, s.v:updated_at::TIMESTAMP_NTZ, s.v:status::STRING, s.v:customer_email::STRING, s.v:grand_total::NUMBER(12,2), s.v:tax_amount::NUMBER(12,2), s.v:shipping_amount::NUMBER(12,2), s.v:currency::STRING, s.v:country::STRING, s.v:raw)",
"bindings": { "1": { "type": "TEXT", "value": "{{ flat.result.orders }}" } },
"warehouse": "LOAD_WH",
"role": "LOADER",
"timeout": 120
}Step 5: Replace the Lines for These Orders
Lines can change when an order is edited, so replace them per order. Send two requests (Snowflake does not allow bind variables in a multi-statement request): first DELETE FROM commerce.magento.order_lines WHERE order_id IN ({{ flat.result.ids }}), then an INSERT INTO commerce.magento.order_lines SELECT f.value:order_id::NUMBER, f.value:item_id::NUMBER, f.value:sku::STRING, f.value:name::STRING, f.value:qty_ordered::NUMBER(12,4), f.value:price::NUMBER(12,2), f.value:row_total::NUMBER(12,2) FROM TABLE(FLATTEN(INPUT => PARSE_JSON(?))) f with {{ flat.result.lines }} as the binding. Add a Condition before both that skips them when no orders changed.
Tips
- Keep the raw order: the
rawVARIANT column keeps every field Magento returned, so you can query payment method, coupon codes or custom attributes later without changing the pipeline. - Backfill in months: for history, run the workflow manually with fixed monthly windows rather than one huge request.
- Customers and products: the same pattern works for
customers/searchandproductswith anupdated_atfilter. - Join with marketing: with Klaviyo data in the same warehouse (see "How to Load Klaviyo Data into Snowflake"), you can compare email-attributed revenue with real order revenue.
Common Pitfalls
- Timezones: Magento stores
created_atandupdated_atin UTC. Buildsincein UTC and load intoTIMESTAMP_NTZcolumns treated as UTC. - Store code: REST calls go to the store code set on the connection. Use the default store code so orders from every store view are returned.
- Large pages: very large orders make big responses. Keep
pageSizeat 100 or below. - Asynchronous statements: a merge that does not finish within Snowflake's synchronous wait (about 45 seconds) returns
202with a statement handle and keeps running; poll it, or load in smaller batches.timeoutis when Snowflake cancels the statement.
Testing
Run the workflow manually with a two-hour window, then compare SELECT COUNT(*), SUM(grand_total) FROM commerce.magento.orders for those orders with the Magento admin's order grid. Change one test order's status, run again, and check that the row updated rather than duplicated.