Klaviyo und Snowflake verbinden

Für die Zwecke dieses Artikels verwenden wir den Begriff „Tabelle“, aber Ansichten, materialisierte Ansichten und Tabellen sind alle gültige Snowflake-Objekte, die importiert werden können. Solange Klaviyo SELECT col1 FROM table_name für das Objekt ausführen kann, kannst du verwenden, was du bevorzugst.

Die Schlüsselwörter „MUST“, „MUST NOT“, „REQUIRED“, „SHALL“, „SHALL NOT“, „SHOULD“, „SHOULD NOT“, „RECOMMENDED“, „MAY“ und „OPTIONAL“ in diesem Dokument sind so auszulegen, wie in RFC 2119 beschrieben.

Snowflake Administrator – Einrichtung

In diesem Abschnitt werden die Schritte beschrieben, die du in deiner Snowflake-Umgebung ausführen musst, damit Klaviyo deine Daten importieren kann.

1. Generiere einen privaten Schlüssel, indem du den folgenden Befehl in deinem lokalen Terminal ausführst:

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

1. Generiere einen öffentlichen Schlüssel, der auf den privaten Schlüssel verweist, indem du den folgenden Befehl in deinem Terminal ausführst:

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

1. Kopiere die Datei rsa_key.pub und füge sie in das folgende Skript ein, um den Platzhalter „GENERATE_PUBLIC_KEY“ für „user_rsa_public_key“ zu ersetzen. Das folgende Skript funktioniert für Mac-Benutzer*innen. Du kannst aber auch rsa_key.pub in einer IDE öffnen und den vollständigen Inhalt der Datei kopieren, wenn du möchtest.

# 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. Führe das folgende Skript in deiner Snowflake-Umgebung aus, um einen Dienstbenutzer für Klaviyo zu erstellen. Du benötigst die Berechtigungen securityAdmin und sysadmin, um die unten stehende Einrichtung abzuschließen. Um zu überprüfen, welche Rolle(n) du hast, führe SHOW GRANTS TO USER <your_username> aus und stelle sicher, dass du beide Rollen aufgelistet hast. Wende dich an einen Systemadministrator, wenn du deine Rolle anpassen lassen musst.
   1. Du kannst die am Anfang des Skripts eingestellten Variablen jederzeit aktualisieren.
   2. Zusammenfassend kannst du Folgendes tun:
      1. Wähle ein vorhandenes Warehouse oder erstelle ein neues
      2. Wähle eine vorhandene Datenbank oder erstelle eine neue, um die neuen Schemata aufzunehmen
      3. Erstelle zwei neue Schemas KLAVIYO_TMP und KLAVIYO_IMPORT_FROM_DWH
      4. Erstelle eine neue Netzwerkrichtlinie und lasse Listen-Klaviyo-IPs zu.
      5. Erstelle eine Benutzerrolle für Klaviyo
   3. Dieses Skript ist idempotent (kann sicher mehrmals ausgeführt werden), überschreibt aber keine vorhandenen Objekte mit widersprüchlichen Namen.

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-Daten einrichten

Oben hast du zwei neue Schemata erstellt.

• KLAVIYO_TMP wird ausschließlich von Klaviyo verwendet. Du darfst KEINE Tabellen ändern, die in diesem Schema erstellt wurden. Klaviyo löscht diese Tabellen, wenn sie nicht mehr benötigt werden.
• Klaviyo_IMPORT_FROM_DWH ist der Ort, an dem du deine endgültigen Tabellen für den Klaviyo-Import speichern solltest. Wenn du den Synchronisierungsprozess durchläufst, werden alle Tabellen in diesem Schema aufgelistet, aus denen du wählen kannst. Daher SOLLTEST du nur die endgültigen Tabellen speichern, die du importieren möchtest, um Verwirrung bei der Einrichtung zu vermeiden.

Alle Tabellen, die du in Klaviyo importieren möchtest, müssen die folgenden Kriterien erfüllen.

Anforderungen an Zeitstempel

1. Tabellen MÜSSEN ein Zeitstempelfeld enthalten, das anzeigt, wann die Zeile erstellt oder aktualisiert wurde. Oft wird dies insert_at oder updated_at sein. Du legst dies für jede Tabelle während der Erstellung der Synchronisierung fest.
   1. Das Zeitstempelfeld MUSS monoton größer werden (d. h. es muss immer größer werden oder gleich bleiben, nie kleiner werden).
   2. Nach der Synchronisierungserstellung DARF der Zeitstempelwert einer Zeile NICHT auf einen früheren Zeitpunkt festgelegt werden. Andernfalls wird diese Zeile von Klaviyo möglicherweise nicht übernommen.
   3. Die Zeitzone dieses speziellen Felds ist für Klaviyo nicht wichtig, solange du die oben genannten Anforderungen erfüllst
   4. Deine Zeitstempel MÜSSEN in UTC angegeben sein oder Zeitzoneninformationen enthalten. Wenn Informationen zur Zeitzone fehlen, übernimmt Klaviyo UTC. Bei benutzerdefinierten Eigenschaften bleiben diese Zeitstempel im String-Format, sodass du sie in deiner bevorzugten Zeitzone interpretieren kannst.
2. Das Zeitstempelfeld MUSS widerspiegeln, wann die Zeile eingefügt wurde, und sollte in der Nähe des aktuellen Datums gruppiert werden. Klaviyo synchronisiert Daten, indem es 1-Stunden-Fenster ab dem ältesten Zeitstempelwert in deiner Tabelle scannt. Eine einzige Zeile mit einem Zeitstempel, der weit in der Vergangenheit liegt (z. B. ein Datensatz für 2023, wenn alle anderen neu sind), führt dazu, dass Klaviyo von diesem Datum an bei jedem Synchronisierungszyklus alle 1 Stunde durchläuft. Dies ist eine aktuelle Einschränkung, die in einer bevorstehenden Version behoben werden sollte.
3. Ziehe die Zeilendichte pro Zeitstempelfenster von einer Stunde in Betracht. Da die Daten in Stapeln von Zeitstempelfenstern von einer Stunde geladen werden, können Millionen von Datensätzen im selben Zeitfenster von einer Stunde zu langsamen oder verzögerten Synchronisierungen führen. Während die obere Grenze der Zeilendichte von der Datenmenge in jeder Zeile abhängt, ist eine gute Faustregel, die du bedenken solltest, 100.000 Zeilen pro 1 Stunde Zeitstempelfenster.
4. Klaviyo empfiehlt dir, das Zeitstempelfeld mit CURRENT_TIMESTAMP() oder einer entsprechenden Funktion festzulegen, wenn du Zeilen zu der Tabelle hinzufügst, aus der wir die Synchronisierung durchführen. Mehrere Zeilen können denselben Zeitstempel haben. Siehe Beispiel unten.

INSERT INTO table_name AS 

SELECT ... 

    , CURRENT_TIMESTAMP() AS inserted_at

... 

Tabellenstruktur

1. Tabellen SOLLTEN als reine Anhängen-Tabellen behandelt werden (auch „nur einfügen“ genannt).
   1. Wenn du stattdessen die Zeilen aktualisieren möchtest, MUSST du das Zeitstempelfeld aktualisieren, damit Klaviyo die Änderung identifizieren kann.
2. Die Tabellen SOLLTEN in deiner Zeitstempelspalte angeordnet werden. Snowflake übernimmt das Clustering und die Partitionierung basierend auf deiner Einfügereihenfolge. Dies trägt zur Optimierung der Importabfragen von Klaviyo bei und hält die Computerkosten in Snowflake niedrig

Eindeutigkeit und Konsistenz von Profilen

1. Du MUSST sicherstellen, dass jede Profileigenschaft aus nur einer Datenquelle (Tabelle) importiert wird. Klaviyo verhindert, dass dieselbe Eigenschaft während der Synchronisierungserstellung aus verschiedenen Tabellen ausgewählt wird, was diese Anforderung vereinfacht.
2. Du SOLLTEST in all deinen Importtabellen die gleiche(n) Profilkennung(en) (E-Mail, Telefonnummer, externe ID usw.) verwenden, um das Risiko der Erstellung doppelter Profile zu minimieren.
   1. Klaviyo erstellt neue Profile, wenn die von dir angegebene Profilkennung nicht mit einem bestehenden Profil in Klaviyo übereinstimmt.
   2. Beispiel: Table1 (E-Mail, fav_color) + Table2 (Phone, birthday)
      1. Dadurch können zwei Profile für dieselbe Person erstellt werden, wenn das Profil derzeit nicht vorhanden ist. Wenn ein Profil vorhanden ist, übernimmt Klaviyo die Profilauflösung und aktualisiert es intern. 
   3. Eine Möglichkeit, dieses Problem zu vermeiden, ist, nur eine einzige Importtabelle für alle deine Profile zu verwenden.

Zirkuläre Import-Export-Schleifenprävention

1. Du SOLLTEST Szenarien, in denen sowohl Import- als auch Exportfunktionen verwendet werden, sorgfältig verwalten, um zirkuläre Import-Export-Schleifen zu vermeiden. Stelle sicher, dass dein Exportprozess keine Daten in eine Tabelle zurückführt, die deiner Importtabelle vorgeschaltet ist, da Klaviyo dieses Szenario derzeit nicht erkennt.
   1. Klaviyo hat noch keine Logik, um dieses Szenario zu erkennen.
   2. Das würde etwa so aussehen:
      1. Bei jedem Export-Synchronisierungszyklus exportiert Klaviyo alle deine Profile
      2. Anschließend fügst du alle deine exportierten Profile durch einige Transformationen zu deiner Importtabelle hinzu.
      3. Bei jedem Importsynchronisierungszyklus liest Klaviyo alle Profile in deiner Importtabelle, die schließlich erneut exportiert werden
   3. Szenarien, in denen dies wahrscheinlich sicher ist
      1. Wenn du nur die Exporttabelle verwendest, um die Zeilen einzuschränken, die zu deiner Importtabelle hinzugefügt werden
      2. Wenn du sicherstellst, dass die Exporttabelle keine Zeilen zu deiner Importtabelle hinzufügt.
   4. Was sind die Folgen einer zirkulären Import-Export-Schleife?
      1. Dies führt zu unnötigen Rechenkosten für dich und Klaviyo.

Fehlerbehebung

Synchronisierung bleibt hängen

Wenn deine Synchronisierung ausgeführt wird, die Daten aber nach mehreren Stunden nicht in Klaviyo angezeigt werden oder die Synchronisierung ungewöhnlich lange dauert, liegt die wahrscheinlichste Ursache bei einem weit zurückliegenden Zeitstempelwert:

• Überprüfe deine Tabelle auf Zeilen mit einem Zeitstempel, der deutlich älter ist als der Rest deiner Daten (z. B. eine Zeile aus dem Jahr 2023, wenn alle anderen aus der letzten Woche stammen). Schon eine einzige Ausreißerzeile zwingt Klaviyo dazu, Tausende leerer 1-Stunden-Zeitfenster zu durchlaufen, bevor es die neuesten Daten erreicht.
• Anpassung: Aktualisiere oder entferne alle Zeilen mit Zeitstempeln, die weit in der Vergangenheit liegen, oder setze sie auf einen aktuellen Wert, bevor du die Synchronisierung aktivierst oder erneut aktivierst. Für Backfills solltest du alle historischen Zeilen auf denselben Zeitstempel der letzten Zeit setzen (z. B. die aktuelle Auftragslaufzeit), um die Anzahl der 1-Stunden-Fenster zu minimieren, die Klaviyo durchsuchen muss. Wenn mehr als ~ 100.000 Zeilen vorhanden sind, solltest du die Zeitstempel in Stapeln von ~ 100.000 Zeilen im Abstand von (mindestens) 61 Minuten festlegen.

Empfohlene Einrichtung des Snowflake-Clustering-Schlüssels

Wenn du deine Snowflake-Tabelle in der Zeitstempelspalte gruppierst, können Klaviyos Importabfragen unnötige Mikropartitionen überspringen und sowohl die Synchronisierungszeit als auch die Kosten für deine Snowflake-Berechnung reduzieren:

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