Create a new stream with the specified columns and properties. Columns can be any of the data types supported by ksqlDB.
ksqlDB adds the implicit columns
ROWKEY to every stream
and table, which represent the corresponding Kafka message timestamp and
message key, respectively. The timestamp has milliseconds accuracy.
The WITH clause supports the following properties:
|KAFKA_TOPIC (required)||The name of the Kafka topic that backs this source. The topic must either already exist in Kafka, or PARTITIONS must be specified to create the topic. Command will fail if the topic exists with different partition/replica counts.|
|VALUE_FORMAT (required)||Specifies the serialization format of the message value in the topic. Supported formats:
|PARTITIONS||The number of partitions in the backing topic. This property must be set if creating a STREAM without an existing topic (the command will fail if the topic does not exist).|
|REPLICAS||The number of replicas in the backing topic. If this property is not set but PARTITIONS is set, then the default Kafka cluster configuration for replicas will be used for creating a new topic.|
|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.|
|KEY||Optimization hint: If the Kafka message key is also present as a field/column in the Kafka message value, you may set this property to associate the corresponding field/column with the implicit
|TIMESTAMP||By default, the implicit
|TIMESTAMP_FORMAT||Used in conjunction with TIMESTAMP. If not set will assume that the timestamp field is a
|WRAP_SINGLE_VALUE||Controls how values are deserialized where the value schema contains only a single field. The setting controls how ksqlDB will deserialize the value of the records in the supplied
If set to
If set to
If not supplied, the system default, defined by ksql.persistence.wrap.single.values and defaulting to
Note: Supplying this property for formats that do not support wrapping, for example
|WINDOW_TYPE||By default, the topic is assumed to contain non-windowed data. If the data is windowed, i.e., was created using ksqlDB using a query that contains a
|WINDOW_SIZE||By default, the topic is assumed to contain non-windowed data. If the data is windowed, i.e., was created using ksqlDB using a query that contains a
For more information on timestamp formats, see DateTimeFormatter.
- To use Avro or Protobuf, you must have Schema Registry enabled and
ksql.schema.registry.urlmust be set in the ksqlDB Server configuration file. See Configure ksqlDB for Avro or Protobuf.
- Avro and Protobuf field names are not case sensitive in ksqlDB. This matches the ksqlDB column name behavior.
1 2 3
Page last revised on: 2020-04-29