Łączenie Klaviyo i Snowflake

Dowiedz się, jak przygotować Snowflake do importów, utworzyć zaplanowaną synchronizację profili w Klaviyo i rozwiązywać problemy z wydajnością importu. Aby zapoznać się z przeglądem obsługiwanych danych i przypadków użycia, zobacz Zrozumienie importu z hurtowni danych w Klaviyo.

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

Zanim zaczniesz, potwierdź, że masz role securityadmin i sysadmin w Snowflake. Uruchom SHOW GRANTS TO USER <your_username>, aby sprawdzić swoje role, i skontaktuj się z administratorem Snowflake, jeśli potrzebujesz dostępu. Następnie wykonaj te kroki, aby umożliwić Klaviyo zaimportowanie Twoich 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. Otwórz rsa_key.pub w edytorze tekstu i skopiuj klucz publiczny bez wierszy -----BEGIN PUBLIC KEY----- i -----END PUBLIC KEY----- oraz bez podziałów wierszy. W skrypcie konfiguracji SQL zamień GENERATE_PUBLIC_KEY w rsa_public_key na tę wartość, zachowując otaczające cudzysłowy pojedyncze. Na komputerze Mac poniższe polecenie kopiuje wartość do schowka.

# 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. Uruchom poniższy skrypt SQL w swoim środowisku Snowflake, aby utworzyć użytkownika serwisowego dla Klaviyo.
   1. Na początku skryptu dostosuj nazwy użytkownika, roli, hurtowni danych, bazy danych i sieci zgodnie z potrzebami. Zachowaj nazwy zapisane WIELKIMI LITERAMI i nie zmieniaj nazw schematów w sekcji DO NOT CHANGE.
   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 2 nowe schematy, KLAVIYO_TMP i 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. Skrypt używa IF NOT EXISTS, aby zachować istniejące obiekty o tych samych nazwach. Jeśli ponownie używasz istniejącego użytkownika usługi, potwierdź, że jego skonfigurowany klucz publiczny pasuje do klucza prywatnego, który wprowadzisz w Klaviyo. Skrypt nie zastępuje klucza publicznego istniejącego użytkownika.

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;

Konfiguracja danych Snowflake

Skrypt konfiguracji tworzy 2 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 sygnatur czasowych

1. Każda tabela MUSI zawierać pole znacznika czasu, które wskazuje, kiedy wiersz został utworzony lub ostatnio zaktualizowany, na przykład inserted_at lub updated_at. Użyj typu danych znacznika czasu Snowflake, aby pole było dostępne w New Row Detection podczas 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. 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. Znacznik czasu wykrywania zmian MUSI odzwierciedlać moment, w którym w tabeli importu wstawiono lub zaktualizowano wiersz. Importy historyczne i ręczne zaczynają od najstarszego znacznika czasu i skanują do przodu w 1-godzinnych oknach. Pojedynczy stary znacznik czasu, np. rekord z 2023 r. wśród nowszych wierszy, może spowolnić te importy, ponieważ Klaviyo skanuje pośrednie puste okna. Kolejne zaplanowane uruchomienia kontynuują od zapisanego postępu; nie restartują od najstarszego wiersza.
3. Weź pod uwagę zagęszczenie wierszy w każdym 1-godzinnym oknie znacznika czasu. Miliony rekordów w tym samym oknie mogą spowolnić synchronizację. Około 100 000 wierszy na okno to wytyczna dotycząca wydajności, a nie stały limit; rozmiar wiersza także wpływa na wydajność.
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 (email, inserted_at)
SELECT email, CURRENT_TIMESTAMP()
FROM source_table;

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

Unikalność i spójność profilu

1. Każda właściwość profilu niebędąca identyfikatorem MUST być importowana tylko z jednej tabeli. Klaviyo odrzuca zduplikowane mapowania właściwości profilu między tabelami, gdy zapisujesz synchronizację. Identyfikatory profilu, takie jak adres e-mail, numer telefonu i zewnętrzny identyfikator, można mapować z wielu tabel.
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. Te tabele mogą utworzyć 2 profile dla tej samej osoby, jeśli adres e-mail i numer telefonu nie zostaną dopasowane do tego samego istniejącego profilu. Aby zaktualizować jeden istniejący profil z obu tabel, użyj identyfikatorów, które już należą do tego profilu, lub uwzględnij współdzielony identyfikator w obu tabelach.
   3. Jednym ze sposobów na uniknięcie tego problemu jest używanie tylko jednej tabeli importów dla całego profilu.

Zapobieganie cyklicznej pętli importu-eksportu

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. Wyglądałoby to mniej więcej tak:
      1. Klaviyo eksportuje nowe lub zaktualizowane profile do Twojego magazynu danych.
      2. Twoje transformacje zapisują te profile w tabeli importu i przesuwają ich znaczniki czasu wykrywania zmian.
      3. Klaviyo ponownie importuje te wiersze, aktualizując profile, które następnie można ponownie wyeksportować.
   2. Przykłady, które pozwalają uniknąć tej pętli:
      1. Używaj wyłącznie wyeksportowanych danych do filtrowania, które wiersze źródłowe są importowane, bez nadpisywania tych wierszy ani ich znaczników czasu wykrywania zmian przy każdym eksporcie.
      2. Nie uwzględniaj wyeksportowanych danych w przekształceniach, które dodają lub aktualizują wiersze w tabeli importu.
   3. Jakie są konsekwencje kołowej pętli importu-eksportu?
      1. Zarówno Ty, jak i Klaviyo poniesiesz niepotrzebne koszty obliczeniowe.

Utwórz zaplanowany import profilu w Klaviyo

Po przygotowaniu tabel utwórz synchronizację dla każdej tabeli, którą chcesz zaimportować. Przygotuj identyfikator konta Snowflake, nazwy hurtowni i bazy danych, nazwę użytkownika usługi oraz pełną zawartość rsa_key.p8. Potrzebujesz też uprawnień do tworzenia synchronizacji hurtowni danych w Klaviyo. Użyj tutaj klucza prywatnego, łącznie z nagłówkiem i stopką PEM; klucz publiczny powinien znaleźć się w skrypcie konfiguracji Snowflake.

1. Otwórz Synchronizacja danych w Klaviyo. Jeśli masz już połączenie, kliknij Utwórz synchronizację.
2. Wybierz Importuj dane z hurtowni danych, wybierz Snowflake, a następnie kliknij Połącz z Snowflake.
3. Wprowadź swoje konto, magazyn, bazę danych, nazwę użytkownika i klucz prywatny. Użyj nazw magazynu, bazy danych i użytkownika ze swojego skryptu konfiguracji. Kliknij przycisk Dalej.
4. Kliknij Test. Gdy połączenie się powiedzie, a Klaviyo znajdzie Twoje tabele importu, kliknij Dalej.
5. Wpisz nazwę synchronizacji, wybierz swoją tabelę i wybierz Profile jako typ obiektu danych. W sekcji Cadence wybierz Zaplanuj uruchomienie i określ częstotliwość, datę rozpoczęcia, godzinę rozpoczęcia oraz strefę czasową. Kliknij przycisk Dalej.
6. W sekcji Change detection wybierz pole znacznika czasu w New Row Detection. Odwzoruj kolumnę identyfikatora tabeli na odpowiadający jej identyfikator profilu Klaviyo, np. adres e-mail, numer telefonu lub zewnętrzny identyfikator, i odwzoruj właściwości profilu, które chcesz zaimportować.
7. Kliknij Gotowe, aby zapisać połączenie i zsynchronizować dane. Po pomyślnym zapisaniu Klaviyo przeniesie Cię z powrotem do Synchronizacji danych.

Zaplanowana synchronizacja importuje istniejące wiersze oraz pobiera nowe lub zaktualizowane wiersze na podstawie wybranego przez Ciebie pola znacznika czasu. Aktualizuj to pole za każdym razem, gdy zmieni się wiersz importu.

Rozwiązywanie problemów

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

Jeśli import jest uruchomiony, ale dane nie pojawiają się w Klaviyo, sprawdź wartości znaczników czasu i liczbę wierszy w każdym oknie znacznika czasu:

• W przypadku wstępnego importu historycznego lub importu ręcznego sprawdź, czy nie ma sygnatur czasowych znacznie starszych niż reszta danych — np. jednego wiersza z 2023 roku, gdy pozostałe pochodzą z ostatniego tygodnia. Klaviyo skanuje pośrednie 1-godzinne okna, zanim dotrze do najnowszych wierszy. W przypadku trwającej, zaplanowanej synchronizacji sprawdź, czy nowe lub zaktualizowane wiersze mają bieżące sygnatury czasowe wykrywania zmian; zmiany z datą wsteczną mogą wypaść przed zapisanym postępem i zostać pominięte.
• Przed włączeniem importu popraw błędne znaczniki czasu wykrywania zmian. W przypadku backfillu użyj niedawnego znacznika czasu w polu wykrywania zmian w tabeli importu, a jednocześnie zachowaj daty opisujące pierwotną aktywność w osobnych kolumnach. W przypadku dużych backfilli celuj w partie po około 100 000 wierszy, ładuj je w odstępach co najmniej 61 minut i oznacz każdą partię znacznikiem CURRENT_TIMESTAMP() w momencie jej załadowania. Dzięki temu wiersze są rozłożone na okna 1-godzinne, zamiast umieszczać cały backfill w jednym oknie.

Zalecana konfiguracja klucza klastrowego Snowflake

Klastrowanie tabeli Snowflake według kolumny znacznika czasu może pomóc zapytaniom importu pomijać niepotrzebne mikro-partycje. Zanim dodasz klucz klastrowania, sprawdź koszty Automatycznego klastrowania w Snowflake; niższe koszty zapytań nie gwarantują niższych kosztów całkowitych. Aby dodać klucz, w tym przykładzie podmień nazwy bazy danych, tabeli i kolumny, a następnie uruchom go w Snowflake:

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