10.5.1 – Introduction to ksqlDB

Module: Section 10 – Kafka Ecosystem & Tools
Section: 10.5 – ksqlDB Fundamentals
Goal: Understand what ksqlDB is, its purpose, and when to use it

Introduction

ksqlDB is a SQL-based stream processing engine built on top of Kafka Streams. It enables developers to build real-time applications using familiar SQL syntax without writing Java code.

Why This Module Matters: ksqlDB makes stream processing accessible to SQL developers and significantly reduces development time for common use cases.

What Is ksqlDB?

Definition

ksqlDB: Event streaming database that enables real-time data processing using SQL.

Key Characteristics:

SQL interface for Kafka Streams
Runs as a server (not a library)
Supports continuous queries (never-ending SELECT)
Built-in state stores
Exactly-once processing

ksqlDB vs Kafka Streams

Feature	ksqlDB	Kafka Streams API
Language	SQL	Java/Scala/Python
Deployment	Server (standalone)	Embedded library
Learning Curve	Low (SQL knowledge)	High (programming required)
Flexibility	Limited to SQL features	Full programmatic control
State Management	Automatic	Manual (developer-managed)
Use Case	Standard transformations	Complex custom logic
Development Time	Fast (declarative)	Slower (imperative)

When to Use ksqlDB

✅ Use ksqlDB If:

Familiar with SQL, not Java
Standard transformations (filter, map, aggregate, join)
Rapid prototyping needed
Team prefers declarative approach
Need interactive query development

❌ Use Kafka Streams API If:

Complex custom logic required
Need fine-grained control over state stores
Performance optimization critical
Need custom serdes or processors
Building library (not server)

ksqlDB Architecture

High-Level Components

┌────────────────────────────────────────────────┐
│              ksqlDB Cluster                    │
│  ┌──────────────┐  ┌──────────────┐          │
│  │  ksqlDB      │  │  ksqlDB      │          │
│  │  Server 1    │  │  Server 2    │          │
│  │              │  │              │          │
│  │ - REST API   │  │ - REST API   │          │
│  │ - SQL Engine │  │ - SQL Engine │          │
│  │ - State      │  │ - State      │          │
│  └──────────────┘  └──────────────┘          │
└────────────────────────────────────────────────┘
         ↓ (reads/writes)       ↓
┌────────────────────────────────────────────────┐
│             Kafka Cluster                      │
│  Topics: users, orders, enriched-orders        │
└────────────────────────────────────────────────┘
         ↑ (SQL queries)
┌────────────────────────────────────────────────┐
│  ksqlDB CLI / REST API / Confluent Control    │
│  Center                                        │
└────────────────────────────────────────────────┘

ksqlDB Server

Role: Executes SQL queries, manages state, interacts with Kafka.

Key Features:

REST API on port 8088
Embedded Kafka Streams application
Local state stores (RocksDB)
Query processing engine

ksqlDB CLI

Role: Interactive SQL shell for ksqlDB.

Example:

ksql http://localhost:8088

ksql> SHOW TOPICS;
ksql> SELECT * FROM users EMIT CHANGES;

Core Concepts

1. Streams

Definition: Unbounded sequence of events (insert-only).

Example: User clickstream, IoT sensor data, logs.

SQL:

CREATE STREAM user_clicks (
  user_id BIGINT,
  page VARCHAR,
  timestamp BIGINT
) WITH (
  KAFKA_TOPIC='clicks',
  VALUE_FORMAT='JSON'
);

2. Tables

Definition: Mutable state (latest value per key).

Example: User profiles, product catalog, account balances.

SQL:

CREATE TABLE users (
  user_id BIGINT PRIMARY KEY,
  username VARCHAR,
  email VARCHAR
) WITH (
  KAFKA_TOPIC='users',
  VALUE_FORMAT='AVRO'
);

3. Queries

Types:

Push Query (continuous): SELECT ... EMIT CHANGES
Pull Query (point-in-time): SELECT ... WHERE key = 123

Push Query Example:

SELECT user_id, page, timestamp
FROM user_clicks
WHERE page = 'checkout'
EMIT CHANGES;

Result: Continuous stream of checkout events.

Pull Query Example:

SELECT username, email
FROM users
WHERE user_id = 123;

Result: Current state for user 123.

ksqlDB Workflow Example

Scenario: Real-Time Order Enrichment

Goal: Enrich orders with user information.

Step 1: Create Source Stream

CREATE STREAM orders (
  order_id VARCHAR,
  user_id BIGINT,
  product_id VARCHAR,
  amount DOUBLE
) WITH (
  KAFKA_TOPIC='orders',
  VALUE_FORMAT='JSON'
);

Step 2: Create User Table

CREATE TABLE users (
  user_id BIGINT PRIMARY KEY,
  username VARCHAR,
  email VARCHAR
) WITH (
  KAFKA_TOPIC='users',
  VALUE_FORMAT='AVRO'
);

Step 3: Enrich Orders with User Data

CREATE STREAM enriched_orders AS
SELECT
  o.order_id,
  o.amount,
  u.username,
  u.email
FROM orders o
LEFT JOIN users u ON o.user_id = u.user_id
EMIT CHANGES;

Result: New Kafka topic enriched_orders with joined data.

{
  "order_id": "12345",
  "amount": 99.99,
  "username": "alice",
  "email": "alice@example.com"
}

ksqlDB Features

1. Continuous Queries

Traditional SQL (batch):

SELECT * FROM orders WHERE amount > 100;

Result: Snapshot at query time.

ksqlDB (streaming):

SELECT * FROM orders WHERE amount > 100 EMIT CHANGES;

Result: Continuous stream of high-value orders.

2. Windowed Aggregations

SELECT
  user_id,
  COUNT(*) AS click_count,
  WINDOWSTART AS window_start,
  WINDOWEND AS window_end
FROM user_clicks
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY user_id
EMIT CHANGES;

Result: Hourly click counts per user.

3. Materialized Views

CREATE TABLE user_click_counts AS
SELECT
  user_id,
  COUNT(*) AS total_clicks
FROM user_clicks
GROUP BY user_id
EMIT CHANGES;

Result: Queryable table with latest counts.

Pull Query:

SELECT total_clicks FROM user_click_counts WHERE user_id = 123;

4. UDFs (User-Defined Functions)

Built-in Functions: UCASE(), CONCAT(), CAST(), etc.

Custom UDF (Java):

@UdfDescription(name = "multiply", description = "multiplies 2 numbers")
public class MultiplyUdf {
  @Udf(description = "multiply two non-nullable INTs")
  public long multiply(
    @UdfParameter(value = "v1") long v1,
    @UdfParameter(value = "v2") long v2
  ) {
    return v1 * v2;
  }
}

Usage:

SELECT multiply(price, quantity) AS total FROM orders EMIT CHANGES;

Use Cases

1. Real-Time Analytics

-- Top products by revenue (last hour)
SELECT
  product_id,
  SUM(amount) AS revenue
FROM orders
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY product_id
ORDER BY revenue DESC
LIMIT 10
EMIT CHANGES;

2. Fraud Detection

-- Flag users with >5 failed login attempts in 5 minutes
CREATE TABLE suspicious_users AS
SELECT
  user_id,
  COUNT(*) AS failed_attempts
FROM login_events
WHERE status = 'FAILED'
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY user_id
HAVING COUNT(*) > 5
EMIT CHANGES;

3. Data Enrichment

-- Enrich transactions with merchant details
CREATE STREAM enriched_transactions AS
SELECT
  t.transaction_id,
  t.amount,
  m.merchant_name,
  m.category
FROM transactions t
LEFT JOIN merchants m ON t.merchant_id = m.merchant_id
EMIT CHANGES;

4. Data Transformation

-- Transform JSON to Avro
CREATE STREAM orders_avro
WITH (VALUE_FORMAT='AVRO') AS
SELECT * FROM orders_json EMIT CHANGES;

Key Takeaways

ksqlDB: SQL-based stream processing engine built on Kafka Streams.
SQL for Streams: Enables SQL developers to build real-time applications.
Server Architecture: Runs as standalone server (not embedded library).
Streams vs Tables: Streams are insert-only, tables are mutable.
Push vs Pull Queries: Push for continuous, pull for point-in-time.
Continuous Queries: EMIT CHANGES creates never-ending queries.
Use Cases: Real-time analytics, fraud detection, enrichment, transformation.
When to Use: Standard transformations, SQL expertise, rapid development.
vs Kafka Streams: ksqlDB is declarative (SQL), Kafka Streams is imperative (Java).
Materialized Views: Queryable state stores with pull queries.

What's Next?

Next Module: 10.5.2 – Streams vs Tables in ksqlDB

What We Covered:

✅ What ksqlDB is
✅ ksqlDB vs Kafka Streams
✅ Architecture and components
✅ Core concepts
✅ Use cases

What's Coming:

❌ Deep dive into streams vs tables
❌ ksqlDB architecture details

References:

Kafka Streams → Section 7: Kafka Streams
ksqlDB operations → 10.8: ksqlDB in Production