Snowflake: Unload Data to Angler S3

Snowflake can unload your backend data directly into Angler's S3 bucket using a storage integration and an external stage. No AWS access keys are exchanged: Angler creates an IAM role scoped to your workspace folder, and Snowflake assumes that role when it writes files. Once set up, a scheduled Snowflake task runs the unload with no external scheduler.

Folder layout

Each object lands in its own folder under your workspace path. Folder names are case-sensitive.

Path pattern:

s3://<angler_bucket>/inbound/<workspace_id>/<folder>/

Your onboarding specialist will send the exact bucket name and workspace id.

Prerequisites

• A Snowflake role with ACCOUNTADMIN, or the global CREATE INTEGRATION privilege, to create the storage integration.
• A role with CREATE STAGE on the target schema and USAGE on the integration, to create the stage.
• A warehouse for the scheduled task, and read access to the four source tables.
• A Snowflake account hosted on AWS in a standard (non-government, non-China) region. If yours differs, tell your onboarding specialist before starting.

Values exchanged

Setup overview

You run Steps 1 to 5 in Snowflake; Angler handles two AWS steps (A and B) around them.

Step 1: Create the storage integration

Use the role ARN and S3 path from your onboarding specialist. Limit allowed locations to your workspace path.

USE ROLE ACCOUNTADMIN;

CREATE STORAGE INTEGRATION angler_s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<role_arn_from_angler>'
  STORAGE_ALLOWED_LOCATIONS = ('s3://<angler_bucket>/inbound/<workspace_id>/');

Step 2: Retrieve the Snowflake IAM user and external ID

DESC INTEGRATION angler_s3_int;

Record STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID from the output and send both to your onboarding specialist. Angler adds them to the role's trust policy and confirms when it is done; Step 4 will fail until then.

Step 3: Grant usage to the unload role

Skip this step if ACCOUNTADMIN will own the stage and task.

GRANT USAGE ON INTEGRATION angler_s3_int TO ROLE <unload_role>;
GRANT CREATE STAGE ON SCHEMA <db>.<schema> TO ROLE <unload_role>;

Step 4: Create the file format and external stage

Point the stage at your workspace root, so one stage serves all four folders. The trailing slash on the URL is required.

USE ROLE <unload_role>;
USE SCHEMA <db>.<schema>;

CREATE FILE FORMAT angler_csv_gz
  TYPE = CSV
  COMPRESSION = GZIP
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  NULL_IF = ('')
  EMPTY_FIELD_AS_NULL = TRUE;

CREATE STAGE angler_stage
  STORAGE_INTEGRATION = angler_s3_int
  URL = 's3://<angler_bucket>/inbound/<workspace_id>/'
  FILE_FORMAT = angler_csv_gz;

Validate the connection:

SELECT SYSTEM$VALIDATE_STORAGE_INTEGRATION(
  'angler_s3_int',
  's3://<angler_bucket>/inbound/<workspace_id>/',
  'validate_test.txt',
  'write');

Step 5: Unload data to the four folders

Run one COPY INTO per object. Start with a small sample of each, so Angler can confirm the expected fields before you schedule anything.

COPY INTO @angler_stage/customers/
  FROM (SELECT * FROM <db>.<schema>.customers)
  HEADER = TRUE
  INCLUDE_QUERY_ID = TRUE
  MAX_FILE_SIZE = 268435456;

COPY INTO @angler_stage/orders/
  FROM (SELECT * FROM <db>.<schema>.orders)
  HEADER = TRUE
  INCLUDE_QUERY_ID = TRUE
  MAX_FILE_SIZE = 268435456;

COPY INTO @angler_stage/lineItems/
  FROM (SELECT * FROM <db>.<schema>.line_items)
  HEADER = TRUE
  INCLUDE_QUERY_ID = TRUE
  MAX_FILE_SIZE = 268435456;

COPY INTO @angler_stage/products/
  FROM (SELECT * FROM <db>.<schema>.products)
  HEADER = TRUE
  INCLUDE_QUERY_ID = TRUE
  MAX_FILE_SIZE = 268435456;

• HEADER = TRUE writes column names, which Angler uses to map fields.
• INCLUDE_QUERY_ID = TRUE gives every run unique file names, so a new drop never overwrites an earlier one.
• MAX_FILE_SIZE caps each file at 256 MB.

What to include

• customers: customer id, email, phone, name, address fields, created and updated timestamps.
• orders: order id, customer id, order-level email if available, totals, currency, status, channel or source, created and updated timestamps.
• lineItems: line id, order id, product or variant id, quantity, price, discount.
• products: product and variant ids, title, category, brand, price.

Your onboarding specialist will share the full field mapping. Two rules matter most: the customer id on orders must match the id in customers exactly, and each row should appear once per drop.

Backfill, then incremental

Send history first as a one-time full unload; agree the start date with your onboarding specialist. After that, filter each run on updated_at with a short overlap, so late edits and refunds are captured:

COPY INTO @angler_stage/orders/
  FROM (
    SELECT * FROM <db>.<schema>.orders
    WHERE updated_at >= DATEADD(day, -3, CURRENT_DATE())
  )
  HEADER = TRUE
  INCLUDE_QUERY_ID = TRUE;

Apply the same filter to customers, line items, and products.

Schedule with a Snowflake task

Wrap the four incremental statements in a stored procedure, then run it on a schedule:

CREATE OR REPLACE TASK angler_daily_unload
  WAREHOUSE = <warehouse>
  SCHEDULE = 'USING CRON 0 6 * * * UTC'
AS
  CALL <db>.<schema>.angler_unload();

ALTER TASK angler_daily_unload RESUME;

A daily cadence is strongly recommended.

Troubleshooting

Reference

• Snowflake: Configure a storage integration to access Amazon S3