How to Create Stream from Query in ksqlDB

REMEMBER - You can create streams by either importing existing topics as streams, by creating empty streams, or by querying other streams.

This is an example of creating a new stream from a query:

CREATE STREAM orlando_flights
AS
  SELECT flight_id, from_airport, to_airport
  FROM source_stream
  EMIT CHANGES;

This will also create a physical topic in your Apache Kafka cluster where ksqlDB stores the query's output.

Important! By default, your new stream will only process newly produced records from the source streams. To process the source stream from the beginning, run SET 'auto.offset.reset' = 'earliest'; before executing the CREATE statement.

You can also define the stream properties using the WITH clause.

CREATE STREAM orlando_flights
WITH (
    KAFKA_TOPIC = 'orlando_flights_v2',
  PARTITIONS = 10
)
AS
  SELECT flight_id, from_airport, to_airport
  FROM source_stream
  EMIT CHANGES;

There isn’t any required property. By default, the stream name is used as KAFKA_TOPIC and the rest of the properties are inferred from the source_stream and the query.