Snowflake

Snowflake is a fully managed cloud-based data warehouse and analytics service for storing and analyzing data.

Available integrations

To send Snowflake data to Airship, see Export Snowflake data to Airship as attributes.

You can send Airship event data to Snowflake using CSV loading through external stages. Airship delivers event data as hourly CSV files that you load into Snowflake using COPY INTO statements from an external stage. This approach fits when you want full control over load timing and tooling, or when you already have a CSV-based ingestion pipeline.

Export Snowflake data to Airship as attributes

Sending Snowflake data to Airship capitalizes on existing, readily available tools to achieve the highest degree of flexibility and control, all while maintaining cost-effectiveness. Your data team uses Snowflake to generate valuable insights and audience segments by employing a tool they are already proficient in: SQL. You can expand on these existing processes to send data to Airship. This collaboration unlocks a multitude of potential use cases and drives a substantial improvement in personalization capabilities.

Follow these steps to create attributesMetadata used for audience segmentation and personalization. Attributes extend the concept of tags by adding comparison operators and values to determine whether or not to target a user, helping you better evaluate your audience. from Snowflake data. You can also leverage any of the supported features listed in SFTP upload for CSV files, including uploaded listsAn audience list you populate by uploading a CSV of channel or named user IDs. In the API, uploaded lists are called static lists..

To use your Snowflake data directly, without copying or importing the data into Airship, see Zero-copy data integration.

Generate SFTP keys

Airship’s SFTP implementation authenticates with an SSH key pair: a private key you keep in Snowflake, and a public key you upload to Airship.

Follow these steps in your Airship project:

1. Generate your key pair. Follow the steps in Generate an SSH key pair in SFTP upload for CSV files.

2. Add your public key, making sure to set the Purpose to Attributes. Follow the steps in Add your public key to Airship in SFTP upload for CSV files. Note the host, port, and username for use in the following Snowflake steps.

Store the SFTP credentials

In Snowflake, store the private key in a SECRET type password, since this keeps the tokens encrypted:

-- create secret credential object (removed the actual key in this example)
-- this creates an object with a username and password (it's a key in this case)
CREATE SECRET sftp_airship_attribute_cred
    TYPE = password
    USERNAME = 'OZzRx6y4Rm690T6KkReIuQ'
    PASSWORD = '-----BEGIN RSA PRIVATE KEY-----
    ...
-----END RSA PRIVATE KEY-----
';

Create a network rule and integration

Create a network rule and external access integration in Snowflake to allow outbound connections to Airship’s SFTP server:

-- This gets used in creating the external access integration
CREATE OR REPLACE NETWORK RULE airship_sftp_network_rule
  TYPE = HOST_PORT
  VALUE_LIST = ('sftp.airship.com:5222') 
  MODE= EGRESS
;

-- This gets used later creating the stored proc
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION sftp_airship_ext_int
  ALLOWED_NETWORK_RULES = (airship_sftp_network_rule)
  ALLOWED_AUTHENTICATION_SECRETS = (sftp_airship_attribute_cred)
  ENABLED = true
;

Create a function to retrieve the SSH authentication

Create a function in Snowflake that retrieves the SFTP credentials from the secret you just created, so other stored procedures can use them:

CREATE OR REPLACE FUNCTION get_secret_username_password()
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = 3.9
HANDLER = 'get_secret_username_password'
EXTERNAL_ACCESS_INTEGRATIONS = (sftp_airship_ext_int)
SECRETS = ('cred' = sftp_airship_attribute_cred )
AS
$$
import _snowflake

def get_secret_username_password():
  username_password_object = _snowflake.get_username_password('cred')
  username_password_dictionary = {}
  username_password_dictionary["Username"] = username_password_object.username
  username_password_dictionary["Password"] = username_password_object.password
  return username_password_dictionary
$$;

Create a sample attributes table

Next, create a sample SQL table for your Snowflake attributes. It’s unlikely that attributes will be pulled from a single table formatted precisely like this, but it’s a good place to start to illustrate the functionality. Create the table and insert a few sample rows:

-- create a table for attributes and insert a row or two
CREATE TABLE sample_attributes (
    named_user VARCHAR(100),
    first_name VARCHAR(100),
    last_name VARCHAR(100),
    loyalty_id INT,
    favourite_color VARCHAR(100),
    tacos_desired INT
);
INSERT INTO sample_attributes VALUES ('8732eda2-c13c-4a2a-9123-fb5bf0bccffb', 'John', 'Smith', 60001, 'Green', 77);
INSERT INTO sample_attributes VALUES ('ba06daf2-66f8-43ce-a152-b0605b9b834e', 'Alice', 'Jones', 87301, 'Blue', 1);

Create a stored procedure

Create a stored procedure that retrieves your SFTP credentials from the function you created, exports the sample_attributes table as a CSV file, and uploads it to Airship over SFTP. This example is intentionally basic: it takes no arguments, and the table name is hardcoded.

CREATE OR REPLACE PROCEDURE upload_to_sftp()
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
HANDLER = 'main'
EXTERNAL_ACCESS_INTEGRATIONS = (sftp_airship_ext_int)
PACKAGES = ('snowflake-snowpark-python','paramiko')
SECRETS = ('cred' = sftp_airship_attribute_cred)
AS
$$
import _snowflake
import paramiko
from snowflake.snowpark.files import SnowflakeFile
from io import StringIO


def main(session):
    # use the function for grabbing the creds
    sftp_cred = _snowflake.get_username_password('cred');
    
    # convert the private key to a file-like object
    private_key_file = StringIO(sftp_cred.password)
    private_key = paramiko.RSAKey.from_private_key(private_key_file)
    
    # grab the data from the sample attributes table
    df = session.sql("select * from sample_attributes").toPandas()
    df.to_csv('/tmp/out.csv',index=False)
    
    # connect and upload via SFTP using Paramiko
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        ssh.connect("sftp.airship.com",port=5222, username=sftp_cred.username, pkey=private_key)
        sftp = ssh.open_sftp()
        ret = sftp.put('/tmp/out.csv','/out.csv', confirm=False)
        # confirm=False is important because Airship sftp doesn't support `ls`
        return ret
    except Exception as e:
        return f" Error with SFTP : {e}"

$$;

Verify transfer

In a Snowsight worksheet or with snowsql, run call upload_to_sftp();, then verify your data was successfully transferred to Airship: Select the Audience menu, then Attributes, then Upload History. For more information, see View upload history in Adding attributes to your project.

Targeting users

You can now use the transferred data for targeting. See Targeting your audience using attributes. You can also use the data to build a segmentA reusable audience group you create by selecting unique or shared user data..

Send Airship data to Snowflake in CSV files

This integration sends your Airship Real-Time Data StreamingA service that delivers user-level events in real time to your backend or third-party systems using the Data Streaming API. event data as batched CSV files to your cloud storage in Amazon S3, Google Cloud Storage, or Azure Blob Storage. Snowflake reads those files into tables through an external stage.

After configuring the integration, you create Snowflake tables using provided event schemas, configure your cloud storage bucket as a Snowflake external stage, and then copy the CSV files into your tables using COPY INTO statements. The event schemas help you create the appropriate Snowflake tables and ensure your data loads correctly.

This integration gives you flexibility in designing your data pipeline while maintaining full control over your data. Exports run hourly, but you can choose how frequently to copy data into Snowflake and implement the loading approach that best fits your infrastructure and requirements.

Requirements

This integration requires the following:

• A Snowflake account
• A cloud provider account with Amazon S3, Google Cloud Storage, or Azure Blob Storage enabled
• An Airship account with messaging and Real-Time Data Streaming

Configure a cloud storage provider

Create a storage location for your Airship data and configure credentials that the integration will use to upload files:

• Amazon S3 — Configure an IAM user and retrieve the account access keys. See Managing access keys  in Amazon’s IAM documentation.
• Azure — Configure an Azure connection string. See Configure Azure Storage connection strings  in Azure’s documentation.
• Google Cloud Storage — Create a JSON-formatted service account key with the Storage Object Creator role. See Create and delete service account keys  in Google’s IAM documentation.

Configure Snowflake in Airship

You will need the following information about your cloud storage:

• Amazon S3 — Access key ID, secret access key, bucket name, and AWS region
• Azure — Connection string and container name
• Google Cloud Storage — Bucket name and JSON-formatted service account key

Set up the integration in your Airship project:

1. Next to your project name, select the dropdown menu (), then Settings.
2. Under Project settings, select Partner Integrations.
3. Select Snowflake.
4. Select Configure for the outbound integration and use the onscreen instructions to complete the following:
   • Select your storage provider and enter your credentials. For Amazon S3, you must also select your region.
   • (Optional) Enable compressing your data to save on storage space.
   • (Optional for Amazon S3) Enable encrypting your data using server-side encryption .
   • Select the Airship events to send to your storage location.

After completing configuration, it may take several minutes to begin populating events.

Configure a Snowflake external stage

Create an external stage in Snowflake that points at your cloud storage so Snowflake can read your exported Airship data. See CREATE STAGE  in Snowflake’s SQL reference for setup steps.

Create Airship tables in Snowflake

Copy the SQL DDL script from the Snowflake event schemas reference page and run it in a Snowsight worksheet or with snowsql -f <file>.sql. The script creates an AIRSHIP_EVENTS schema and one table per event type, with column definitions that match the CSV files exported by the Snowflake integration.

When Airship adds new event types or new columns to existing events later, re-run the corresponding CREATE TABLE or ALTER TABLE statements to keep your tables in sync.

Load data into Snowflake tables

After Airship exports data to your external stage, use Snowflake’s COPY INTO statement to load the CSV files into your Airship event tables. You control which event types to load and how frequently, and you can make this a step in your existing data pipeline.

Airship generates one CSV file per event type per hour when events occur. Each file includes a header row and follows this path pattern:

appKey/cloudProvider/integrationId/EVENT_TYPE/YYYY_MM_DD/YYYY_MM_DD_HH_mm_ss.csv

"id","offset","occurred","processed","app_key","channel","device_type","named_user","custom_identifiers","locale_variant","locale_country_code","locale_timezone","locale_language_code","iana_timezone","app_version","device_model","ua_sdk_version","push_opt_in","device_os","location_enabled","location_permission","background_push_enabled","web_browser_name","web_browser_type","web_browser_version","web_user_agent_string","open_platform_name","alerting","campaigns","push_id","group_id","variant_id"
"f5e05251-b128-11ec-8a12-0242eb00e5a9","1000042149690","2022-03-31 19:30:02.357 +00:00","2022-03-31 19:30:14.867 +00:00","YOUR_APP_KEY","b1ab9a9e-9634-4fb4-875a-9a02dcd68e66","ANDROID","","","","FR","7200","fr","Europe/Paris","2022-02-09T000134-goat","VOG-L29","16.3.0","true","10","false","NOT_ALLOWED","true","","","","","","true","","f5d2bdc0-b128-11ec-bae4-024205acadbe","ac4058f7-981b-4f0e-b9c4-8b7af3a647da",""
"f8fb4df1-b128-11ec-a0e3-0242e24c72c5","1000042149691","2022-03-31 19:30:07.567 +00:00","2022-03-31 19:30:16.760 +00:00","YOUR_APP_KEY","b1ab9a9e-9634-4fb4-875a-9a02dcd68e66","ANDROID","","","","FR","7200","fr","Europe/Paris","2022-02-09T000134-goat","VOG-L29","16.3.0","true","10","false","NOT_ALLOWED","true","","","","","","true","","f6586880-b128-11ec-99dd-02423b3f5d45","879056c8-0b34-4c8c-8cdc-1df4f3d40b40",""
"9aa98182-e854-4fa7-9c9f-ddea2082cc4c","1000042149694","2022-03-31 19:33:51.225 +00:00","2022-03-31 19:33:51.416 +00:00","YOUR_APP_KEY","62d92a24-0ced-40d1-ad1d-e1ea953189b7","SMS","","{""sender"":""17372004196""}","","","","","","","","","","","","","","","","","","","true","","75fdde30-b129-11ec-99dd-02423b3f5d45","0e2a3d4b-b4b4-4208-838f-08eb7333aa5e",""

Define a reusable file format that skips the header row and handles the other common CSV parsing cases:

create or replace file format airship_csv_events
  type = csv
  skip_header = 1
  timestamp_format = 'AUTO'
  null_if = ''
  field_optionally_enclosed_by = '"'
  error_on_column_count_mismatch = false
  field_delimiter = ',';

Reference the file format in COPY INTO and scope the pattern to a single event type’s CSV path in your stage:

use "DB_NAME"."AIRSHIP_EVENTS";

copy into "TAG_CHANGE"
  from @sample_airship_stage
  file_format = airship_csv_events
  pattern = 'YOUR_APP_KEY/YOUR_CLOUD_PROVIDER/YOUR_INTEGRATION_ID/TAG_CHANGE/2022_03_31/.*[.]csv';