Collegamento tra Klaviyo e Snowflake

Scopri come preparare Snowflake per le importazioni, creare una sincronizzazione pianificata dei profili in Klaviyo e risolvere i problemi relativi alle prestazioni di importazione. Per una panoramica dei dati supportati e dei casi d’uso, vedi Informazioni sull’importazione del magazzino dati in Klaviyo.

Ai fini di questo articolo utilizziamo il termine "tabella", ma le viste, le viste materializzate e le tabelle sono tutti oggetti Snowflake validi che possono essere importati. Fintanto che Klaviyo può eseguire SELECT col1 FROM table_name sull'oggetto, puoi utilizzare quello che preferisci.

Le parole chiave "DEVI", "NON DEVI", "OBBLIGATORIO", "DEVE", "NON DEVE", "DOVRESTI", "NON DOVRESTI", "CONSIGLIATO", "PUOI" e "FACOLTATIVO" in questo documento devono essere interpretate come descritto in RFC 2119.

Configurazione amministratore Snowflake

Prima di iniziare, verifica di avere i ruoli securityadmin e sysadmin in Snowflake. Esegui SHOW GRANTS TO USER <your_username> per controllare i tuoi ruoli e contatta il tuo amministratore Snowflake se ti serve l'accesso. Poi segui questi passaggi per consentire a Klaviyo di importare i tuoi dati.

1. Genera una chiave privata eseguendo il seguente comando nel tuo terminale locale:

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

1. Genera una chiave pubblica che faccia riferimento alla chiave privata eseguendo il seguente comando nel tuo terminale:

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

1. Apri rsa_key.pub in un editor di testo e copia la chiave pubblica senza le righe -----BEGIN PUBLIC KEY----- e -----END PUBLIC KEY----- né interruzioni di riga. Nello script di configurazione SQL, sostituisci GENERATE_PUBLIC_KEY in rsa_public_key con questo valore, mantenendo le virgolette singole circostanti. Su Mac, il comando qui sotto copia il valore negli appunti.

# 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. Esegui il seguente script SQL nel tuo ambiente Snowflake per creare un utente di servizio per Klaviyo.
   1. Personalizza i nomi di utente, ruolo, warehouse, database e rete all'inizio dello script, in base alle necessità. Mantieni i nomi in maiuscolo e non modificare i nomi dello schema nella sezione DO NOT CHANGE.
   2. In sintesi, potrai:
      1. Scegli un magazzino esistente o creane uno nuovo
      2. Scegli un database esistente o creane uno nuovo per contenere i nuovi schemi
      3. Crea 2 nuovi schemi, KLAVIYO_TMP e KLAVIYO_IMPORT_FROM_DWH
      4. Crea un nuovo criterio di rete e consenti l'invio di un elenco di IP Klaviyo
      5. Crea un utente e un ruolo per Klaviyo
   3. Lo script utilizza IF NOT EXISTS per preservare gli oggetti esistenti con gli stessi nomi. Se riutilizzi un service user esistente, verifica che la public key configurata corrisponda alla private key che inserirai in Klaviyo. Lo script non sostituisce la public key di un utente esistente.

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;

Configurazione dei dati con Snowflake

Lo script di configurazione crea 2 schemi:

• KLAVIYO_TMP sarà utilizzato esclusivamente da Klaviyo. NON DEVI modificare le tabelle create in questo schema. Klaviyo eliminerà queste tabelle quando non saranno più necessarie.
• KLAVIYO_IMPORT_FROM_DWH è il luogo in cui devi archiviare le tabelle finali affinché Klaviyo le importi. Durante il processo di creazione della sincronizzazione, tutte le tabelle di questo schema saranno elencate tra cui potrai scegliere. Pertanto, DOVRESTI memorizzare solo le tabelle finali che desideri importare per evitare confusione durante la configurazione.

Tutte le tabelle che intendi importare in Klaviyo devono soddisfare i seguenti criteri.

Requisiti relativi al timestamp

1. Ogni tabella DEVE contenere un campo timestamp che indichi quando una riga è stata creata o aggiornata per l’ultima volta, ad esempio inserted_at o updated_at. Utilizza un tipo di dati timestamp di Snowflake, così il campo è disponibile in Rilevamento nuove righe durante la creazione della sincronizzazione.
   1. Il campo della data e dell'ora DEVE aumentare in modo monotono (cioè deve sempre diventare più grande o rimanere lo stesso, mai più piccolo).
   2. Dopo la creazione della sincronizzazione, NON DEVI impostare il valore della data e dell'ora di una riga su un'ora passata, altrimenti Klaviyo potrebbe non rilevare quella riga.
   3. Le tue marche temporali DEVONO essere in UTC o includere le informazioni sul fuso orario. Se mancano le informazioni sul fuso orario, Klaviyo presumerà l'UTC. Per le proprietà personalizzate, questi timestamp rimangono in formato stringa, consentendoti di interpretarli nel tuo fuso orario preferito.
2. Il timestamp di rilevamento delle modifiche MUST riflettere il momento in cui una riga viene inserita o aggiornata nella tabella di importazione. Le importazioni storiche e manuali partono dal timestamp più vecchio ed eseguono la scansione in avanti in finestre di 1 ora. Un singolo timestamp vecchio, come un record del 2023 tra righe recenti, può rallentare queste importazioni perché Klaviyo scansiona le finestre vuote intermedie. Le esecuzioni programmate successive continuano dall’avanzamento salvato; non ripartono dalla riga più vecchia.
3. Considera la densità delle righe all’interno di ciascuna finestra temporale di 1 ora. Milioni di record nella stessa finestra possono rallentare una sincronizzazione. Circa 100.000 righe per finestra è un’indicazione per le prestazioni, non un limite fisso; anche la dimensione delle righe influisce sulle prestazioni.
4. Klaviyo consiglia di impostare il campo della data e dell'ora con CURRENT_TIMESTAMP() o una funzione equivalente ogni volta che aggiungi righe alla tabella da cui ci sincronizzeremo. Più righe possono avere lo stesso timestamp. Vedi l'esempio qui sotto.

INSERT INTO table_name (email, inserted_at)
SELECT email, CURRENT_TIMESTAMP()
FROM source_table;

Struttura della tabella

1. Le tabelle DEVONO essere trattate come tabelle di sola aggiunta (dette anche tabelle di sola aggiunta)
   1. Se preferisci aggiornare le righe in posizione, DEVI aggiornare il campo della data e dell'ora in modo che Klaviyo possa identificare la modifica.
2. Le tabelle DOVREBBERO essere ordinate sulla tua colonna della data e dell'ora. Snowflake gestirà il clustering e il partizionamento in base al tuo ordine di inserimento. Questo aiuterà a ottimizzare le richieste di importazione di Klaviyo, mantenendo i costi di calcolo in Snowflake

Univocità e coerenza del profilo

1. Ogni Proprietà del profilo non identificativa DEVE essere importata da una sola tabella. Klaviyo rifiuta le mappature di proprietà duplicate tra le tabelle quando salvi una sincronizzazione. Gli identificatori del profilo, come e-mail, numero di telefono e ID esterno, possono essere mappati da più tabelle.
2. DOVRESTI utilizzare gli stessi identificatori del profilo (e-mail, numero di telefono, ID esterno, ecc.) in tutte le tabelle di importazione, per ridurre al minimo il rischio di creazione di profili duplicati.
   1. Klaviyo creerà nuovi profili se l'identificatore del profilo che fornisci non corrisponde a un profilo esistente all'interno di Klaviyo.
   2. Esempio: tabella 1 (e-mail, colore_fax) + tabella 2 (telefono, compleanno)
      1. Queste tabelle possono creare 2 profili per la stessa persona se l’e-mail e il numero di telefono non vengono associati allo stesso profilo esistente. Per aggiornare 1 profilo esistente da entrambe le tabelle, usa identificatori che appartengono già a quel profilo oppure includi un identificatore condiviso in entrambe le tabelle.
   3. Un modo per evitare questo problema è utilizzare una sola tabella di importazione per tutti i tuoi profili.

Prevenzione dei loop circolari di importazione ed esportazione

1. DOVRESTI gestire con attenzione gli scenari in cui vengono utilizzate sia le funzionalità di importazione che quelle di esportazione per evitare loop di importazione-esportazione circolari. Assicurati che il processo di esportazione non reinserisca i dati in una tabella che si trova a monte della tabella di importazione, poiché al momento Klaviyo non rileva questo scenario.
   1. Avrebbe un aspetto simile a:
      1. Klaviyo esporta i profili nuovi o aggiornati nel tuo data warehouse.
      2. Le tue trasformazioni scrivono questi profili in una tabella di importazione e fanno avanzare i relativi timestamp di rilevamento delle modifiche.
      3. Klaviyo importa di nuovo quelle righe, aggiornando i profili che poi possono essere esportati di nuovo.
   2. Esempi che evitano questo ciclo:
      1. Usa i dati esportati solo per filtrare quali righe di origine vengono importate, senza riscrivere tali righe o i relativi timestamp di rilevamento delle modifiche a ogni esportazione.
      2. Escludi i dati esportati dalle trasformazioni che aggiungono o aggiornano righe nella tua tabella di importazione.
   3. Quali sono le conseguenze di un circuito circolare di importazione-esportazione?
      1. Ciò comporterà costi di calcolo inutili sia per te che per Klaviyo.

Crea un'importazione programmata del profilo in Klaviyo

Dopo aver preparato le tabelle, crea una sincronizzazione per ogni tabella che vuoi importare. Tieni a portata di mano l'identificatore dell'account Snowflake, i nomi del data warehouse e del database, il nome utente del servizio e l'intero contenuto di rsa_key.p8. Ti serve anche l'autorizzazione per creare sincronizzazioni del magazzino dati in Klaviyo. Usa qui la chiave privata, includendo l'intestazione e il piè di pagina PEM; la chiave pubblica va inserita nello script di configurazione di Snowflake.

1. Apri Sincronizzazione dei dati in Klaviyo. Se hai già una connessione, fai clic su Crea sincronizzazione.
2. Seleziona Importa dati dal tuo magazzino dati, seleziona Snowflake, quindi fai clic su Connetti a Snowflake.
3. Inserisci account, warehouse, database, nome utente e chiave privata. Usa i nomi del warehouse, del database e dell’utente dal tuo script di configurazione. Clicca su Avanti.
4. Fai clic su Test. Dopo che la connessione va a buon fine e Klaviyo trova le tue tabelle di importazione, fai clic su Avanti.
5. Inserisci un nome per la sincronizzazione, seleziona la tua tabella e seleziona Profili come tipo di oggetto dati. In Cadenza, seleziona Pianifica esecuzione e scegli la frequenza, la data di inizio, l’ora di inizio e il fuso orario. Clicca su Avanti.
6. In Rilevamento modifiche, seleziona il campo timestamp in Rilevamento nuove righe. Mappa la colonna identificatore della tabella all’identificatore del profilo Klaviyo corrispondente, come e-mail, numero di telefono o ID esterno, e mappa le Proprietà del profilo che vuoi importare.
7. Fai clic su Fine per salvare la connessione e sincronizzare. Dopo un salvataggio riuscito, Klaviyo ti riporta a Sincronizzazione dei dati.

La sincronizzazione pianificata importa le righe esistenti e rileva le righe nuove o aggiornate utilizzando il campo timestamp che hai selezionato. Continua ad aggiornare quel campo ogni volta che una riga di importazione cambia.

Risoluzione dei problemi

La sincronizzazione sembra bloccata

Se un'importazione è in esecuzione ma i dati non vengono visualizzati in Klaviyo, controlla i valori dei timestamp e il numero di righe in ciascuna finestra di timestamp:

• Per un’importazione storica iniziale o un’importazione manuale, verifica la presenza di timestamp molto più vecchi rispetto al resto dei dati, ad esempio 1 riga del 2023 quando le altre sono della scorsa settimana. Klaviyo analizza gli intervalli intermedi di 1 ora prima di arrivare alle righe più recenti. Per una sincronizzazione pianificata continua, verifica che le righe nuove o aggiornate abbiano timestamp di rilevamento delle modifiche aggiornati; le modifiche retrodatate possono risultare precedenti all’avanzamento salvato e non essere rilevate.
• Prima di abilitare un’importazione, correggi i timestamp di rilevamento delle modifiche errati. Per un backfill, usa un timestamp recente per il campo di rilevamento delle modifiche della tabella di importazione, mantenendo però eventuali date che descrivono l’attività originale in colonne separate. Per backfill di grandi dimensioni, punta a lotti di circa 100.000 righe, caricali a distanza di almeno 61 minuti l’uno dall’altro e assegna a ogni lotto il timestamp CURRENT_TIMESTAMP() quando viene caricato. In questo modo le righe vengono distribuite nelle finestre di 1 ora, invece di inserire l’intero backfill in un’unica finestra.

Configurazione consigliata della chiave di clustering Snowflake

Eseguire il clustering di una tabella Snowflake sulla colonna timestamp può aiutare le query di importazione a saltare micro-partizioni non necessarie. Verifica i costi dell’Automatic Clustering di Snowflake prima di aggiungere una chiave di clustering; costi inferiori delle query non garantiscono costi totali inferiori. Per aggiungere una chiave, sostituisci i nomi di database, tabella e colonna in questo esempio ed eseguilo in Snowflake:

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