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.

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) Ingestion 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 a bit 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

Creating a JSON Object:

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

Converting timestamp column to milliseconds:

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

Converting milliseconds to 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.

TO_TIMESTAMP_NTZ(TIME_COLUMN_IN_MILLIS, 3) as "update_time_column"

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

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