Data Warehouses

{
  "config": {
    "batch_size": 1000,
    "batch_interval_seconds": 30
  }
}

curl -X POST http://localhost:8084/v1/admin/integrations \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "src_abc123",
    "vendor": "bigquery",
    "name": "BigQuery Events",
    "enabled": true,
    "config": {
      "project_id": "my-gcp-project",
      "dataset": "datafly_events",
      "table": "events",
      "service_account_json": "{\"type\": \"service_account\", \"project_id\": \"...\", ...}",
      "batch_size": 500,
      "batch_interval_seconds": 60
    }
  }'

CREATE TABLE `my-gcp-project.datafly_events.events` (
  event_id STRING NOT NULL,
  type STRING,
  event STRING,
  anonymous_id STRING,
  user_id STRING,
  timestamp TIMESTAMP,
  received_at TIMESTAMP,
  sent_at TIMESTAMP,
  context JSON,
  properties JSON,
  traits JSON,
  source_id STRING,
  integration_id STRING
)
PARTITION BY DATE(timestamp)
CLUSTER BY type, event;

-- Revenue by day
SELECT
  DATE(timestamp) AS day,
  SUM(CAST(JSON_VALUE(properties, '$.total') AS FLOAT64)) AS revenue
FROM `my-gcp-project.datafly_events.events`
WHERE event = 'Order Completed'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
GROUP BY day
ORDER BY day DESC;
 
-- Top pages by views
SELECT
  JSON_VALUE(context, '$.page.url') AS page_url,
  COUNT(*) AS views
FROM `my-gcp-project.datafly_events.events`
WHERE type = 'page'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY page_url
ORDER BY views DESC
LIMIT 20;

curl -X POST http://localhost:8084/v1/admin/integrations \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "src_abc123",
    "vendor": "snowflake",
    "name": "Snowflake Events",
    "enabled": true,
    "config": {
      "account": "abc12345.us-east-1",
      "warehouse": "DATAFLY_WH",
      "database": "ANALYTICS",
      "schema": "DATAFLY",
      "table": "EVENTS",
      "stage": "DATAFLY_STAGE",
      "pipe": "DATAFLY_PIPE",
      "user": "DATAFLY_SERVICE",
      "private_key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
      "batch_size": 1000,
      "batch_interval_seconds": 60
    }
  }'

CREATE TABLE ANALYTICS.DATAFLY.EVENTS (
  event_id STRING NOT NULL,
  type STRING,
  event STRING,
  anonymous_id STRING,
  user_id STRING,
  timestamp TIMESTAMP_NTZ,
  received_at TIMESTAMP_NTZ,
  sent_at TIMESTAMP_NTZ,
  context VARIANT,
  properties VARIANT,
  traits VARIANT,
  source_id STRING,
  integration_id STRING
);

-- Revenue by day
SELECT
  DATE(timestamp) AS day,
  SUM(properties:total::FLOAT) AS revenue
FROM ANALYTICS.DATAFLY.EVENTS
WHERE event = 'Order Completed'
  AND timestamp >= DATEADD(day, -30, CURRENT_TIMESTAMP())
GROUP BY day
ORDER BY day DESC;

curl -X POST http://localhost:8084/v1/admin/integrations \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "src_abc123",
    "vendor": "s3",
    "name": "S3 Event Archive",
    "enabled": true,
    "config": {
      "bucket": "my-analytics-bucket",
      "prefix": "datafly/events/",
      "region": "us-east-1",
      "access_key_id": "AKIAIOSFODNN7EXAMPLE",
      "secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
      "file_format": "json",
      "compression": "gzip",
      "batch_size": 1000,
      "batch_interval_seconds": 300
    }
  }'

s3://my-analytics-bucket/datafly/events/year=2026/month=01/day=29/hour=14/events-1706540000-abc123.json.gz

{"event_id":"evt_001","type":"track","event":"Order Completed","anonymous_id":"anon_123","timestamp":"2026-01-29T14:30:00Z","properties":{"order_id":"ORD-001","total":129.99}}
{"event_id":"evt_002","type":"page","event":"Page Viewed","anonymous_id":"anon_456","timestamp":"2026-01-29T14:30:01Z","properties":{}}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetBucketLocation"
      ],
      "Resource": [
        "arn:aws:s3:::my-analytics-bucket",
        "arn:aws:s3:::my-analytics-bucket/datafly/events/*"
      ]
    }
  ]
}

curl -X POST http://localhost:8084/v1/admin/integrations \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "source_id": "src_abc123",
    "vendor": "redshift",
    "name": "Redshift Events",
    "enabled": true,
    "config": {
      "host": "my-cluster.abc123.us-east-1.redshift.amazonaws.com",
      "port": 5439,
      "database": "analytics",
      "schema": "datafly",
      "table": "events",
      "user": "datafly_service",
      "password": "your_password",
      "s3_bucket": "my-redshift-staging",
      "s3_prefix": "datafly/staging/",
      "s3_region": "us-east-1",
      "iam_role_arn": "arn:aws:iam::123456789012:role/RedshiftS3Access",
      "batch_size": 1000,
      "batch_interval_seconds": 120
    }
  }'

CREATE TABLE datafly.events (
  event_id VARCHAR(64) NOT NULL,
  type VARCHAR(20),
  event VARCHAR(256),
  anonymous_id VARCHAR(64),
  user_id VARCHAR(256),
  timestamp TIMESTAMP,
  received_at TIMESTAMP,
  sent_at TIMESTAMP,
  context SUPER,
  properties SUPER,
  traits SUPER,
  source_id VARCHAR(64),
  integration_id VARCHAR(64)
)
DISTKEY(anonymous_id)
SORTKEY(timestamp);

-- Revenue by day
SELECT
  DATE(timestamp) AS day,
  SUM(properties.total::FLOAT) AS revenue
FROM datafly.events
WHERE event = 'Order Completed'
  AND timestamp >= GETDATE() - INTERVAL '30 days'
GROUP BY day
ORDER BY day DESC;