Systemy reklamowe świetnie liczą konwersje, ale nie zawsze wiedzą, czy firma naprawdę zarobiła. Zamówienie za dwa tysiące złotych może wyglądać w panelu jak sukces, nawet jeśli klient nigdy nie odebrał paczki, płatność nie została zaksięgowana, a sklep został z kosztem zwrotu. Dlatego w pierwszej części serii pokazuje, jak spiąć CRM z BigQuery i zacząć optymalizować kampanie nie pod wirtualny ROAS, ale pod realne, opłacone zamówienia.
Za nim zaczniemy wdrożenie zróbmy szybki test trzeźwości biznesowej. Klient wchodzi na Twój sklep, klika jak szalony, wrzuca do koszyka kurtkę za dwa tysiące złotych, wybiera płatność za pobraniem i… nigdy jej nie odbiera. Co widzi Meta Pixel albo tag Google Analytics? Wielki sukces! Kampania dowozi, algorytm pęka z dumy i uczy się, żeby szukać więcej takich klientów. Co widzi Twoje konto bankowe? Koszt zwrotu, zamrożony towar i fakturę od agencji za „świetne ROAS-y”.
Czas skończyć z tą analityczną fikcją. Systemy reklamowe żyją w świecie iluzji, bo karmiasz je danymi z przeglądarki, które są dziurawe jak szwajcarski ser. Jeśli chcesz, żeby algorytmy optymalizowały kampanie pod realny zysk, a nie pod pobożne życzenia, musisz wpuścić je na swój back-end.
W pierwszej części naszej serii zdemolujemy arkusze Excela i postawimy bezserwerowy potok danych (Data Pipeline). Cel? Automatyczne zrzucanie czystych, opłaconych statusów zamówień z Twojego CRM-u prosto do chmury Google Cloud Platform (GCP). Bez drogich narzędzi SaaS i bez lania wody. Do dzieła.
Plan bitwy: Architektura ELT
Zapomnij o starym, ociężałym ETL. My robimy to w nowoczesnym stylu ELT (Extract, Load, Transform). Zasada jest prosta: najpierw zasysamy surowe dane z CRM w stanie nienaruszonym do BigQuery, a całe sprzątanie i logikę biznesową zostawiamy potężnej mocy obliczeniowej chmury.
Pociąg rusza co godzinę. Cloud Scheduler daje sygnał, budzi lekką funkcję Cloud Functions, a ta odpytuje API Twojego sklepu o zamówienia z ostatnich 60 minut i „streamuje” je prosto do bazy.
Krok 1. Budujemy fundament, czyli stół w BigQuery
Zanim odpalimy Pythona, musimy przygotować miejsce na dane. Wchodzisz do konsoli BigQuery i tworzysz tabelę w warstwie Stage.
Wskazówka: tabela działa w trybie append-only, czyli dopisujemy każdy nowy stan zamówienia zamiast nadpisywać poprzedni. Dzięki temu zachowujemy historię zmian: od PENDING, przez PAID, aż po ewentualne CANCELLED albo RETURNED. Partycjonujemy tabelę po ingested_at, czyli po momencie załadowania danych do hurtowni. To ułatwia kontrolę kosztów i debugowanie pipeline’u.
Wklej to do edytora w BigQuery i kliknij Run:
CREATE OR REPLACE TABLE `your_project_id.analytics_raw.crm_orders` (
order_id STRING OPTIONS(description="ID zamówienia z CRM"),
customer_email STRING OPTIONS(description="Surowy email do późniejszego hashowania"),
customer_phone STRING OPTIONS(description="Surowy telefon"),
total_revenue NUMERIC OPTIONS(description="Wartość brutto"),
payment_status STRING OPTIONS(description="Status: PAID, PENDING, CANCELLED"),
created_at TIMESTAMP OPTIONS(description="Data złożenia zamówienia"),
updated_at TIMESTAMP OPTIONS(description="Data ostatniej modyfikacji w CRM"),
ingested_at TIMESTAMP OPTIONS(description="Moment załadowania rekordu do BigQuery")
)
PARTITION BY DATE(ingested_at)
CLUSTER BY payment_status, order_id;
Krok 2. Python wchodzi na scenę (Cloud Function)
Teraz czas na magię, która zastąpi Twojego pracownika klikającego „pobierz CSV”. Tworzymy Cloud Function (środowisko Python 3.10+). Ten skrypt zaloguje się do API Twojego CRM-u, zbierze zamówienia, ładnie je sformatuje i wrzuci do BigQuery.
Oto minimalny szkielet pliku main.py. Trzeba go dopasować do konkretnego API sklepu, szczególnie w zakresie paginacji, statusów zamówień i formatu dat (nie jest łatwo ale trudno też nie, pamiętaj nie wklejać od tak wszystkiego co znajdziesz w necie). To nie jest jeszcze pełny produkcyjny system, ale solidny fundament pod automatyczny pipeline:
iimport os
import requests
from decimal import Decimal
from datetime import datetime, timedelta, timezone
from google.cloud import bigquery
# Autoryzacja wewnątrz GCP dzieje się automatycznie, bez haseł w kodzie
bq_client = bigquery.Client()
CRM_API_URL = os.environ.get("CRM_API_URL", "https://api.twoj-sklep.pl/v1/orders")
CRM_API_TOKEN = os.environ.get("CRM_API_TOKEN")
DATASET_ID = "analytics_raw"
TABLE_ID = "crm_orders"
if not CRM_API_TOKEN:
print("Missing CRM_API_TOKEN")
return "Missing CRM_API_TOKEN", 500
# Sprawdzamy tylko to, co zmieniło się w ciągu ostatniej godziny
now = datetime.now(timezone.utc)
one_hour_ago = now - timedelta(hours=1)
formatted_start_time = one_hour_ago.strftime("%Y-%m-%dT%H:%M:%SZ")
headers = {
"Authorization": f"Bearer {CRM_API_TOKEN}",
"Content-Type": "application/json"
}
params = {
"modified_since": formatted_start_time,
"limit": 250
}
#Uwaga produkcyjna: wiele API zwraca dane stronicami. Jeśli Twój sklep ma więcej niż 250 zmienionych zamówień w godzinę, musisz dopisać obsługę paginacji: page, cursor, next_page albo inny mechanizm właściwy dla danego CRM-u. W przeciwnym razie część zamówień może nie trafić do hurtowni.
try:
response = requests.get(CRM_API_URL, headers=headers, params=params)
response.raise_for_status()
crm_data = response.json().get("orders", [])
except requests.exceptions.RequestException as e:
print(f"💥 API CRM leży i kwiczy: {e}")
return f"Error: {e}", 500
if not crm_data:
print("☕ Brak nowych ruchów w CRM w ciągu ostatniej godziny. Idziemy na kawę.")
return "Success: No data to load", 200
rows_to_insert = []
for order in crm_data:
# Mapujemy chaos z API na porządek w BigQuery
mapped_row = {
"order_id": str(order.get("id")),
"customer_email": order.get("email").strip().lower() if order.get("email") else None,
"customer_phone": order.get("phone_number") if order.get("phone_number") else None,
"total_revenue": str(Decimal(str(order.get("total_amount", "0")))),
"payment_status": str(order.get("status")).upper(),
"created_at": order.get("date_created"),
"updated_at": order.get("date_modified"),
"ingested_at": now.isoformat()
}
rows_to_insert.append(mapped_row)
# Wpychamy dane strumieniem do BigQuery
table_ref = bq_client.dataset(DATASET_ID).table(TABLE_ID)
table = bq_client.get_table(table_ref)
errors = bq_client.insert_rows_json(table, rows_to_insert)
if errors == []:
print(f"🚀 Sukces! {len(rows_to_insert)} zamówień wylądowało w chmurze.")
return f"Loaded {len(rows_to_insert)} rows.", 200
else:
print(f"🔴 Coś rypnęło przy wsadzie do BQ: {errors}")
return f"BQ Errors: {errors}", 500
Żeby to ruszyło, w zakładce requirements.txt musisz dorzucić listę zakupów dla Pythona:
functions-framework==3.*
requests==2.31.0
google-cloud-bigquery==3.18.0
Trzy rzeczy, o których musisz pamiętać (i nie dać ciała)
- Duplikaty to Twój nowy kumpel: Ponieważ streamujemy dane co godzinę, to samo zamówienie pojawi się w bazie kilka razy (np. najpierw ze statusem
PENDING, a potemPAID). I bardzo dobrze! Nie filtruj tego w Pythonie. Historię zmian wyciągniemy za pomocą SQL-a w kolejnych krokach. - RODO patrzy: w tej warstwie trzymasz surowe maile i telefony klientów. Mogą być potrzebne do późniejszego hashowania i dopasowania konwersji po stronie Google albo Meta, ale nie powinny trafiać do dashboardów, arkuszy ani szerokiego zespołu marketingowego. Dostęp do tabeli raw powinien mieć wyłącznie pipeline i wąska grupa techniczna. W kolejnej warstwie dane osobowe należy hashować, np. SHA-256, zanim użyjesz ich w integracjach reklamowych.
- Sekrety nie mieszkają w kodzie: token do API CRM-u nie powinien leżeć w pliku ani w repozytorium. W przykładzie używamy zmiennej środowiskowej dla czytelności, ale w produkcji lepiej trzymać go w Secret Managerze i nadać funkcji dostęp tylko do konkretnego sekretu.
- Endpoint nie może być publiczny: jeśli funkcja działa jako HTTP endpoint, ogranicz jej wywołanie do Cloud Scheduler i odpowiedniego konta serwisowego. Inaczej ktoś z zewnątrz może odpalać funkcję, generować koszty albo obciążać API sklepu.
- Odporność na ponowne uruchomienie: jeśli Cloud Scheduler wywoła funkcję dwa razy, proces nie powinien się wysypać. W warstwie raw dopuszczamy duplikaty, bo chcemy zachować historię zmian. W kolejnej warstwie SQL trzeba jednak jawnie wybrać najnowszy status zamówienia po
order_id,updated_atiingested_at.
Co dalej?
Pierwszy krok za nami. Twoja hurtownia danych właśnie zyskała układ pokarmowy i stabilnie trawi dane finansowe z CRM-u. Ale to dopiero połowa sukcesu.
Teraz potrzebujemy drugiego strumienia: lepszych danych o zachowaniu użytkowników na stronie www. Nie chodzi o magiczne „obejście” prywatności, tylko o uporządkowanie server-side trackingu, ograniczenie strat danych i odzyskanie kontroli nad tym, co naprawdę trafia do systemów analitycznych. W Części 2 wejdziemy poziom wyżej: postawimy Server-Side Google Tag Manager na własnej subdomenie, uporządkujemy wysyłkę eventów i pokaże, jak robić to bez udawania, że technologia zwalnia nas z obowiązków zgód, RODO i transparentności.
Stay tuned, robimy inżynierię, nie mity!