Klaviyo und Snowflake verbinden

Erfahre, wie du Snowflake für Importe vorbereitest, in Klaviyo eine geplante Profil-Synchronisierung erstellst und die Import-Performance behebst. Einen Überblick über unterstützte Daten und Anwendungsfälle findest du unter Data-Warehouse-Import in Klaviyo verstehen.

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-Admin-Einrichtung

Bevor du beginnst, bestätige, dass du in Snowflake die Rollen securityadmin und sysadmin hast. Führe SHOW GRANTS TO USER <your_username> aus, um deine Rollen zu überprüfen, und kontaktiere deine*n Snowflake-Administrator*in, wenn du Zugriff benötigst. Folge dann diesen Schritten, um Klaviyo zu erlauben, deine Daten zu importieren.

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. Öffne rsa_key.pub in einem Texteditor und kopiere den öffentlichen Schlüssel ohne die Zeilen -----BEGIN PUBLIC KEY----- und -----END PUBLIC KEY----- oder Zeilenumbrüche. Ersetze im SQL-Setup-Skript GENERATE_PUBLIC_KEY in rsa_public_key durch diesen Wert und behalte dabei die umgebenden einfachen Anführungszeichen bei. Auf einem Mac kopiert der unten stehende Befehl den Wert in deine Zwischenablage.

# Mac terminal command to copy the public key without PEM delimiters or line breaks
sed '/-----BEGIN PUBLIC KEY-----/d; /-----END PUBLIC KEY-----/d' rsa_key.pub | tr -d '\r\n' | pbcopy

1. Führe das folgende SQL-Skript in deiner Snowflake-Umgebung aus, um eine*n Service-Nutzer*in für Klaviyo zu erstellen.
   1. Passe die Benutzer-, Rollen-, Warehouse-, Datenbank- und Netzwerknamen am Anfang des Skripts nach Bedarf an. Lass die Namen in GROẞBUCHSTABEN, und ändere die Schemanamen im Abschnitt DO NOT CHANGE nicht.
   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 2 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. Das Skript verwendet IF NOT EXISTS, um vorhandene Objekte mit denselben Namen beizubehalten. Wenn du eine*n bestehende*n Service-Benutzer*in wiederverwendest, bestätige, dass der konfigurierte öffentliche Schlüssel mit dem privaten Schlüssel übereinstimmt, den du in Klaviyo eingeben wirst. Das Skript ersetzt nicht den öffentlichen Schlüssel einer bestehenden Nutzer*in.

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 ALL TABLES IN 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 ALL ON ALL VIEWS IN SCHEMA IDENTIFIER($schema_name_tmp) TO ROLE IDENTIFIER($role_name);
GRANT ALL ON FUTURE VIEWS 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 ALL TABLES IN 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);
GRANT SELECT ON ALL VIEWS IN SCHEMA IDENTIFIER($schema_name_import) TO ROLE IDENTIFIER($role_name);
GRANT SELECT ON FUTURE VIEWS IN SCHEMA IDENTIFIER($schema_name_import) TO ROLE IDENTIFIER($role_name);

COMMIT;

Snowflake-Dateneinrichtung

Das Einrichtungsskript erstellt 2 Schemas:

• 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. Jede Tabelle MUSS ein Zeitstempelfeld enthalten, das angibt, wann eine Zeile erstellt oder zuletzt aktualisiert wurde, z. B. inserted_at oder updated_at. Verwende einen Snowflake-Zeitstempel-Datentyp, damit das Feld bei der Erstellung der Synchronisierung in Neuerkennung von Zeilen verfügbar ist.
   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. 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. Der Änderungs-Erkennungs-Zeitstempel MUST widerspiegeln, wann eine Zeile in die Importtabelle eingefügt oder aktualisiert wird. Historische und manuelle Importe beginnen mit dem ältesten Zeitstempel und scannen in 1-Stunden-Fenstern vorwärts. Ein einzelner alter Zeitstempel, z. B. ein Datensatz aus 2023 unter aktuellen Zeilen, kann diese Importe verlangsamen, weil Klaviyo die dazwischenliegenden leeren Fenster scannt. Nachfolgende geplante Ausführungen setzen mit ihrem gespeicherten Fortschritt fort; sie starten nicht erneut bei der ältesten Zeile.
3. Berücksichtige die Zeilendichte innerhalb jedes 1-Stunden-Timestamp-Fensters. Millionen von Datensätzen im selben Fenster können eine Synchronisierung verlangsamen. Etwa 100.000 Zeilen pro Fenster sind eine Leistungsrichtlinie, kein fester Grenzwert; auch die Zeilengröße beeinflusst die Leistung.
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 (email, inserted_at)
SELECT email, CURRENT_TIMESTAMP()
FROM source_table;

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. Jede nicht identifizierende Profileigenschaft MUST aus nur 1 Tabelle importiert werden. Klaviyo lehnt doppelte Eigenschaftszuordnungen über Tabellen hinweg ab, wenn du eine Synchronisierung speicherst. Profil-Identifikatoren wie E-Mail, Telefonnummer und externe ID können aus mehreren Tabellen zugeordnet werden.
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. Diese Tabellen können 2 Profile für dieselbe Person erstellen, wenn die E-Mail-Adresse und die Telefonnummer nicht zum selben vorhandenen Profil aufgelöst werden. Um 1 vorhandenes Profil aus beiden Tabellen zu aktualisieren, verwende Identifikatoren, die bereits zu diesem Profil gehören, oder füge in beiden Tabellen einen gemeinsamen Identifikator ein.
   3. Eine Möglichkeit, dieses Problem zu vermeiden, ist, nur eine einzige Importtabelle für alle deine Profile zu verwenden.

Verhinderung von zirkulären Import-Export-Schleifen

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. Das würde etwa so aussehen:
      1. Klaviyo exportiert neue oder aktualisierte Profile in dein Warehouse.
      2. Deine Transformationen schreiben diese Profile in eine Importtabelle und setzen ihre Zeitstempel für die Änderungserkennung weiter.
      3. Klaviyo importiert diese Zeilen erneut und aktualisiert Profile, die dann wieder exportiert werden können.
   2. Beispiele, die diese Schleife vermeiden:
      1. Verwende exportierte Daten nur, um zu filtern, welche Quellzeilen importiert werden, ohne diese Zeilen oder ihre Zeitstempel zur Änderungserkennung bei jedem Export neu zu schreiben.
      2. Halte exportierte Daten aus Transformationen heraus, die Zeilen in deiner Importtabelle hinzufügen oder aktualisieren.
   3. Was sind die Folgen einer zirkulären Import-Export-Schleife?
      1. Dies führt zu unnötigen Rechenkosten für dich und Klaviyo.

Erstelle einen geplanten Profilimport in Klaviyo

Nachdem du deine Tabellen vorbereitet hast, erstelle eine Synchronisierung für jede Tabelle, die du importieren möchtest. Halte die Konto-Kennung deines Snowflake-Kontos, die Namen von Warehouse und Datenbank, den Service-Benutzernamen und den vollständigen Inhalt von rsa_key.p8 bereit. Du benötigst außerdem die Berechtigung, Data-Warehouse-Synchronisierungen in Klaviyo zu erstellen. Verwende hier den privaten Schlüssel, einschließlich PEM-Kopfzeile und -Fußzeile; der öffentliche Schlüssel gehört in das Snowflake-Einrichtungsskript.

1. Öffne Datensynchronisierung in Klaviyo. Wenn du bereits eine Verbindung hast, klicke auf Synchronisierung erstellen.
2. Wähle Daten aus deinem Data Warehouse importieren aus, wähle Snowflake aus und klicke dann auf Mit Snowflake verbinden.
3. Gib dein Konto, Lager, Datenbank, Benutzername und Privater Schlüssel ein. Verwende die Warehouse-, Datenbank- und Benutzernamen aus deinem Setup-Skript. Klicke auf Weiter.
4. Klicke auf Test. Nachdem die Verbindung erfolgreich hergestellt wurde und Klaviyo deine Importtabellen gefunden hat, klicke auf Weiter.
5. Gib einen Namen für die Synchronisierung ein, wähle deine Tabelle aus und wähle Profile als Datenobjekttyp aus. Wähle unter Cadence die Option Geplante Ausführung aus und wähle Häufigkeit, Startdatum, Startzeit und Zeitzone. Klicke auf Weiter.
6. Wähle unter Change detection dein Zeitstempelfeld in New Row Detection aus. Ordne die Identifikatorspalte deiner Tabelle der entsprechenden Klaviyo-Profilkennung zu, z. B. E-Mail-Adresse, Telefonnummer oder externe ID, und ordne die Profileigenschaften zu, die du importieren möchtest.
7. Klicke auf Fertig, um die Verbindung zu speichern und zu synchronisieren. Nach dem erfolgreichen Speichern führt dich Klaviyo zurück zu Datensynchronisierung.

Die geplante Synchronisierung importiert vorhandene Zeilen und erfasst neue oder aktualisierte Zeilen anhand des Zeitstempelfelds, das du ausgewählt hast. Halte dieses Feld bei jeder Änderung einer Importzeile auf dem neuesten Stand.

Fehlerbehebung

Synchronisierung bleibt hängen

Wenn ein Import läuft, aber in Klaviyo keine Daten angezeigt werden, prüfe seine Zeitstempelwerte und die Anzahl der Zeilen in jedem Zeitstempelfenster:

• Bei einem ersten historischen Import oder einem manuellen Import solltest du nach Zeitstempeln suchen, die viel älter sind als der Rest der Daten, z. B. 1 Zeile aus 2023, während die anderen aus der vergangenen Woche stammen. Klaviyo scannt die dazwischenliegenden 1‑Stunden‑Fenster, bevor es die aktuellen Zeilen erreicht. Bei einer laufenden geplanten Synchronisierung solltest du prüfen, dass neue oder aktualisierte Zeilen aktuelle Zeitstempel für die Änderungserkennung haben; rückdatierte Änderungen können vor den gespeicherten Fortschritt fallen und übersehen werden.
• Bevor du einen Import aktivierst, korrigiere fehlerhafte Zeitstempel für die Änderungserkennung. Für ein Backfill verwende einen aktuellen Zeitstempel für das Feld zur Änderungserkennung der Importtabelle und behalte alle Datumsangaben, die die ursprüngliche Aktivität beschreiben, in separaten Spalten bei. Bei großen Backfills solltest du Batches von etwa 100.000 Zeilen anstreben, sie mit mindestens 61 Minuten Abstand laden und jeden Batch beim Laden mit CURRENT_TIMESTAMP() versehen. So werden die Zeilen über die 1-Stunden-Fenster verteilt, statt das gesamte Backfill in 1 Fenster zu packen.

Empfohlene Einrichtung des Snowflake-Clustering-Schlüssels

Wenn du eine Snowflake-Tabelle nach der Zeitstempelspalte clusterst, können Importabfragen unnötige Mikro-Partitionen überspringen. Prüfe die Kosten für Snowflakes Automatic Clustering, bevor du einen Clustering-Schlüssel hinzufügst; niedrigere Abfragekosten garantieren keine niedrigeren Gesamtkosten. Um einen Schlüssel hinzuzufügen, ersetze in diesem Beispiel die Datenbank-, Tabellen- und Spaltennamen und führe es in Snowflake aus:

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