CREATE TABLE AS SELECT¶
1 2 3 4 5 6 7 8 9 10
Create a new ksqlDB materialized table view, along with the corresponding Kafka topic, and stream the result of the query as a changelog into the topic.
Note that the WINDOW clause can only be used if the
from_item is a stream and the query contains
GROUP BY clause.
Note that EMIT
output_refinement defaults to
CHANGES unless explicitly set to
FINAL on a
Joins to streams can use any stream column. If the join criteria is not the key column of the stream ksqlDB will internally repartition the data.
Kafka guarantees the relative order of any two messages from one source partition only if they are also both in the same partition after the repartition. Otherwise, Kafka is likely to interleave messages. The use case will determine if these ordering guarantees are acceptable.
See Partition Data to Enable Joins for more information about how to correctly partition your data for joins.
The primary key of the resulting table is determined by the following rules, in order of priority:
1. if the query has a
GROUP BY, then the resulting number of primary key columns will match the
number of grouping expressions. For each grouping expression:
1. if the grouping expression is a single source column reference, the corresponding primary key
column will match the name, type and contents of the source column.
1. if the grouping expression is a reference to a field within a
STRUCT-type column, then the
corresponding primary key column will match the name, type, and contents of the
1. if the
GROUP BY is any other expression, the primary key will have a system
generated name, unless you provide an alias in the projection, and will match the type and
contents of the result of the expression.
1. if the query has a join see Join Synthetic Key Columns for more info.
1. otherwise, the primary key will match the name, unless you provide an alias in the projection,
and type of the source table's primary key.
The projection must include all columns required in the result, including any primary key columns.
For supported serialization formats,
ksqlDB can integrate with the Confluent Schema Registry.
ksqlDB registers the value schema of the new table with Schema Registry automatically.
The schema is registered under the subject
Specify the WINDOW clause to create a windowed aggregation. For more information, see Time and Windows in ksqlDB.
The WITH clause supports the following properties:
|KAFKA_TOPIC||The name of the Kafka topic that backs this table. If this property is not set, then the name of the table will be used as default.|
|KEY_FORMAT||Specifies the serialization format of the message key in the topic. For supported formats, see Serialization Formats. If this property is not set, the format from the left-most input stream/table is used.|
|VALUE_FORMAT||Specifies the serialization format of the message value in the topic. For supported formats, see Serialization Formats. If this property is not set, the format from the left-most input stream/table is used.|
|FORMAT||Specifies the serialization format of both the message key and value in the topic. It is not valid to supply this property alongside either
|VALUE_DELIMITER||Used when VALUE_FORMAT='DELIMITED'. Supports single character to be a delimiter, defaults to ','. For space and tab delimited values you must use the special values 'SPACE' or 'TAB', not an actual space or tab character.|
|PARTITIONS||The number of partitions in the backing topic. If this property is not set, then the number of partitions of the input stream/table will be used. In join queries, the property values are taken from the left-most stream or table.|
|REPLICAS||The replication factor for the topic. If this property is not set, then the number of replicas of the input stream or table will be used. In join queries, the property values are taken from the left-most stream or table.|
|TIMESTAMP||Sets a column within this stream's schema to be used as the default source of
Note: This doesn't affect the processing of the query that populates this stream. For example, given the following statement:
CREATE STREAM foo WITH (TIMESTAMP='t2') ASThe window into which each row of
|TIMESTAMP_FORMAT||Used in conjunction with TIMESTAMP. If not set the timestamp column must be of type
|WRAP_SINGLE_VALUE||Controls how values are serialized where the values schema contains only a single column. The setting controls how the query will serialize values with a single-column schema.
If set to
If set to
If not supplied, the system default, defined by ksql.persistence.wrap.single.values, then the format's default is used.
Note: Supplying this property for formats that do not support wrapping, for example
- To use Avro or Protobuf, you must have Schema Registry enabled and
ksql.schema.registry.url must be set in the ksqlDB server configuration
file. See Configure ksqlDB for Avro, Protobuf, and JSON schemas.
- Avro and Protobuf field names are not case sensitive in ksqlDB. This matches the ksqlDB
column name behavior.
1 2 3 4 5 6 7 8
1 2 3 4 5 6 7 8 9
1 2 3 4 5 6 7
1 2 3 4 5 6