Łączenie Klaviyo i Snowflake

Na potrzeby tego artykułu używamy terminu „tabela”, ale widoki, widoki zmaterializowane i tabele są prawidłowymi obiektami płatka śniegu, które można zaimportować. Tak długo, jak Klaviyo może wykonać dla obiektu polecenie SELECT col1 FROM nazwa_tabeli, możesz użyć tego, co wolisz.

Słowa kluczowe "MUST", "MUST NOT", "REQUIRED", "SHALL", "SHALL NOT", "SHOULD", "SHOULD NOT", "RECOMMENDED",  "MAY" i "OPTIONAL" w tym dokumencie należy interpretować zgodnie z opisem w RFC 2119.

Konfiguracja administratora Snowflake

W tej sekcji opisano kroki, które należy wykonać w środowisku Snowflake, aby umożliwić Klaviyo zaimportowanie danych.

1. Wygeneruj klucz prywatny, uruchamiając na lokalnym terminalu następujące polecenie:

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

1. Wygeneruj klucz publiczny, który odwołuje się do klucza prywatnego, uruchamiając na terminalu następujące polecenie:

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

1. Skopiuj plik rsa_key.pub i wklej go do poniższego skryptu, aby zastąpić wartość symbolu zastępczego „GENERATE_PUBLIC_KEY” dla parametru user_rsa_public_key. Poniższy skrypt będzie działać na komputerach Mac. Jeśli wolisz, możesz jednak otworzyć rsa_key.pub w środowisku IDE i skopiować pełną zawartość pliku.

# 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. Uruchom poniższy skrypt w środowisku Snowflake, aby utworzyć użytkownika usługi, z którego Klaviyo będzie mogło korzystać. Aby ukończyć poniższą konfigurację, musisz mieć uprawnienia securityadmin i sysadmin. Aby sprawdzić, jakie role masz, uruchom polecenie SHOW GRANTS TO USER <your_username> <twoja_nazwa_użytkownika> i upewnij się, że na liście znajdują się obie role. Jeśli zajdzie potrzeba dostosowania roli, skontaktuj się z administratorem systemu.
   1. Wszelkie zmienne ustawione na początku skryptu należy zaktualizować.
   2. Podsumowując, będziesz mieć następujące możliwości:
      1. Wybierz istniejącą hurtownię lub utwórz nową
      2. Aby przechowywać nowe schematy, wybierz istniejącą bazę danych lub utwórz nową
      3. Utwórz dwa nowe schematy KLAVIYO_TMP oraz KLAVIYO_IMPORT_FROM_DWH
      4. Utwórz nową politykę sieciową i wyświetl listę dozwolonych adresów IP Klaviyo
      5. Utwórz użytkownika i rolę dla Klaviyo
   3. Ten skrypt jest idempotentny (można go bezpiecznie uruchomić wiele razy), ale nie zastąpi istniejących obiektów o nazwach powodujących konflikt.

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;

Konfiguracja danych płatka śniegu

Powyżej utworzono dwa nowe schematy.

• Z KLAVIYO_TMP będzie korzystać wyłącznie Klaviyo. NIE WOLNO modyfikować żadnych tabel utworzonych w tym schemacie. Klaviyo usunie te tabele, gdy nie będą już potrzebne.
• KLAVIYO_IMPORT_FROM_DWH to miejsce, w którym należy przechowywać tabele finałowe, aby Klaviyo mogło je zaimportować. Gdy przejdziesz przez proces tworzenia synchronizacji, wszystkie tabele w tym schemacie zostaną wyświetlone na liście, z której możesz wybierać. Aby uniknąć zamieszania podczas konfiguracji, NALEŻY przechowywać tylko tabele finałowe, które chcesz zaimportować.

Wszystkie tabele, które zamierzasz zaimportować do Klaviyo, muszą spełniać następujące kryteria.

Wymagania dotyczące sygnatury czasowej

1. Tabele MUSZĄ zawierać pole znacznika czasu, który wskazuje, kiedy wiersz został utworzony lub zaktualizowany. Często będzie to wstawić_at lub zaktualizować_at. Ustawienie to można ustawić dla każdej tabeli podczas procesu tworzenia synchronizacji.
   1. Pole sygnatury czasowej MUSI stale rosnąć (czyli zawsze musi stawać się większe lub stale utrzymywać się na tym samym poziomie, ale nigdy nie maleć).
   2. Po utworzeniu synchronizacji NIE WOLNO ustawiać wartości znacznika czasu dla wiersza na czas przypadający w przeszłości, w przeciwnym razie Klaviyo może nie wybrać tego wiersza.
   3. Jeśli przestrzegasz powyższych wymagań, strefa czasowa tego konkretnego pola nie jest istotna dla Klaviyo
   4. Sygnatury czasowe MUSZĄ być podane w strefie czasowej UTC lub zawierać informacje o strefie czasowej. W przypadku braku informacji o strefie czasowej Klaviyo przyjmie UTC. W przypadku właściwości niestandardowych sygnatury te pozostają w postaci ciągu znaków, co umożliwia ich interpretację w preferowanej strefie czasowej.
2. Pole znacznika czasu MUSI odzwierciedlać datę wstawienia wiersza i powinno się zgrupować w pobliżu bieżącej daty. Klaviyo synchronizuje dane, skanując 1-godzinne okna, zaczynając od najstarszej wartości znacznika czasu w tabeli. Pojedynczy wiersz ze znacznikiem czasu daleko w przeszłości (np. rekord z 2023 roku, gdy wszystkie pozostałe są najnowsze) powoduje, że Klaviyo iteruje w każdym cyklu synchronizacji co godzinę od tej daty. Jest to obecne ograniczenie, które powinno zostać usunięte w nadchodzącej wersji.
3. Weź pod uwagę gęstość wierszy w oknie sygnatury czasowej 1–godzina. Ponieważ dane są ładowane w partiach po 1-godzinnych oknach sygnatury czasowej, miliony rekordów w tym samym 1-godzinnym oknie mogą spowodować powolne działanie lub wstrzymanie synchronizacji. Chociaż górny limit gęstości wierszy zależy od ilości danych w każdym wierszu, dobrą zasadą, o której należy pamiętać, jest 100 000 wierszy na 1 godzinę okna znacznika czasu.
4. Za każdym razem, gdy dodajesz do tabeli wiersze, z których korzystasz, Klaviyo zaleca ustawianie pola znacznika czasu za pomocą funkcji CURRENT_TIMESTAMP() lub podobnej funkcji. Wiele wierszy może mieć ten sam znacznik czasu. Zobacz przykład poniżej.

INSERT INTO table_name AS 

SELECT ... 

    , CURRENT_TIMESTAMP() AS inserted_at

... 

Struktura tabeli

1. Tabele POWINNY być traktowane jako zawierające tylko załącznik (tzw. „tylko wstawianie”)
   1. Jeśli wolisz zaktualizować wiersze w miejscu, MUSZĄ zaktualizować pole znacznika czasu, aby Klaviyo mogło zidentyfikować zmianę.
2. Tabele POWINNY być uporządkowane według kolumny znacznika czasu. Snowflake zajmie się grupowaniem i partycjonowaniem na podstawie Twojego zamówienia reklamowego. Pomoże to zoptymalizować zapytania importowe Klaviyo, a jednocześnie obniży koszty obliczeń w Snowflake

Wyjątkowość i spójność profilu

1. Dla każdego profilu MUSZĄ one wynikać tylko z jednego źródła danych (tabeli). Klaviyo uniemożliwia wybranie tej samej właściwości z różnych tabel podczas tworzenia synchronizacji, co upraszcza ten wymóg.
2. Aby zminimalizować ryzyko utworzenia zduplikowanego profilu, NALEŻY używać tych samych identyfikatorów profilu (adresu e-mail, numeru telefonu, identyfikatora zewnętrznego itp.) we wszystkich tabelach importu.
   1. Klaviyo utworzy nowy profil, jeśli podany identyfikator profilu nie pasuje do istniejącego profilu w Klaviyo.
   2. Przykład: Tabela1 (e-mail, fav_color) + Tabela2 (telefon, urodziny)
      1. To może spowodować utworzenie 2 profili dla tej samej osoby, jeśli obecnie taki profil nie istnieje. Jeśli profil istnieje, Klaviyo zajmie się rozwiązaniem profilu i wewnętrznymi aktualizacjami. 
   3. Jednym ze sposobów na uniknięcie tego problemu jest używanie tylko jednej tabeli importów dla całego profilu.

Zapobieganie zapętleniom importu i eksportu w obiegu

1. NALEŻY starannie zarządzać scenariuszami, w których funkcje importu i eksportu są używane, aby zapobiec zapętleniu się importu i eksportu. Upewnij się, że proces eksportu nie powoduje przesyłania danych z powrotem do tabeli znajdującej się powyżej tabeli importu, ponieważ Klaviyo obecnie nie wykrywa takiego scenariusza.
   1. Klaviyo nie dysponuje jeszcze logiką pozwalającą wykryć taki scenariusz.
   2. Wyglądałoby to mniej więcej tak:
      1. W każdym cyklu synchronizacji eksportu Klaviyo wyeksportuje cały Twój profil
      2. Następnie poprzez serię przekształceń dodajesz cały wyeksportowany profil do tabeli importów.
      3. W każdym cyklu synchronizacji importu Klaviyo odczytuje wszystkie profile z tabeli importów, które ostatecznie zostaną ponownie wyeksportowane
   3. Scenariusze, w których jest to prawdopodobnie bezpieczne
      1. jeśli tabela eksportu służy tylko do ograniczenia liczby wierszy dodawanych do tabeli importu
      2. Jeśli potwierdzisz, tabela eksportu nie dodaje wierszy do tabeli importu.
   4. Jakie są konsekwencje kołowej pętli importu-eksportu?
      1. Zarówno Ty, jak i Klaviyo poniesiesz niepotrzebne koszty obliczeniowe.

Rozwiązywanie problemów

Wygląda na to, że synchronizacja utknęła

Jeśli po kilku godzinach synchronizacja jest uruchomiona, ale dane nie są wyświetlane w Klaviyo lub trwa niezwykle długo, najbardziej prawdopodobną przyczyną jest wartość sygnatury czasowej należąca do dawnej przeszłości:

• Sprawdź, czy w tabeli nie znajdują się wiersze ze znacznikiem czasu znacznie starsze niż pozostałe dane (na przykład jeden wiersz z 2023 r., gdy wszystkie pozostałe pochodzą z ubiegłego tygodnia). Nawet pojedynczy wiersz o wartości odstającej zmusza Klaviyo do przechodzenia przez tysiące pustych godzinnych okienek przed wyświetleniem najnowszych danych.
• Środki zaradcze: przed włączeniem lub ponownym włączeniem synchronizacji zaktualizuj lub usuń wszystkie wiersze ze znacznikami czasu daleko w przeszłości albo ustaw je na wartości najnowszej. W przypadku uzupełnień zapasowych ustaw wszystkie wiersze danych historycznych na ten sam znacznik czasu (np. bieżący czas wykonania zadania), aby zminimalizować liczbę okien, które Klaviyo przeskanować będzie po 1 godzinie. Jeśli wierszy zawiera więcej niż ~100 000, ustawiaj ich znaczniki czasu partiami po ~100 000 co (co najmniej) 61 minut.

Zalecana konfiguracja klucza klastrowego Snowflake

Dzięki skupieniu tabeli Snowflake w kolumnie znacznika czasu zapytania importu Klaviyo pominą niepotrzebne mikropartycje, co pozwoli skrócić czas synchronizacji i skrócić koszty obliczeniowe usługi Snowflake:

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