Snowflake Data Import

With Amplitude's Snowflake integration, you can ingest Snowflake data directly into your Amplitude project. The integration supports four strategies to import your Snowflake data, depending on the data types you select.

Limits

• Maximum running time for a single Snowflake SQL query is 12 hours.
• Max batch size for a single Snowflake SQL query is 1B events.

Add and configure the Snowflake source

Complete the following steps to configure the Snowflake source:

1. Set up and verify the connection
2. Select data
3. Select the import strategy
4. Map your data
5. Schedule your sync

Set up and verify the connection

To add Snowflake as a data source in your Amplitude project, follow these steps:

1. In Amplitude Data, navigate to Catalog → Sources.

2. In the Warehouse Sources section, click Snowflake.

3. Enter the required credentials for the Snowflake instance you want to connect:

   • Account: Snowflake account name. Case sensitive. This is the first part of your Snowflake URL, before snowflakecomputing.com. Don't include ".snowflakecomputing.com" in your account name.
   • Database: Name of the database where Amplitude can find the data.
   • Warehouse: Used by Amplitude to execute SQL.
   • Username: Used by Amplitude for authentication.
   • Password: Used by Amplitude for authentication.

   Amplitude offers password-based and key pair authentication for Snowflake.

   • If you want to use password authentication, select Password and enter your password in the Password field.
   • If you want to use key pair authentication, select Key pair and then click Generate Key. Then provide the organization and account names in the format ORGNAME-ACCOUNTNAME.

4. Copy the autogenerated SQL query and run it in Snowflake to give Amplitude the proper permissions.

5. After running the query, click Next to test the connection.

6. After the test succeeds, click Next again to move on to the data selection stage.

Select the data type

The data type you select defines the strategies and settings available to you for configuration.

Select the import strategy

Select from the following strategies, depending on your data type selection.

See the following table to understand which data types are compatible with which import strategies.

Map your data

Depending on the import strategy you choose, either map your data with a SQL statement to transform the data (Timestamp, Full Sync) or use the data selection tool to map column names directly to Amplitude properties.

Schedule your sync

Provide a name for the source, and set the frequency with which Amplitude imports your data.

Choose the best integration for your use case

When choosing an integration strategy, consider the following:

• Full Sync: Choose this option if you need to periodically ingest the entire dataset and can't track which rows have changed. This method is best for smaller datasets where tracking incrementally isn't possible. This method isn't suitable for large datasets due to the overhead required to ingest all data each time.

• Timestamp Import: Choose this option if you can incrementally import data using a monotonically increasing timestamp column that indicates when records when Snowflake loads the records. This is efficient and works well when you append new data with timestamps.

• Change Data Capture (CDC) Append Only: Choose this option to import data based on changes detected by Snowflake's CDC feature while still using Amplitude's enrichment services. This method only supports reading INSERT operations from the CDC

• Change Data Capture (CDC) Continuous Sync: Choose this option to directly mirror the data in Snowflake with INSERT, UPDATE, and DELETE operations based on changes detected by Snowflake's CDC feature. This method disables Amplitude's enrichment services to remain in sync with your source of truth and is ideal when you need to keep Amplitude data fully synchronized with your Snowflake data. UPDATE and DELETE operations mutate data in Amplitude.

Prerequisites and considerations for CDC

When using CDC Continuous Sync, keep the following things in mind:

• Enable Change Tracking: Enable change tracking for the source table or view. See Enabling Change Tracking on Views and Underlying Tables in Snowflake's documentation.

• Data Retention Settings: DATA_RETENTION_TIME_IN_DAYS must be greater than or equal to one, but Amplitude recommends at least seven days. Otherwise, the change-based import fails. For more details, see Time Travel in Snowflake's documentation. Setting DATA_RETENTION_TIME_IN_DAYS to 0 disables the change tracking and renders the connection unrecoverable. If this happens, recreate the source.

• Disable Change Tracking: If you disable change tracking in Snowflake, or disconnect the Amplitude source for a period longer than the value of DATA_RETENTION_TIME_IN_DAYS, Amplitude loses the ability to track historical changes. In this case, recreate the connection. To avoid duplicate events, ensure all events have an insert_id set, and recreate the connection within seven days.

• Unique and Immutable insert_id: Ensure the data to be imported has a unique and immutable insert_id for each row to prevent data duplication if there are any unexpected issues. More about Amplitude deduplication and insert_id is available in Event Deduplication.

• Complex SQL Statements: If a data source is represented as a complex SQL SELECT statement (for instance, with a JOIN clause), create a VIEW in your Snowflake account that wraps the data source to use it with a change-based import strategy. See Streams on Views for considerations when using CDC with views in Snowflake.

• Views with JOINs: While Snowflake CDC is efficient, using views that contain JOINs can have performance implications. Consider syncing joined data as user profiles instead.

• Avoid table deletion and re-creation: Don't delete and recreate tables with the same name, as Snowflake CDC doesn't capture changes in this scenario. Use incremental models with tools like dbt to prevent table replacement.

• Handling schema changes: CDC supports adding new columns with default NULL values to CDC-tracked tables or views. Amplitude recommends against other kinds of schema changes. Snowflake CDC only reflects changes from DML statements. DDL statements that logically modify data (such as adding new columns with default values, dropping existing columns, or renaming columns) affect future data sent to Amplitude, but Snowflake doesn't update historical data with changes caused by DDL statements. As a result, Amplitude doesn't reflect these updates for historical data.

• Amplitude enrichment services disabled: When using CDC Continuous Sync, Amplitude disables enrichment services like ID resolution, property and attribution syncing, and resolving location info to remain in sync with your source of truth.

Migrate to Change Data Capture (CDC) Continuous Sync

Amplitude recommends that you create a new project to test sending and mutating data. When you confirm that data is mapped and mutated correctly, complete the following steps in your main project:

1. Modify your existing connection to have a filtering definition like WHERE time < {cutOffDate}, where time is the event time and cutOffDate is tomorrow in milliseconds since epoch.
2. Wait until the cutOffDate you set in the previous step.
3. Verify that no new data flows in with the existing source connection.
4. Create a new source with a filtering definition like WHERE time >= {cutOffDate}, where time is event time and cutOffDate is tomorrow in milliseconds since epoch.
5. Delete the source connection you modified in step 1.

Data fields

Include the mandatory fields for the data type when you create the SQL query. These tables outline the mandatory and optional fields for each data type. Find a list of other supported fields for events in the HTTP V2 API documentation and for user properties in the Identify API documentation. Add any columns not in those lists to either event_properties or user_properties, otherwise it's ignored.

Events

Find other supported fields can in the HTTP V2 API documentation.

User properties

Find other supported fields in the Identify API documentation.

Group properties

Each group property in group_properties would apply to every group in groups.

To use a group property:

• Set group properties. The following is an example of how you can do it in Snowflake Group Property Import:

  1SELECT OBJECT_CONSTRUCT('customerId', account_id)                      AS "groups",          -- must be JSON
  2      OBJECT_CONSTRUCT('companyName', name, 'customerType', type) AS "group_properties" -- must be JSON
  3FROM "AMPLITUDE"."DWH"."ACCOUNTS"

• Send events with group properties associated. These can be dummy events, so long as the user ID and groups are there. Specify the following in your Snowflake Event Import:

  1"groups": {"customerId": <account_id>}

SQL query examples

To make the data selection step easier, here are few example SQL snippets to get you started.

Event data example

1SELECT
2    EVENT_TYPE_COLUMN AS "event_type",
3    EVENT_PROPERTIES_VARIANT_COLUMN AS "event_properties",
4    TIME_EPOCH_MS_COLUMN AS "time",
5    USER_ID_COLUMN AS "user_id",
6    USER_PROPERTIES_VARIANT_COLUMN AS "user_properties"
7FROM DATABASE_NAME.SCHEMA_NAME.TABLE_OR_VIEW_NAME

User property example

1SELECT
2    USER_ID_COLUMN AS "user_id",
3    USER_PROPERTIES_VARIANT_COLUMN AS "user_properties"
4FROM DATABASE_NAME.SCHEMA_NAME.TABLE_OR_VIEW_NAME

Group property example

1SELECT
2    GROUPS_OBJ AS "groups",
3    GROUP_PROPS_OBJ AS "group_properties"
4FROM DATABASE_NAME.SCHEMA_NAME.TABLE_OR_VIEW_NAME

Common snippets

Create a JSON Object:

1OBJECT_CONSTRUCT('city', CITY, 'state', STATE) as "user_properties"

Convert a timestamp column to milliseconds:

1DATE_PART('EPOCH_MILLISECOND', TIMESTAMP_COLUMN) as "time"

Convert milliseconds to the TIMESTAMP_NTZ format needed for time-based import. This example uses the scale argument set to 3 to convert to milliseconds. See the Snowflake documentation for more details.

1TO_TIMESTAMP_NTZ(TIME_COLUMN_IN_MILLIS, 3) as "update_time_column"

Convert a timestamp column with a timezone to TIMESTAMP_NTZ format needed for time-based import.

1TO_TIMESTAMP_NTZ(CONVERT_TIMEZONE('UTC', TIMESTAMP_TZ_COLUMN)) as "update_time_column"

SQL troubleshooting

The following sections provide example SQL queries you can use to configure your import connectors.

Required event properties

The Snowflake SQL queries you write for Amplitude's data warehouse import connectors must return specific columns that match Amplitude's Event API schema. Use the following examples to help structure your query.

1SELECT
2    event_type,             -- String: Name of the event
3    user_id,                -- String: Unique identifier for the user
4    EXTRACT(EPOCH_MILLISECOND FROM event_timestamp) as time  -- Timestamp in milliseconds
5FROM your_events_table

 1SELECT
 2    event_name as event_type,
 3    user_identifier as user_id,
 4    EXTRACT(EPOCH_MILLISECOND FROM event_timestamp) as time,
 5    device_id,
 6    session_id,
 7 
 8    -- Event Properties as JSON object using OBJECT_CONSTRUCT
 9    OBJECT_CONSTRUCT(
10        'property1', property1_value,
11        'property2', property2_value,
12        'category', category,
13        'value', amount
14    ) as event_properties,
15 ...
16    -- User Properties as JSON object
17    OBJECT_CONSTRUCT(
18        'user_type', user_type,
19        'subscription_status', subscription_status,
20        'city', data:address:city::string,
21        'last_updated', TO_VARCHAR(last_updated)
22    ) as user_properties,
23 
24    app_version,
25    platform,
26    os_name,
27    os_version,
28    device_brand,
29    device_manufacturer,
30    device_model,
31    carrier,
32    country,
33    region,
34    city,
35    dma,
36    language,
37    price::FLOAT as price,
38    quantity::INTEGER as quantity,
39    revenue::FLOAT as revenue,
40    product_id as productId,
41    revenue_type as revenueType,
42    location_lat::FLOAT as location_lat,
43    location_lng::FLOAT as location_lng,
44    ip
45 
46FROM your_events_table
47WHERE event_timestamp >= DATEADD(day, -7, CURRENT_DATE())

Basic event query with properties

 1SELECT
 2    event_name as event_type,
 3    user_id,
 4    EXTRACT(EPOCH_MILLISECOND FROM event_timestamp) as time,
 5    device_id,
 6    -- Construct event properties from multiple columns
 7    OBJECT_CONSTRUCT(
 8        'page_name', page_name,
 9        'button_id', button_id,
10        'interaction_type', interaction_type,
11        'duration_ms', duration_ms
12    ) as event_properties,
13    -- Construct user properties
14    OBJECT_CONSTRUCT(
15        'account_type', account_type,
16        'subscription_tier', subscription_tier,
17        'last_login', TO_VARCHAR(last_login_date)
18    ) as user_properties,
19    platform,
20    app_version
21FROM app_events
22WHERE event_timestamp >= DATEADD(day, -7, CURRENT_DATE())

Snowflake-specific features and best practices

The following are examples of Snowflake-specific features and best practices.

Working with JSON

 1-- Combining multiple JSON objects
 2SELECT
 3    event_type,
 4    user_id,
 5    EXTRACT(EPOCH_MILLISECOND FROM event_timestamp) as time,
 6    OBJECT_CONSTRUCT(
 7        'base_properties', base_properties,  -- existing JSON column
 8        'additional_data', OBJECT_CONSTRUCT(
 9            'new_field1', value1,
10            'new_field2', value2
11        )
12    ) as event_properties
13FROM events
14 
15-- Parsing JSON fields
16SELECT
17    event_type,
18    user_id,
19    time,
20    PARSE_JSON(raw_properties):field_name::string as extracted_value
21FROM events

Handling timestamps

 1-- Converting different timestamp formats
 2SELECT
 3    event_type,
 4    user_id,
 5    CASE
 6        WHEN TRY_TO_TIMESTAMP(timestamp_string) IS NOT NULL
 7            THEN EXTRACT(EPOCH_MILLISECOND FROM TRY_TO_TIMESTAMP(timestamp_string))
 8        WHEN TRY_TO_TIMESTAMP_NTZ(timestamp_string) IS NOT NULL
 9            THEN EXTRACT(EPOCH_MILLISECOND FROM TRY_TO_TIMESTAMP_NTZ(timestamp_string))
10        ELSE NULL
11    END as time
12FROM events

Data validation queries

 1-- Validate required fields
 2SELECT COUNT(*)
 3FROM (
 4    YOUR_QUERY_HERE
 5) t
 6WHERE event_type IS NULL
 7    OR user_id IS NULL
 8    OR time IS NULL;
 9 
10-- Validate JSON structure
11SELECT COUNT(*)
12FROM (
13    YOUR_QUERY_HERE
14) t
15WHERE NOT (
16    TRY_CAST(event_properties AS OBJECT) IS NOT NULL
17    AND TRY_CAST(user_properties AS OBJECT) IS NOT NULL
18);
19 
20-- Validate timestamp range
21SELECT
22    MIN(time) as min_time,
23    MAX(time) as max_time,
24    TIMEADD(millisecond, MIN(time), '1970-01-01'::timestamp) as min_readable_time,
25    TIMEADD(millisecond, MAX(time), '1970-01-01'::timestamp) as max_readable_time
26FROM (
27    YOUR_QUERY_HERE
28) t;

Performance optimization tips

Use the following examples to help optimize the performance of your integration.

Use clustering keys

Use the appropriate clusting keys on your source tables.

1ALTER TABLE your_events_table CLUSTER BY (event_timestamp, user_id);

Use materialized views

Use materialized views for complex transformations.

1CREATE MATERIALIZED VIEW amplitude_ready_events AS
2SELECT
3    -- Your transformed columns here
4FROM source_events;

Date partitioning in WHERE clauses

1WHERE event_timestamp >= DATEADD(day, -7, CURRENT_DATE())
2  AND event_timestamp < CURRENT_DATE()

Micro-partitions

1SELECT ...
2FROM your_table
3WHERE TO_DATE(event_timestamp) BETWEEN '2024-01-01' AND '2024-01-31'