9

Data Warehouses: Getting Started

Our data warehouse integrations let you send Customer.io data about messages, people, metrics, etc to your data warehouse by way of an Amazon S3 or Google Cloud Project (GCP) storage bucket. From there, you can ingest the data to your warehouse.

You can find our data warehouse integrations by going to Data & Integrations > Integrations and selecting the Databases option. From here, you can select your data warehouse or storage bucket. If your data warehouse appears twice in this list, pick the Data out integration. Check out the specific documentation for your data warehouse for help setting up your integration.

How it works

This integration exports individual parquet files for Deliveries, Metrics, Subjects, Outputs, Content, People, and Attributes to your storage bucket. Each parquet file contains data that changed since the last export.

Once the parquet files are in your storage bucket, you can import them into data platforms like Fivetran or data warehouses like Redshift, BigQuery, and Snowflake.

Note that this integration only publishes parquet files to your storage bucket. You must set your data warehouse to ingest this data. There are many approaches to ingesting data, but it typically requires a COPY command to load the parquet files from your bucket. After you load parquet files, you should set them to expire to delete them automatically.

We attempt to export parquet files every 15 minutes, though actual sync intervals and processing times may vary. When syncing large data sets, or Customer.io experiences a high volume of concurrent sync operations, it can take up to several hours to process and export data. This feature is not intended to sync data in real time.

The initial export vs incremental exports

Your initial sync is a set of files containing historical data to represent your workspace’s current state. Subsequent sync files contain changesets.

• Metrics: The initial metrics sync is broken up into files with two sequence numbers, as follows. <name>_v4_<workspace_id>_<sequence1>_<sequence2>.
• Attributes: The initial Attributes sync includes a list of profiles and their current attributes. Subsequent files will only contain attribute changes, with one change per row.

When you set up your sync, you can choose the parquet files that you want to export to your storage bucket. If you temporarily disable a file, and then turn it back on, the next parquet file will contain the changeset between when you disabled the file and when you enabled it.

For example, let’s say you’ve enabled the Attributes export. We will attempt to sync your data to your storage bucket every 15 minutes:

1. 12:00pm We sync your Attributes Schema for the first time. This includes a list of profiles and their current attributes.
2. 12:05pm User1’s email is updated to company-email@example.com.
3. 12:10pm User1’s email is updated to personal-email@example.com.
4. 12:15 We sync your data again. In this export, you would only see attribute changes, with one change per row. User1 would have one row dedicated to his email changing.

How do I get data into my data warehouse?

There are many approaches to ingesting data from your storage bucket, but here’s an example moving data from from a Google Cloud Storage bucket to Google BigQuery.

• Implement a Cloud Function to automatically import the parquet files from your GCS bucket to a BigQuery table.
• Set an expiration on the parquet files so they’re automatically deleted.

Below is a screenshot of an example Cloud Function. You can download sample code for this cloud function—based on our v3 schema—below here:

Our sample does not work for v4 schema, and it won’t work out of the box if you use a subfolder in the sync option. Make sure you review the code and making appropriate modifications to fit your use case.

Exported parquet files

This section describes the different kinds of files you can export from our Database-out integrations. Many schemas include an internal_customer_id—this is the cio_idAn identifier for a person that is automatically generated by Customer.io and cannot be changed. This identifier provides a complete, unbroken record of a person across changes to their other identifiers (id, email, etc).. You can use it to resolve a person associated with a subject, delivery, etc.

Deliveries schema

Deliveries are individual email, in-app, push, SMS, slack, and webhook records sent from your workspace. The first deliveries export file includes baseline historical data. Subsequent files contain rows for data that changed since the last export.

Delivery Content schema

The delivery_content schema represents message contents; each row corresponds to an individual delivery. Use the delivery_id to find more information about the contents of a message, or the recipient to find information about the person who received the message.

If your delivery was produced from a campaign, it’ll include campaign and action IDs, and the newsletter and content IDs will be null. If your delivery came from a newsletter, the row will include newsletter and content IDs, and the campaign and action IDs will be null.

Delivery content might lag behind other tables by 15-30 minutes (or roughly 1 sync operation). We package delivery contents on a 15 minute interval, and can export to your data warehouse up to every 15 minutes. If these operations don’t line up, we might occasionally export delivery_content after other tables.

Metrics schema

Metrics exports detail events relating to deliveries (e.g. messages sent, opened, etc). Your initial metrics export contains baseline historical data, broken up into files with two sequence numbers, as follows: <name>_v4_<workspace_id>_<sequence1>_sequence2>.

Subsequent files contain rows for data that changed since the last export.

Subjects schema

Subjects are the unique workflow journeys that people take through Campaigns and API Triggered Broadcasts. The first subjects export file includes baseline historical data. Subsequent files contain rows for data that changed since the last export.

Outputs schema

Outputs are the unique steps within each workflow journey. The first outputs file includes historical data. Subsequent files contain rows for data that changed since the last export.

People schema

The first People export file includes a list of current people at the time of your first sync (deleted or suppressed people are not included in the first file). Subsequent exports include people who were created, deleted, or suppressed since the last export.

People exports come in two different files:

• people_v4_<env>_<seq>.parquet: Contains new people.
• people_v4_chngs_<env>_<seq>.parquet: Contains changes to people since the previous sync.

These files have an identical structure and a part of the same data set. You should import them to the same table.

Attributes schema

Attribute exports represent changes to people (by way of their attribute values) over time. The initial Attributes export includes a list of profiles and their current attributes. Subsequent files contain attribute changes, with one change per row.

Changelog

v4

If you’re on a prior schema, your Objects and Subjects tables were disabled on October 31, 2022. You’ll need to upgrade to continue receiving subjects and outputs.

Version 4 includes the following changes from v3:

• The Subjects and Outputs tables have been disabled for previous versions. You must update to v4 to use Subjects and Outputs.
• Added a Delivery Content table, similar to the Deliveries table but this new table includes the actual message content for each delivery. You can associate content with a delivery by delivery_id.
• The Outputs schema now includes a new subject_name column. This is the same column from the Subjects table.
• Removed subject_id from the Outputs table.

Upgrade to v4

1. In your Customer.io workspace, go to Data & Integrations > Integrations, and click your data warehouse integration.
2. Click Upgrade Sync. We’ll automatically disable the Subjects and Outputs schemas, so you can update your database.
3. In your database, add a string-type subject_name column to the Outputs table.
4. Query the Subjects table to populate the new outputs.subject_name column.
5. Drop the subject_id column from the Subjects and Outputs tables in your database.
6. (Optional) Create a Delivery Content table based on the new delivery_content schema
7. Return to Customer.io and re-enable the Subjects, Outputs, and/or Delivery Content schemas.

v3

Version 3 includes the following changes from v2:

• A seq_num column in the Deliveries, Subjects, Outputs, and Metrics tables. This is a constantly increasing value, where a larger value indicates a more current record.
• People tables now contain record updates, rather than only the record as first created. For example, we produce a new record if you change a person’s email, delete them, or suppress them.
• The People table now has an updated_at column. Because many data warehouses don’t replace rows when adding duplicate primary keys, you can select the most recent updated_at value for each profile.
• The People table now contains an email_addr column.

v2

Data warehouse sync v2 includes the following changes from v1:

• Support for transactional messages (as transactional_message_id) in the Deliveries schema.
• A fix for an issue that caused missing rows in Subjects and Outputs data. As a result of this bug, data warehouse v1 no longer supports Subjects or Outputs data.

If you used our initial data warehouse release, we recommend updating to the v2 implementation. However, you can continue using our original data warehouse sync feature if you don’t use:

• Subjects and/or Outputs data
• Transactional messages

Frequently asked questions

How are exported parquet files organized?

Each parquet file is named <name>_v<x>_<workspace_id>_<sequence>.

• <name> is either deliveries, metrics, subjects, outputs, or people.
• v Indicates the schema version. The current version is v2; v1 schemas do not have a version indicator and are deprecated.
• <workspace_id> refers to the Customer.io workspace whose data is included.
• <sequence> is an ever-increasing value over time.

How do you handle export failures?

If we experience an internal failure, we monitor and repair it promptly. If there’s an external failure such as a networking problem, we retry the export. If there’s an issue with bucket access, we’ll reach out to you with details.

In each retry case, the next export will contain all data since the prior export. No data will be lost.

How should I import data from my bucket to my data warehouse?

There are many approaches to this, but here’s one example we’ve seen work for moving data from a GCS bucket to BigQuery.

• Implement a Cloud Function to automatically import the parquet files from your GCS bucket to a BigQuery table.
• Set an expiration on the parquet files so they get automatically deleted.

Below is a screenshot of an example Cloud Function. You can download sample code for this cloud function—based on our v4 schemas—below here:

How can I get information about a campaign, workflow action, or message that’s not included in the export?

With the data in our export, it’s possible to query our API to pull any extra information you need.

For example, calling https://api.customer.io/v1/campaigns/:id with the id of the campaign from a subject record will give you details about the campaign.

Check out our full API docs here.

How do I get all of the current attributes for a profile? On setup, the initial Attributes export will include a list of profiles and their current attributes. Subsequent files will only contain attribute changes, with one change per row. In order to get the most recent attributes for a particular profile, you’ll need to use the timestamp value to query for the latest.

An example query for Snowflake would be:

select internal_customer_id, attribute_name, attribute_value 
from (select internal_customer_id, attribute_name, attribute_value, row_number() over (partition by internal_customer_id,attribute_name order by timestamp desc) RNO from attributes) 
where rno=1 and internal_customer_id='xxxxxxxxxxx';