Klaviyo en Snowflake met elkaar verbinden

In dit artikel gebruiken we de term 'tabel', maar weergaven, geconcretiseerde weergaven en tabellen zijn allemaal geldige Snowflake-objecten die kunnen worden geïmporteerd. Zolang Klaviyo SELECT col1 FROM tabelnaam op het object kan uitvoeren, ben je vrij om te gebruiken wat je wilt.

De sleutelwoorden "MUST", "MUST NOT", "REQUIRED", "SHALL", "SHALL NOT", "SHOULD", "SHOULD NOT", "RECOMMENDED", "MAY" en "OPTIONAL" in dit document moeten worden geïnterpreteerd zoals beschreven in RFC 2119.

Snowflake Beheerder-instellingen

Dit gedeelte beschrijft de stappen die je in je Snowflake-omgeving moet volgen om Klaviyo toe te staan je gegevens te importeren.

1. Genereer een privésleutel door de volgende opdracht uit te voeren in je lokale apparaat:

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt

1. Genereer een openbare sleutel die verwijst naar de privésleutel door de volgende opdracht in je terminal uit te voeren:

openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

1. Kopieer het rsa_key.pub en plak in het onderstaande script om de tijdelijke aanduiding 'GENERATE_PUBLIC_KEY' te vervangen door gebruiker_rsa_public_key. Het onderstaande script werkt voor Mac-gebruikers, of je kunt rsa_key.pub openen in een IDE en de volledige inhoud van het bestand kopiëren als je dat wilt.

# Mac terminal command to write the key to your terminal and copy it to the clipboard
cat rsa_key.pub | tee /dev/tty | pbcopy

1. Voer het volgende script uit in je Snowflake-omgeving om een servicegebruiker aan te maken voor Klaviyo. Je moet beveiligingsadmin- en systeemadmin-rechten hebben om de onderstaande installatie te voltooien. Om te bekijken welke rol(len) je hebt, voer je TOELATING TOT USER uit <your_username> en zorg je ervoor dat beide rollen vermeld worden. Neem contact op met een systeembeheerder als je je rol moet laten aanpassen.
   1. Voel je vrij om alle variabelen aan het begin van het script bij te werken.
   2. Samenvattend kun je:
      1. Kies een bestaand magazijn of maak een nieuw magazijn
      2. Kies een bestaande database of maak een nieuwe aan voor de nieuwe schema's
      3. Maak twee nieuwe schema's KLAVIYO_TMP, en KLAVIYO_IMPORT_FROM_DWH
      4. Een nieuw netwerkbeleid en positieve lijst van Klaviyo IP's maken
      5. Een gebruiker en rol aanmaken voor Klaviyo
   3. Dit script is idempotent (kan veilig meerdere keren worden uitgevoerd), maar zal geen bestaande objecten met conflicterende namen overschrijven.

BEGIN;

-- create variables for user / password / role / warehouse / database. 
-- Change these to whatever you prefer.
SET role_name = 'KLAVIYO_DATA_TRANSFER_ROLE'; -- all letters must be uppercase, ex. 'KLAVIYO_DATA_TRANSFER_ROLE'
SET user_name = 'KLAVIYO_DATA_TRANSFER_USER'; -- all letters must be uppercase, ex. 'KLAVIYO_DATA_TRANSFER_USER'
SET warehouse_name = 'KLAVIYO_DATA_TRANSFER_WAREHOUSE'; -- all letters must be uppercase, ex. 'KLAVIYO_DATA_TRANSFER_WAREHOUSE'
SET database_name = 'KLAVIYO_DATABASE'; -- all letters must be uppercase, ex. 'KLAVIYO_DATABASE'. If this database doesn't exist, a new one will be created.
SET network_policy = 'KLAVIYO_DATA_TRANSFER_NETWORK_POLICY'; -- all letters must be uppercase, ex. 'KLAVIYO_NETWORK_POLICY'
SET network_rule = 'KLAVIYO_DATA_TRANSFER_NETWORK_RULE'; -- all letters must be uppercase, ex. 'KLAVIYO_NETWORK_RULE'
/* replace GENERATE_PUBLIC_KEY below with generated public key */

-- DO NOT CHANGE
SET schema_name_tmp = $database_name || '.KLAVIYO_TMP';  -- DO NOT CHANGE
SET schema_name_import = $database_name || '.KLAVIYO_IMPORT_FROM_DWH';  -- DO NOT CHANGE
SET full_network_rule_tmp = $schema_name_tmp || '.' || $network_rule; -- DO NOT CHANGE
SET full_network_rule_import = $schema_name_import || '.' || $network_rule; -- DO NOT CHANGE


-- change role to sysadmin for warehouse / database steps
USE ROLE sysadmin;

-- create a warehouse for data transfer service
CREATE WAREHOUSE IF NOT EXISTS IDENTIFIER($warehouse_name)
    warehouse_size = xsmall
    warehouse_type = standard
    auto_suspend = 60
    auto_resume = true
    initially_suspended = true;

-- create database for data transfer service
CREATE DATABASE IF NOT EXISTS IDENTIFIER($database_name);

-- create schemas for data transfer service
CREATE SCHEMA IF NOT EXISTS IDENTIFIER($schema_name_tmp);
CREATE SCHEMA IF NOT EXISTS IDENTIFIER($schema_name_import);

-- change role to securityadmin for user / role steps
USE ROLE securityadmin;

-- create network rule and policy for database
GRANT USAGE ON DATABASE IDENTIFIER($database_name) TO ROLE securityadmin;
GRANT USAGE, CREATE NETWORK RULE ON SCHEMA IDENTIFIER($schema_name_tmp) TO ROLE securityadmin;
GRANT USAGE, CREATE NETWORK RULE ON SCHEMA IDENTIFIER($schema_name_import) TO ROLE securityadmin;

-- whitelist klaviyo ip ranges, for KLAVIYO_TMP schema 
CREATE NETWORK RULE IF NOT EXISTS IDENTIFIER($full_network_rule_tmp)
    type = IPV4
    value_list = (
        '184.72.183.187/32', '52.206.71.52/32', '3.227.146.32/32', '44.198.39.11/32', '35.172.58.121/32', '3.228.37.244/32', '54.88.219.8/32', '3.214.211.176/32'
        )
    comment = 'Klaviyo IP Ranges as of April 2025';
CREATE NETWORK POLICY IF NOT EXISTS IDENTIFIER($network_policy)
    allowed_network_rule_list = ($full_network_rule_tmp);

-- whitelist klaviyo ip ranges, for KLAVIYO_IMPORT_FROM_DWH schema 
CREATE NETWORK RULE IF NOT EXISTS IDENTIFIER($full_network_rule_import)
    type = IPV4
    value_list = (
        '184.72.183.187/32', '52.206.71.52/32', '3.227.146.32/32', '44.198.39.11/32', '35.172.58.121/32', '3.228.37.244/32', '54.88.219.8/32', '3.214.211.176/32'
        )
    comment = 'Klaviyo IP Ranges as of April 2025';
CREATE NETWORK POLICY IF NOT EXISTS IDENTIFIER($network_policy)
allowed_network_rule_list = ($full_network_rule_import);


-- create role for data transfer service
CREATE ROLE IF NOT EXISTS IDENTIFIER($role_name);
GRANT ROLE IDENTIFIER($role_name) TO ROLE sysadmin;

-- create a user for data transfer service
CREATE USER IF NOT EXISTS IDENTIFIER($user_name)
    type = SERVICE
    network_policy = $network_policy
    default_role = $role_name
    default_warehouse = $warehouse_name
    rsa_public_key = 'GENERATE_PUBLIC_KEY';
GRANT ROLE IDENTIFIER($role_name) TO USER IDENTIFIER($user_name);
ALTER USER IDENTIFIER($user_name) SET NETWORK_POLICY = $network_policy;

-- grant service role access to warehouse
GRANT USAGE
    ON WAREHOUSE IDENTIFIER($warehouse_name)
    TO ROLE IDENTIFIER($role_name);

-- grant service access to database
GRANT MONITOR, USAGE
    ON DATABASE IDENTIFIER($database_name)
    TO ROLE IDENTIFIER($role_name);

-- Grant privileges for KLAVIYO_TMP 
GRANT USAGE ON SCHEMA IDENTIFIER($schema_name_tmp) TO ROLE IDENTIFIER($role_name);
GRANT MONITOR, USAGE, CREATE TABLE, CREATE VIEW, CREATE SEQUENCE, CREATE FUNCTION, CREATE PROCEDURE
    ON SCHEMA IDENTIFIER($schema_name_tmp)
    TO ROLE IDENTIFIER($role_name);
GRANT ALL ON FUTURE TABLES IN SCHEMA IDENTIFIER($schema_name_tmp) TO ROLE IDENTIFIER($role_name);

-- Grant privileges for KLAVIYO_IMPORT_FROM_DWH
GRANT USAGE ON SCHEMA IDENTIFIER($schema_name_import) TO ROLE IDENTIFIER($role_name);
GRANT SELECT
    ON FUTURE TABLES
    IN SCHEMA IDENTIFIER($schema_name_import)
    TO ROLE IDENTIFIER($role_name);

COMMIT;

Snowflake-gegevens instellen

Je hebt hierboven twee nieuwe schema's gemaakt.

• KLAVIYO_TMP wordt exclusief gebruikt door Klaviyo. Je MAG GEEN tabellen wijzigen die in dit schema zijn aangemaakt. Klaviyo verwijdert deze tabellen wanneer ze niet langer nodig zijn.
• KLAVIYO_IMPORT_FROM_DWH is de plek waar je de laatste tafels moet opslaan, zodat Klaviyo ze kan importeren. Wanneer je het proces voor het maken van synchronisaties doorloopt, kun je uit alle tabellen in dit schema kiezen. Daarom MOET je alleen de laatste tafels opslaan die je wilt importeren om verwarring tijdens de installatie te voorkomen.

Alle tabellen die je naar Klaviyo wilt importeren, moeten aan de volgende criteria voldoen.

Tijdstempelvereisten

1. Tabellen MOETEN een tijdstempelveld bevatten dat aangeeft wanneer de rij is gemaakt of bijgewerkt. Vaak wordt dit 'insert_at' of 'updated_at'. Je stelt dit voor elke tabel in tijdens het maken van de synchronisatie.
   1. Het tijdstempelveld MOET monotoon toenemen (maw het moet altijd groter worden of hetzelfde blijven, het wordt nooit kleiner).
   2. Nadat je de synchronisatie hebt aangemaakt, MAG je de tijdstempelwaarde van een rij NIET instellen op een tijd in het verleden, anders haalt Klaviyo die rij niet op.
   3. De tijdzone van dit specifieke veld is niet belangrijk voor Klaviyo, als je maar aan de bovenstaande vereisten voldoet
   4. Je tijdstempels MOETEN in UTC zijn of tijdzone-informatie bevatten. Als er tijdzone-informatie ontbreekt, gaat Klaviyo uit van UTC. Voor aangepaste eigenschappen blijven deze tijdstempels in tekenreeksindeling, zodat je ze in de tijdzone van je voorkeur kunt interpreteren.
2. Het tijdstempelveld MOET aangeven wanneer de rij is ingevoegd en moet in de buurt van de huidige datum worden geclusterd. Klaviyo synchroniseert gegevens door vensters van 1 uur te scannen vanaf de oudste tijdstempelwaarde in je tabel. Een enkele rij met een tijdstempel ver in het verleden (bijv. een record uit 2023 terwijl alle andere recent zijn) zorgt ervoor dat Klaviyo elk venster van 1 uur vanaf die datum bij elke synchronisatiecyclus herhaalt. Dit is een huidige beperking die in een aankomende release moet worden opgelost.
3. Houd rekening met rijdensiteit per tijdstempelvenster van 1 uur. Aangezien gegevens worden geladen in batches van tijdstempelperioden van 1 uur, kunnen miljoenen records in hetzelfde venster van 1 uur leiden tot trage of vastgelopen synchronisaties. Hoewel de bovenlimiet van rijdensiteit afhangt van de hoeveelheid gegevens in elke rij, is een goede regel om in gedachten te houden 100.000 rijen per tijdstempelvenster van 1 uur.
4. Klaviyo raadt aan om het tijdstempelveld in te stellen op CURRENT_TIMESTAMP() of een gelijkwaardige functie wanneer je rijen toevoegt aan de tabel van waaruit we gaan synchroniseren. Meerdere rijen kunnen dezelfde tijdstempel hebben. Zie voorbeeld hieronder.

INSERT INTO table_name AS 

SELECT ... 

    , CURRENT_TIMESTAMP() AS inserted_at

... 

Tabelstructuur

1. Tabellen MOETEN worden behandeld als alleen toevoegen (ook bekend als alleen invoegen)
   1. Als je in plaats daarvan liever de rijen bijwerkt, MOET je het tijdstempelveld bijwerken zodat Klaviyo de wijziging kan identificeren.
2. Tabellen MOETEN de volgorde van je tijdstempelkolom bevatten. Snowflake zal het clusteren en verdelen op basis van je volgorde van invoegen. Dit helpt om Klaviyo's importquery's te optimaliseren, waardoor de rekenkosten in Snowflake laag blijven

Uniekheid en consistentie van profiel

1. Je MOET ervoor zorgen dat elk profiel uit slechts één gegevensbron (tabel) wordt geïmporteerd. Klaviyo voorkomt dat dezelfde eigenschap uit verschillende tabellen wordt geselecteerd tijdens het maken van synchronisatie, waardoor deze vereiste vereenvoudigt.
2. Je MOET dezelfde profiel-ID('s) (e-mailadres, telefoonnummer, externe ID, enz.) gebruiken in al je importtabellen, om het risico op een dubbel profiel te minimaliseren.
   1. Klaviyo maakt een nieuw profiel aan als de profiel-ID die je opgeeft niet overeenkomt met een bestaand profiel binnen Klaviyo.
   2. Voorbeeld: Tabel 1 (e-mail, fav_kleur) + Tabel 2 (telefoon, verjaardag)
      1. Dit kan leiden tot 2 profielen voor dezelfde persoon als deze nog niet bestaat. Als een profiel wel bestaat, zal Klaviyo de oplossing van het profiel en de updates intern afhandelen. 
   3. Een manier om dit probleem te voorkomen is door slechts één importtabel voor je hele profiel te gebruiken.

Circulaire import-export-looppreventie

1. Je MOET zorgvuldig scenario's beheren waarin zowel import- als export-functies worden gebruikt om kringen in de import-export te voorkomen. Zorg ervoor dat je exportproces geen gegevens terugstuurt naar een tabel die zich upstream van je importtabel bevindt, aangezien Klaviyo dit scenario momenteel niet detecteert.
   1. Klaviyo heeft nog geen logica om dit scenario te detecteren.
   2. Dit zou er ongeveer zo uitzien:
      1. Bij elke exportsynchronisatiecyclus exporteert Klaviyo al je profiel
      2. Vervolgens voeg je al je geëxporteerde profiel toe aan je importtabel via een aantal transformaties.
      3. Bij elke importsynchronisatiecyclus leest Klaviyo alle profielen in je importtabel, die uiteindelijk opnieuw worden geëxporteerd
   3. Scenario's waarin dit waarschijnlijk veilig is
      1. als je alleen de exporttabel gebruikt om de rijen te beperken die aan je importtabel zijn toegevoegd
      2. Als je controleert of de exporttabel geen rijen toevoegt aan je importtabel.
   4. Wat zijn de gevolgen van een circulair import-export-loop?
      1. Dit resulteert in onnodige rekenkosten voor zowel jou als Klaviyo.

Probleemoplossing

De synchronisatie lijkt vast te zitten

Als je synchronisatie wordt uitgevoerd, maar de gegevens na enkele uren niet verschijnen in Klaviyo, of als de synchronisatie ongewoon lang duurt, dan is een tijdstempelwaarde ver in het verleden de meest waarschijnlijke oorzaak:

• Controleer je tabel op rijen met een tijdstempel die aanzienlijk ouder is dan de rest van je gegevens (bijvoorbeeld één rij uit 2023 terwijl alle andere van de afgelopen week afkomstig zijn). Zelfs een enkele uitbijterrij dwingt Klaviyo om duizenden lege vensters van 1 uur te doorlopen voordat recente gegevens worden bereikt.
• Herstel: update of verwijder rijen met tijdstempels ver in het verleden, of stel ze in op een recente waarde, voordat je de synchronisatie in- of opnieuw inschakelt. Stel voor aanvullingen alle historische rijen in op dezelfde recente tijdstempel (bijv. de huidige taakuitvoeringstijd) om het aantal vensters van 1 uur dat Klaviyo moet scannen te minimaliseren. Als er meer dan ~ 100.000 rijen zijn, stel dan de tijdstempels in in batches van ~ 100.000 rijen met een tussenruimte van ~ 100.000 rijen.

Aanbevolen instelling van Snowflake-clustersleutels

Door je Snowflake-tabel op de tijdstempelkolom te clusteren, kunnen de geïmporteerde query's van Klaviyo onnodige micropartities overslaan, waardoor zowel de synchronisatietijd als de rekenkosten van je Snowflake worden verminderd:

ALTER TABLE your_database.KLAVIYO_IMPORT_FROM_DWH.your_table CLUSTER BY (your_timestamp_column);