Quick start

CREATE TABLE events (
  event_id VARCHAR,
  user_id VARCHAR,
  event_type VARCHAR,
  page VARCHAR,
  session_id VARCHAR,
  timestamp TIMESTAMP
) WITH (
  connector = 'mock',
  rows_per_second = '100'
);

CREATE TABLE event_stats (
  window_start TIMESTAMP,
  window_end TIMESTAMP,
  total_events BIGINT,
  unique_users BIGINT
) WITH (
  connector = 'preview'
);

INSERT INTO event_stats
SELECT
  window.start as window_start,
  window.end as window_end,
  COUNT(*) as total_events,
  COUNT(DISTINCT user_id) as unique_users
FROM events
GROUP BY TUMBLE(interval '1 minute')

lmnr pipeline create --name event-analytics --query "INSERT INTO..."

curl -X POST http://localhost:8000/api/v1/pipelines \
  -H "Content-Type: application/json" \
  -d '{"name": "event-analytics", "query": "INSERT INTO..."}'

INSERT INTO click_stats
SELECT
  window.start,
  window.end,
  COUNT(*) as clicks,
  COUNT(DISTINCT user_id) as users
FROM events
WHERE event_type = 'click'
GROUP BY TUMBLE(interval '1 minute')

INSERT INTO events_by_type
SELECT
  event_type,
  window.start,
  window.end,
  COUNT(*) as count
FROM events
GROUP BY event_type, TUMBLE(interval '1 minute')

-- 10-second windows for real-time feedback
SELECT window.start, window.end, COUNT(*) as count
FROM events
GROUP BY TUMBLE(interval '10 seconds')

-- 1-hour windows for aggregated reports
SELECT window.start, window.end, COUNT(*) as count
FROM events
GROUP BY TUMBLE(interval '1 hour')

CREATE TABLE events (...) WITH (
  connector = 'kafka',
  bootstrap_servers = 'localhost:9092',
  topic = 'events',
  format = 'json'
);

CREATE TABLE events (...) WITH (
  connector = 'kinesis',
  stream_name = 'events',
  aws_region = 'us-east-1'
);