DataNostro nemá vestavěný BigQuery export jako GA4 → BigQuery z GCP. Místo toho posíláme každý event přes webhook a vy ho stream-insertujete do BigQuery jednou Cloud Function (~30 řádků). Setup je 15 minut, výsledek je identický jako BigQuery streaming v GA4 — včetně schema flexibility (eventy si tvaříte sami).

Architektura

sGTM event → DataNostro engine → POST webhook → Cloud Function → BigQuery (streaming insert)

Webhook se posílá asynchronně z celery workeru, takže neovlivňuje latenci sGTM dispatch na ad platformy. Retry: 3× s exponential backoff. Dead-letter queue v Redisu pro events, které opakovaně selžou.

1. Vytvořte BigQuery dataset + tabulku

V GCP Console:

bq mk --dataset --location=EU YOUR_PROJECT:datanostro_events
bq mk --table --schema='
  event_id:STRING,
  event_name:STRING,
  event_time:TIMESTAMP,
  tenant_id:STRING,
  payload:JSON,
  user_ip_masked:STRING,
  user_agent:STRING,
  request_source:STRING
' YOUR_PROJECT:datanostro_events.events

Tip: EU location = data residency stejně jako DataNostro. payload jako JSON typ vám umožní query SELECT JSON_VALUE(payload, '$.items[0].sku') bez schema migrací při změně eventů.

2. Cloud Function (Python 3.12)

main.py:

import functions_framework
from google.cloud import bigquery
import os, hmac, hashlib

bq = bigquery.Client()
TABLE = "YOUR_PROJECT.datanostro_events.events"
SHARED_SECRET = os.environ["DN_WEBHOOK_SECRET"]


@functions_framework.http
def receive_dn_event(request):
    # Verify HMAC signature — DataNostro posílá X-DN-Signature header
    sig = request.headers.get("X-DN-Signature", "")
    body = request.get_data()
    expected = hmac.new(SHARED_SECRET.encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        return ("invalid signature", 401)

    event = request.get_json(silent=True)
    if not event:
        return ("no payload", 400)

    row = {
        "event_id": event.get("id"),
        "event_name": event.get("event_name"),
        "event_time": event.get("created_at"),
        "tenant_id": event.get("tenant_id"),
        "payload": event.get("request_body"),
        "user_ip_masked": event.get("source_ip_masked"),
        "user_agent": event.get("user_agent"),
        "request_source": event.get("request_source"),
    }
    errors = bq.insert_rows_json(TABLE, [row])
    if errors:
        return (f"bq insert failed: {errors}", 500)
    return ("ok", 204)

requirements.txt:

functions-framework==3.*
google-cloud-bigquery==3.*

3. Deploy

export DN_WEBHOOK_SECRET=$(openssl rand -hex 32)
echo "Uloz si: $DN_WEBHOOK_SECRET"

gcloud functions deploy receive-dn-event \
  --gen2 --runtime=python312 --region=europe-west3 \
  --source=. --entry-point=receive_dn_event \
  --trigger-http --allow-unauthenticated \
  --set-env-vars="DN_WEBHOOK_SECRET=$DN_WEBHOOK_SECRET"

Pozn.: --allow-unauthenticated je OK protože ověřujeme HMAC podpis. Bez podpisu Cloud Function nikdy nedělá BigQuery insert.

4. Nastavit webhook v DataNostro

Dashboard → Webhooks → Vytvořit nový webhook
URL: URL z gcloud functions deploy výstupu (např. https://europe-west3-YOUR_PROJECT.cloudfunctions.net/receive-dn-event)
Events: vyberte event.captured (= každý sGTM event s capture row v naší DB)
HMAC secret: vložte $DN_WEBHOOK_SECRET z kroku 3
Save → klikněte Test webhook → mělo by přijít HTTP 204

5. Ověření v BigQuery

Po 1–2 minutách běžného trafficu:

SELECT event_name, COUNT(*) as n
FROM `YOUR_PROJECT.datanostro_events.events`
WHERE event_time > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 MINUTE)
GROUP BY event_name
ORDER BY n DESC

Co tím získáte vs GA4 BigQuery export

Latence: ~2 s (Cloud Function streaming insert) vs ~6 hodin (GA4 daily batch). Realtime dashboards možné okamžitě.
Schema: JSON payload field = můžete kdykoliv přidat nová pole bez migrací. GA4 export má fixní schema.
Cena: Cloud Functions Free Tier = 2 M invocations / měsíc zdarma. Při 5 M sGTM events/měsíc = ~$0.40/měsíc. BigQuery streaming insert = $0.01/MB (≈ $5/měsíc pro 5 M events). GA4 → BQ export je zdarma ale jen pro GA360 (>$150 k/rok).
Filter na úrovni webhooku: v DataNostro Dashboardu nastavíte, které event typy chcete pushovat — neplatíte za nezajímavé eventy.

Alternativa: denní CSV export do Cloud Storage

Pokud nepotřebujete realtime, dashboard má /dashboard/export/ → ZIP s CSV za posledních N dní. Můžete to volat skrz API klíč + cron + gsutil cp do Cloud Storage → BigQuery scheduled query. Pomalejší, ale zero Cloud Functions setup.

FAQ

Q: Můžu posílat jen subset eventů (např. jen purchase, nikoli page_view)?
Ano. V Webhook settings je filter event_name regex. Pošleme jen eventy které matchují (např. ^(purchase|begin_checkout|add_to_cart)$).

Q: Co s historickými daty před zapnutím webhooku?
Historická EventCapture data můžete jednorázově exportovat přes /api/v1/events/export/?from=…&to=… (API klíč nutný) jako JSONL a importovat přes bq load --source_format=NEWLINE_DELIMITED_JSON.

Q: Funguje to s GA4 BigQuery dataset, kam už pošlu GA4 native export?
Ano — vytvořte druhý dataset (datanostro_events) vedle GA4 (analytics_PROPERTY_ID). Můžete je joinovat přes event_id pokud necháte GA4 client posílat event_id do sGTM (přes X-Gtm-Server-Preview ne, jen v event_data).

Q: Máte to už někde nasazené?
Interně používáme stejný setup pro náš vlastní debug dashboard (datanostro.com je sám sebou prvním klientem). Webhook → Cloud Function → BigQuery. P95 latence webhook delivery: 1,8 s. Ztracenost: 0,003 % (dead-letter queue zachytí zbytek).

BigQuery export přes DataNostro webhook (~30 řádků Cloud Function)