10.6.1 – Creating Streams & Tables

Module: Section 10 – Kafka Ecosystem & Tools
Section: 10.6 – Querying & Processing with ksqlDB
Goal: Master the DDL syntax for creating streams and tables in ksqlDB

Introduction

Creating streams and tables is the foundation of ksqlDB development. This module covers all DDL (Data Definition Language) statements.

Why This Module Matters: Proper stream/table creation is critical for correctdata processing and performance.

CREATE STREAM

Basic Syntax

CREATE STREAM stream_name (
  column_name data_type,
  ...
) WITH (
  KAFKA_TOPIC='topic_name',
  VALUE_FORMAT='format',
  ...
);

Example: User Clicks

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

CREATE TABLE

Basic Syntax

CREATE TABLE table_name (
  key_column data_type PRIMARY KEY,
  ...
) WITH (
  KAFKA_TOPIC='topic_name',
  VALUE_FORMAT='format',
  ...
);

Example: User Profiles

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

WITH Clause Options

Required Properties

KAFKA_TOPIC: Source Kafka topic

WITH (KAFKA_TOPIC='orders')

VALUE_FORMAT: Serialization format

WITH (VALUE_FORMAT='JSON')      -- JSON
WITH (VALUE_FORMAT='AVRO')      -- Avro (requires Schema Registry)
WITH (VALUE_FORMAT='PROTOBUF')  -- Protobuf
WITH (VALUE_FORMAT='DELIMITED') -- CSV

Optional Properties

TIMESTAMP: Timestamp column

WITH (TIMESTAMP='order_timestamp')

TIMESTAMP_FORMAT: Custom timestamp format

WITH (TIMESTAMP_FORMAT='yyyy-MM-dd HH:mm:ss')

PARTITIONS: Number of partitions (for derived streams/tables)

WITH (PARTITIONS=6)

REPLICAS: Replication factor

WITH (REPLICAS=3)

KEY: Message key

WITH (KEY='user_id')

Data Types

DROP Statements

DROP STREAM stream_name;
DROP STREAM stream_name DELETE TOPIC;  -- Also delete Kafka topic

DROP TABLE table_name;
DROP TABLE table_name DELETE TOPIC;

DESCRIBE & SHOW Commands

DESCRIBE user_clicks;
DESCRIBE EXTENDED user_clicks;

SHOW STREAMS;
SHOW TABLES;
SHOW TOPICS;
SHOW QUERIES;

Topic Creation Behavior

Explicit Topic Reference (Recommended for Production)

When you specify an existing Kafka topic, ksqlDB uses that topic as-is:

CREATE STREAM orders_explicit (
  order_id VARCHAR KEY,
  customer_id BIGINT,
  total DOUBLE,
  order_time BIGINT
) WITH (
  KAFKA_TOPIC='existing_orders_topic',  -- Must pre-exist
  VALUE_FORMAT='JSON'
);

Requirements:

• Topic must already exist in Kafka
• ksqlDB will fail if topic doesn't exist
• Use for production deployments

Benefits:

• ✅ Full control over partition count
• ✅ Full control over replication factor
• ✅ Explicit configuration and naming
• ✅ Clear topic lifecycle management

Automatic Topic Creation (Development Only)

If you don't specify KAFKA_TOPIC, ksqlDB auto-creates a topic:

CREATE STREAM orders_auto (
  order_id VARCHAR KEY,
  customer_id BIGINT,
  total DOUBLE
) WITH (
  VALUE_FORMAT='JSON', 
  PARTITIONS=4,  -- Optional override
  REPLICAS=2     -- Optional override
);

Behavior:

• Generates topic with prefix KSQL_ (e.g., KSQL_ORDERS_AUTO_0)
• Uses broker defaults for replication if not specified
• Topic lifecycle tied to ksqlDB stream/table

⚠️ Production Warning:

• Auto-generated topic names are not intuitive
• Harder to manage and monitor
• Risk of orphaned topics
• Always reference existing topics in production

Best Practices for Topic Management

Production Checklist:

# 1. Create topic with proper configuration
kafka-topics --create \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --partitions 12 \
  --replication-factor 3 \
  --config cleanup.policy=delete \
  --config retention.ms=604800000

# 2. Verify topic exists
kafka-topics --describe --topic orders --bootstrap-server localhost:9092

# 3. Then create ksqlDB stream/table
CREATE STREAM orders (
  ...
) WITH (
  KAFKA_TOPIC='orders',  -- References existing topic
  VALUE_FORMAT='AVRO'
);

Creating Streams from Query Results

Persistent Query (creates new topic):

CREATE STREAM high_value_orders WITH (
  KAFKA_TOPIC='high_value_orders',  -- Explicit name
  PARTITIONS=6,
  REPLICAS=3
) AS
SELECT * FROM orders WHERE total > 1000;

Without KAFKA_TOPIC (auto-generated name):

CREATE STREAM high_value_orders AS
SELECT * FROM orders WHERE total > 1000;
-- Results in topic: KSQL_HIGH_VALUE_ORDERS_0 (auto-generated)

Best Practice: Always specify KAFKA_TOPIC for production queries:

CREATE STREAM enriched_orders WITH (
  KAFKA_TOPIC='enriched_orders',  -- Clear, explicit naming
  VALUE_FORMAT='AVRO',
  PARTITIONS=12
) AS
SELECT 
  o.*,
  u.username,
  u.email
FROM orders o
LEFT JOIN users u ON o.customer_id = u.user_id;

Key Takeaways

1. CREATE STREAM: For unbounded event logs
2. CREATE TABLE: For mutable state with primary key
3. WITH Clause: Kafka topic, format, and options
4. VALUE_FORMAT: JSON, AVRO, PROTOBUF, DELIMITED
5. PRIMARY KEY: Required for tables, optional for streams
6. Topic Creation: Always pre-create topics in production with explicit names
7. Development: Auto-creation is convenient for testing only

What's Next?

Next Module: 10.6.2 – Transformations & Filters