Quick start

Build your first streaming SQL pipeline in 5 minutes.

Prerequisites

• e6 Ingestion Engine running locally or in your cluster
• Basic SQL knowledge

Step 1: Create a Source

Define where your data comes from. This example uses a mock source for testing:

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'
);

Step 2: Create a Destination

Define where results go. For testing, use the preview sink:

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

Step 3: Write Your Query

Connect source to destination with a transformation:

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')

window_start	window_end	total_events	unique_users
2024-01-15 10:00:00	2024-01-15 10:01:00	6	3
2024-01-15 10:01:00	2024-01-15 10:02:00	8	4

---

Step 4: Run the Pipeline

Via the UI: Click "Create Pipeline" and paste your SQL.

Via the CLI:

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

Via the API:

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

---

What's Happening?

1. Mock source generates fake events continuously
2. TUMBLE window groups events into 1-minute buckets
3. Aggregations calculate stats for each window
4. Preview sink displays results in the UI

---

Try These Modifications

Filter Events

Add a WHERE clause to process only specific event types:

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')

window_start	window_end	clicks	users
10:00:00	10:01:00	2	2

---

Add Grouping

Group results by an additional dimension:

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')

event_type	window_start	window_end	count
page_view	10:00:00	10:01:00	3
click	10:00:00	10:01:00	2
purchase	10:00:00	10:01:00	1

---

Change Window Size

Adjust the window duration for different granularity:

-- 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')

---

Real Data Sources

Replace the mock source with real connectors:

Kafka:

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

Kinesis:

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

Continue

Continue with DDL for the complete connection-table and view syntax.