Export Wingspan data to Snowflake

A data export runs on a schedule you choose, from hourly to daily. Each run writes the rows that changed since the last run to an S3 bucket in your AWS account, one Parquet file per resource. Snowflake reads that bucket through a storage integration, and Snowpipe loads each file as it lands.

If you'd rather not run an S3 bucket, we can deliver the same files to a bucket we manage. See Without your own S3 bucket.

flowchart LR
  WS[Wingspan data export] -->|Parquet files per run| S3[(Your S3 bucket)]
  S3 -->|Event notification| SP[Snowpipe]
  SP --> RAW[Raw tables in Snowflake]
  RAW --> CUR[Current-state views]

What you can export

Pick any of these resources for entities: Account, Person, Payee, PayeeEngagement, PayerEngagement, Invoice, Payable, PayrollRun, PayStatement, Payout, FundsMovement, Form, TaxFiling, IdentityVerification and Requirement.

Each resource exports all of its non-sensitive fields unless you narrow the columns with fieldSelection. Person and Account default to id, externalId, status, events, actors and metadata. To include their personal data, list the resource in piiEntities, name the fields in fieldSelection, and use a credential with the data-export.{resource}.pii:read scope, such as data-export.person.pii:read.

Before you start

A Wingspan credential with the platform.data-export:write scope to create and change exports, and platform.data-export:read to read them and their runs. A service account is the usual choice.
An S3 bucket in your AWS account, and the AWS region it's in.
A Snowflake role that can create a storage integration, such as ACCOUNTADMIN.
Wingspan's AWS account ID for your trust policy. Your Wingspan contact gives you this.

Step 1. Let Wingspan write to your bucket

Create an IAM role in your AWS account for Wingspan to assume. Its trust policy allows Wingspan's AWS account and requires an external ID in the form wingspan-{accountId}-{dataExportId}, where accountId is your Wingspan Account ID.

You don't have the export's ID until you create it, so start with a wildcard for that part:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "AWS": "arn:aws:iam::<wingspan-aws-account-id>:root" },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringLike": { "sts:ExternalId": "wingspan-<your-wingspan-account-id>-*" }
      }
    }
  ]
}

Give the role permission to write under your export prefix:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::acme-wingspan-exports/wingspan/*"
    }
  ]
}

If the bucket uses SSE-KMS encryption, also allow kms:GenerateDataKey on the key. After Step 2, you can replace the wildcard with your export's ID.

Step 2. Create the export

Send Create a scheduled data export with an Idempotency-Key:

curl -X POST https://api.wingspan.app/v3/platform/data-exports \
  -H "Authorization: Bearer $WINGSPAN_TOKEN" \
  -H "X-Wingspan-Account: $WINGSPAN_ACCOUNT_ID" \
  -H "Idempotency-Key: $(uuidgen)" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Snowflake: AP and AR",
    "externalId": "snowflake-prod",
    "entities": ["Payee", "Invoice", "Payable", "Payout", "TaxFiling"],
    "destination": {
      "type": "S3",
      "roleArn": "arn:aws:iam::111122223333:role/wingspan-export-writer",
      "bucket": "acme-wingspan-exports",
      "prefix": "wingspan/{resource}/{ymd}/{runId}",
      "region": "us-east-1"
    },
    "schedule": { "cron": "0 * * * *", "timezone": "America/New_York" },
    "format": "Parquet"
  }'

The response is 201 with the export's id and status: Active.

Export settings

prefix accepts the tokens {resource}, {ymd} (UTC date), {hourBucket} (UTC hour) and {runId}. Tokens render in lowercase, so Payable files land under wingspan/payable/. Keep {resource} near the front: each Snowpipe in Step 4 reads one resource's folder.
schedule takes a five-field cron expression and an IANA time zone. It runs at most hourly and at least daily.
Use Parquet for Snowflake. JsonLines and Csv are also available.
Each run picks up rows whose events.updatedAt changed since the previous run. Set incrementalWatermark to use a different field.
To export only some rows, add filters. For example, "filters": { "Payable": { "status": { "anyOf": ["Approved", "Paid"] } } } exports only approved and paid payables. A row must have changed in the run's window and match the filter.
An externalId you've already used returns 409.

Step 3. Connect Snowflake to the bucket

Snowflake reads the bucket with its own IAM role. Create a second role in your AWS account with s3:GetObject, s3:GetObjectVersion, s3:ListBucket and s3:GetBucketLocation on the bucket, plus kms:Decrypt on the key if the bucket uses SSE-KMS. Then create the storage integration:

CREATE STORAGE INTEGRATION wingspan_s3
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::111122223333:role/snowflake-wingspan-reader'
  STORAGE_ALLOWED_LOCATIONS = ('s3://acme-wingspan-exports/wingspan/');

DESC INTEGRATION wingspan_s3;

Copy STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID from the output into the reader role's trust policy. Snowflake's guide to configuring a storage integration for Amazon S3 has the full policy.

Then create a file format and a stage:

CREATE FILE FORMAT wingspan_parquet TYPE = PARQUET;

CREATE STAGE wingspan_stage
  URL = 's3://acme-wingspan-exports/wingspan/'
  STORAGE_INTEGRATION = wingspan_s3
  FILE_FORMAT = wingspan_parquet;

After the first run, LIST @wingspan_stage; lists the files, with one folder for each resource.

Step 4. Load each resource with Snowpipe

Give each resource a raw table that keeps every row from every file, with the file it came from:

CREATE TABLE raw_payable (
  record VARIANT,
  file_name STRING,
  loaded_at TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE PIPE payable_pipe AUTO_INGEST = TRUE AS
  COPY INTO raw_payable (record, file_name)
  FROM (SELECT $1, METADATA$FILENAME FROM @wingspan_stage/payable/);

Repeat for each resource you export. Then run SHOW PIPES; and copy the notification_channel value. In the S3 console, add an event notification on the bucket for all object create events under wingspan/, with that SQS queue as the destination. Snowpipe then loads each new file as it lands. Snowflake's guide to automating Snowpipe for Amazon S3 covers the notification setup in detail.

To load files that landed in the seven days before the pipe existed, run ALTER PIPE payable_pipe REFRESH;.

Step 5. Keep a current view of each resource

A record that changes in several runs appears in several files. Keep the copy with the latest events.updatedAt. Load order isn't enough, because a retried run exports an earlier window again and can load after newer files:

CREATE VIEW payable AS
SELECT record:id::STRING AS id, record, loaded_at
FROM raw_payable
QUALIFY ROW_NUMBER() OVER (PARTITION BY record:id ORDER BY record:events:updatedAt::TIMESTAMP_TZ DESC, loaded_at DESC) = 1;

Build your reporting tables on these views, or use a scheduled task to MERGE them into tables if your queries need the extra speed.

Without your own S3 bucket

We can deliver the export files to a Google Cloud Storage bucket we manage, and your Snowflake account reads them from there. Ask your Wingspan contact to set it up, with the resources and schedule you want. We create the export, so skip Steps 1 to 3 and connect Snowflake to our bucket instead.

Create a storage integration for Google Cloud Storage, using the bucket path we give you:

CREATE STORAGE INTEGRATION wingspan_gcs
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://<bucket-from-wingspan>/<your-path>/');

DESC STORAGE INTEGRATION wingspan_gcs;

Send us the STORAGE_GCP_SERVICE_ACCOUNT value from the output, and we give that service account read access to your path. Then create the file format and the stage:

CREATE FILE FORMAT wingspan_parquet TYPE = PARQUET;

CREATE STAGE wingspan_stage
  URL = 'gcs://<bucket-from-wingspan>/<your-path>/'
  STORAGE_INTEGRATION = wingspan_gcs
  FILE_FORMAT = wingspan_parquet;

Create the raw tables from Step 4, but load them with a scheduled task instead of a pipe and an S3 notification. COPY INTO skips files it has already loaded, so each run picks up only the new ones:

CREATE TASK load_payable
  WAREHOUSE = loading_wh
  SCHEDULE = '60 MINUTE'
AS
  COPY INTO raw_payable (record, file_name)
  FROM (SELECT $1, METADATA$FILENAME FROM @wingspan_stage/payable/);

ALTER TASK load_payable RESUME;

Create one task for each resource, on the same schedule as your export. The views from Step 5 work the same way. Snowflake's guide to configuring an integration for Google Cloud Storage has the details.

Watch the runs

List runs for a data export shows each run's status (Started, Completed, PartiallyCompleted or Failed), its windowStart and windowEnd, and recordCount. A failed run carries an error with the same shape as API errors. outputUrl is a signed link to the run's files, valid for 15 minutes.

To redo a Failed or PartiallyCompleted run, send Retry a data export run. It exports the same window again as a new run and keeps the original for your records.

On the Snowflake side, SYSTEM$PIPE_STATUS shows whether a pipe is running, and COPY_HISTORY lists each file it loaded or skipped.

Change, pause or stop an export

Update a data export changes the schedule, resources, destination, filters or columns. New filters and fieldSelection replace the earlier ones for the resources you send.
Pause stops scheduled runs and keeps your place. Resume picks up from there, so nothing is skipped.
Disable stops the export for good. Past runs and their files stay.