`.
### Alternative syntax
If data is in Protobuf or Avro format, alternative syntax can be used, assuming the source topic has a [Schema](https://docs.deltastream.io/reference/sql-syntax/ddl/create-schema_registry) or a [Descriptor](https://docs.deltastream.io/reference/sql-syntax/ddl/create-descriptor_source) associated to it and it will create a Stream definition which includes ALL columns. The user must provide the `value.format`.
### Arguments
#### stream\_name
Specifies the name of the new stream. If the name is case-sensitive, you must wrap it in double quotes; otherwise, the system uses the lower case name.
#### column\_name
The name of a column to be created in the new stream. If the name is case-sensitive, you must wrap it in double quotes; otherwise, the system uses the lower case name.
#### data\_type
The data type of the column. This can include array specifiers. For more information on the data types supported by DeltaStream, see the [data-types](https://docs.deltastream.io/reference/sql-syntax/data-types "mention") reference.
#### NOT NULL
Defines a constraint on the column, ensuring it cannot contain `NULL` values.
#### WITH (stream\_parameter = value \[, … ])
Optionally, this clause specifies [#stream\_parameters](#stream_parameters "mention").
### Stream Parameters
| Parameter Name | Description |
| ------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `topic` | Name of the #entity that has the data for this stream. If the entity doesn't exist, an entity with this name is created in the corresponding store.
Required: No
Default value: Lowercase stream\_name
Type: String
|
| `store` | Name of the store that hosts the #entity for this stream.
Required: No
Default value: Current session's store name
Type: String
Valid values: See list-stores.
|
| `value.format` | Format of message value in the #entity. See data-format-serialization for more information regarding serialization formats.
Required: Yes
Type: String
Valid values: JSON, AVRO, PROTOBUF, PRIMITIVE
Valid values without column definition: AVRO, PROTOBUF
|
| `timestamp` | Name of the column in the stream to use as the timestamp. If not set, the timestamp of the message is used for time based operations such as window aggregations and joins. If the type of this timestamp field is BIGINT, it is expected that the values are epoch milliseconds UTC.
Required: No
Default value: Record's timestamp
Type: String
Valid values: Must be of type BIGINT or TIMESTAMP. See data-types.
|
| `timestamp.format` | The format to use for TIMESTAMP typed fields. See data-types.
Required: No
Default value: sql
Type: String
Valid values: sql, iso8601
|
### Kafka-Specific Parameters
Parameters to be used if the associated [store](https://docs.deltastream.io/overview/core-concepts/store "mention") is type `KAFKA`:
| Parameter Name | Description |
| ----------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `topic.partitions` | The number of partitions to use when creating the entity, if applicable. If the topic already exist, then this value must be equal to the number of partitions in the existing Kafka entity.
Required: Yes, unless topic already exists
Default value: Leftmost source relation topic's partition count
Type: Integer
Valid values: \[1, ...]
|
| `topic.replicas` | The number of replicas to use when creating the topic, if applicable. If the topic already exists, then this value must be equal to the number of replicas in the existing Kafka entity.
Required: Yes, unless topic already exists
Default values: Leftmost source relation topic's replica count
Type: Integer
Valid values: \[1, ...]
|
| `kafka.topic.*` | A configuration specific for the topic being created — for example, Kafka Entity Configuration for Confluent Platform.
Required: No
Default value: None
Type: String
Valid values: Kafka topic configuration specific to the underlying store type.
|
| `key.format` | Format of message key in the #entity. This value can be the same as or different from the value.format. See data-format-serialization for more information regarding serialization formats.
Required: No, unless key.type is set
Default value: None
Type: String
Valid values: JSON, AVRO, PROTOBUF, PRIMITIVE
Valid values without column definition: AVRO, PROTOBUF
|
| `key.type` | Declares the names and data types of key columns. The type is a STRUCT when key.format is a non-primitive value, e.g. 'key.type'='STRUCT\'. For primitive values, the type is one of the #primitive-data-types, e.g. 'key.type'='VARCHAR'.
Note: key.type can not be set if key.columns is already set.
Required: No
Default value: None
Type: String
Valid values: See STRUCT in data-types.
|
| `key.columns` | Specifies the name(s) of the value columns, separated by commas, that will be used to construct the record key.
- For a non-primitive
key.format, the record key will be created as a STRUCT whose fields are the columns listed in this property. - For a primitive
key.format, this property must contain exactly one column with a primitive data type. - Note:
key.columns cannot be set if key.type is already defined.
Required: No
Default: None
Type: String
Valid values: One or more valid column names from the relation’s column list, separated by commas.
|
| `value.columns.exclude` | Specifies the name(s) of the columns, separated by commas, that should be excluded from the record’s value and included only in its key.
- This property can only be set if
key.columns is already defined. - The excluded columns must appear at the end of the relation’s column list and must also be listed in
key.columns.
Required: No
Default: None
Type: String
Valid values: One or more valid column names from the relation’s column list, separated by commas.
|
| `delivery.guarantee` | The fault tolerance guarantees applied when producing to this stream.
Required: No
Default value: at\_least\_once
Type: String
Valid values:
exactly\_once: Produces to the stream using Kafka transactions. These transactions are committed when the query takes a checkpoint. On the consumer side, when setting the Kafka consumer isolation.level configuration to read\_committed, only the committed records display. Since records aren't committed until the query takes a checkpoint, there is some additional delay when using this setting.at\_least\_once: Ensures that records are output to the stream at least once. During query checkpointing, the query waits to receive a confirmation of successful writes from the Kafka broker. If there are issues with the query then duplicate records are possible, as the query attempts to reprocess old data.none: There is no fault tolerance guarantee when producing to the stream. If there are issues on the Kafka broker then records may be lost; if there are issues with the query then output records may be duplicated.
|
| `sink.timestamp.column` | Specifies the name of the value column to be used to set the Kafka record’s timestamp when writing to the Kafka sink’s entity. If no timestamp column is specified, the Kafka producer record is created without an explicit timestamp, allowing the sink’s store to assign a timestamp according to its configured policy.
Required: No
Default value: None
Type: String
Valid values: One of the column names from the sink relation’s column list. Must be of type BIGINT or TIMESTAMP or TIMESTAMP\_LTZ. See data-types.
|
### Kinesis-Specific Parameters
Parameters to be used if the associated [store](https://docs.deltastream.io/overview/core-concepts/store "mention") is type `KINESIS`:
| Parameter Name | Description |
| -------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `topic.shards` | The number of shards to use when creating the topic, if applicable. If the topic already exists, then this value must be equal to the number of shards in the existing Kinesis stream.
Required: Yes, unless topic already exists
Default values: Leftmost source relation topic's shard count
Type: Integer
Valid values: \[1, ...]
Alias: kinesis.shards
|
Kinesis stores provide a delivery guarantee of `at_least_once` when producing events into a sink [#entity](https://docs.deltastream.io/overview/core-concepts/store#entity "mention").
### S3-Specific Parameters
The S3 source periodically discovers file splits under the configured paths and emits them for processing. Discovery interval controls how often the connector looks for new S3 objects. On each run it lists keys in the service’s server-side order (for general-purpose buckets, that’s **lexicographic by key**) and emits files to the job, enforcing a configurable **maximum splits per interval** cap. The connector tracks an internal watermark: the last accepted key for each configured prefix. Therefore, any key **less than or equal to** that watermark is considered already processed and is skipped on later runs (and after restarts). To ensure new data is picked up reliably in long-running jobs, write files under key prefixes that **increase lexicographically over time** (e.g., `YYYY/MM/DD/HH/`) and use zero-padded sequence parts (e.g., `part-000001.jsonl`). Avoid backfilling into older, lexicographically smaller prefixes; if needed, place backfills under a **new, higher-sorting** path so they appear after the currently existing content.
\
Parameters to be used if the associated [store](https://docs.deltastream.io/overview/core-concepts/store "mention") is type `S3`:
| Parameter Name | Description |
| ------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `s3.discovery.interval.seconds` | Specifies how often (in seconds) the S3 source runs discovery to look for new S3 objects under the configured S3 paths. Note that the discovery lists S3 keys in service order (which for general buckets is lexicographic by key) and emits a bounded batch of new splits to the job. Discovery progresses based on the full S3 key order. New files added “behind” the current content (lexicographically smaller keys, e.g., backfills into older date folders) are not picked up.
Required: No
Default value: 10
Type: Long
Valid values: \[1, ...]
|
| `s3.max.splits.per.interval` | An upper bound on how many file splits the connector will emit per discovery interval run.
Required: No
Default value: 20000
Type: Integer
Valid values: \[1, ...]
|
### Format-Specific Parameters
#### Avro
Parameters to be used when writing records into a stream if associated `key.format` or `value.format` is `avro` and the default Avro schema generation must be changed using a base schema for the key and/or value.
When generating an Avro schema for a column using a base schema:
* if the base schema has a field with the same name and data type as that of the column, then the field's definition from the base is used in the generated schema. This includes retaining the base schema's `doc` and `logicalType` for the field.
* if the base schema has a field with the same name as that of the column, but has a different data type, then an Avro schema type definition is generated from the column's data type with the field's `doc` taken from the its corresponding field in the base schema.
{% hint style="info" %}
**Notes**
* Currently supported schema registries are Confluent Cloud and Confluent Platform.
* Known limitation: Confluent Schema Registry must use the default [TopicNameStrategy](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#subject-name-strategy) for creating subject names. See [create-schema\_registry](https://docs.deltastream.io/reference/sql-syntax/ddl/create-schema_registry "mention") for more details.
{% endhint %}
| Parameter Name | Description |
| ------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `avro.base.schema.store` | Name of the store whose schema registry contains the Avro schema subject(s) to be used as the base schema for generating Avro schema for stream's key and/or value.
Required: No
Default values: Current session's store name
Type: Identifier
Valid values: See list-stores.
|
| `avro.base.subject.key` | Name of the subject in the schema registry to obtain the base schema for generating Avro schema for stream's key.
Required: No, unless key.format is set to avro and key.type is defined.
Type: String
|
| `avro.base.subject.value` | Name of the subject in the schema registry to obtain the base schema for generating Avro schema for stream's value columns.
Required: No, unless value.format is set to avro .
Type: String
|
## Examples
#### Create a new stream with passthrough configuration for retention
```sql
CREATE STREAM customers (
ts BIGINT, customer_id VARCHAR, full_name BIGINT, region VARCHAR
) WITH (
'store' = 'kafka_store',
'topic.partitions' = 1,
'topic.replicas' = 2,
'kafka.topic.retention.ms' = '172800000');
```
#### Create a new stream with timestamp column and key/value formats
The following creates a new stream with name `pageviews_json`. This stream reads from an existing topic named `pageviews` in the default store `demostore`, and has a `value.format` of `JSON`. Additionally in the `WITH` clause, we specify that this stream has a key of type `VARCHAR` and uses the `viewtime` column as its `timestamp`:
```sh
demodb.public/demostore# LIST ENTITIES;
+------------------------------------------+------------+
| Name | Is Leaf |
+==========================================+============+
| pageviews | true |
+------------------------------------------+------------+
demodb.public/demostore# CREATE STREAM pageviews_json (
viewtime BIGINT,
userid VARCHAR,
pageid VARCHAR
) WITH (
'topic'='pageviews',
'value.format'='json',
'key.format'='primitive',
'key.type'='VARCHAR',
'timestamp'='viewtime');
+------------+------------------------------+------------+------------------------------------------+
| Type | Name | Command | Summary |
+============+==============================+============+==========================================+
| stream | demodb.public.pageviews_json | CREATE | stream "pageviews_json" was |
| | | | successfully created |
+------------+------------------------------+------------+------------------------------------------+
```
#### Create a new stream in a specific store
The following creates a new stream `pv_kinesis`. This stream reads from an existing topic named `pageviews` in the store `kinesis_store`:
```sh
demodb.public/demostore# CREATE STREAM pv_kinesis (
viewtime BIGINT,
userid VARCHAR,
pageid VARCHAR
) WITH ( 'topic'='pageviews', 'store'='kinesis_store', 'value.format'='json' );
+------------+--------------------------+------------+------------------------------------------+
| Type | Name | Command | Summary |
+============+==========================+============+==========================================+
| stream | demodb.public.pv_kinesis | CREATE | stream "pv_kinesis" was successfully |
| | | | created |
+------------+--------------------------+------------+------------------------------------------+
demodb.public/demostore# LIST STREAMS;
+------------+------------+------------+------------+------------------------------------------+-------------------------------+-------------------------------+
| Name | Type | Owner | State | Properties | Created At | Updated At |
+============+============+============+============+==========================================+===============================+===============================+
| pv_kinesis | stream | sysadmin | created | { "value.format": "json", "topic": | 2024-07-02 21:28:35 +0000 UTC | 2024-07-02 21:28:36 +0000 UTC |
| | | | | "pageviews", "store": | | |
| | | | | "kinesis_store" } | | |
+------------+------------+------------+------------+------------------------------------------+-------------------------------+-------------------------------+
```
#### Create a new stream without an existing entity
The following creates a new stream `visit_count`. As its corresponding topic doesn't exist in the store `kinesis_store`, it requires an additional topic parameter — for example, `topic.shards` — to create the new Kinesis data stream `pv_count` in the store:
```sh
demodb.public/kinesis_store# LIST ENTITIES;
+------------------------------------------+------------+
| Name | Is Leaf |
+==========================================+============+
| pageviews | true |
+------------------------------------------+------------+
demodb.public/kinesis_store# CREATE STREAM visit_count (userid VARCHAR, pgcount BIGINT) WITH ('topic'='pv_count', 'value.format'='json', 'topic.shards' = 1);
+------------+---------------------------+------------+------------------------------------------+
| Type | Name | Command | Summary |
+============+===========================+============+==========================================+
| stream | demodb.public.visit_count | CREATE | stream "visit_count" was successfully |
| | | | created |
+------------+---------------------------+------------+------------------------------------------+
demodb.public/kinesis_store# LIST ENTITIES;
+------------------------------------------+------------+
| Name | Is Leaf |
+==========================================+============+
| pageviews | true |
| pv_count | true |
+------------------------------------------+------------+
demodb.public/kinesis_store# DESCRIBE ENTITY visit_count;
+------------------+------------+-------------+
| Name | Shards | Descriptor |
+==================+============+=============+
| pageviews_one_kb | 1 | |
+------------------+------------+-------------+
demodb.public/kinesis_store# LIST STREAMS;
+-------------+------------+------------+------------+------------------------------------------+-------------------------------+-------------------------------+
| Name | Type | Owner | State | Properties | Created At | Updated At |
+=============+============+============+============+==========================================+===============================+===============================+
| visit_count | stream | sysadmin | created | { "topic.shards": 1, "value.format": | 2024-07-02 21:32:20 +0000 UTC | 2024-07-02 21:32:32 +0000 UTC |
| | | | | "json", "topic": "pv_count", "store": | | |
| | | | | "kinesis_iam_store" } | | |
+-------------+------------+------------+------------+------------------------------------------+-------------------------------+-------------------------------+
```
#### Create a new stream for an existing entity
The following creates a new `users` stream for the existing `users` [#entity](https://docs.deltastream.io/overview/core-concepts/store#entity "mention") in the current [store](https://docs.deltastream.io/overview/core-concepts/store "mention"). This DDL implies that the name of the stream should be used as the name of the entity that hosts the records. This DDL also implies the original structure for the `users` entity:
```sql
CREATE STREAM "users" (
registertime BIGINT,
userid VARCHAR,
regionid VARCHAR,
gender VARCHAR,
interests ARRAY,
contactinfo STRUCT<
phone VARCHAR,
city VARCHAR,
"state" VARCHAR,
zipcode VARCHAR>
) WITH ( 'value.format'='json' );
```
#### Create a new stream with case-sensitive columns
The following creates a new stream `CaseSensitivePV` in the database `DataBase` and schema `Schema2`. This stream reads from a topic named `case_sensitive_pageviews` in store `OtherStore` and has a `value.format` of AVRO and `key.format` of PROTOBUF. As the `key.format` is included, `key.type` must be provided and the value in this example is `STRUCT`.
Note that many of the columns are in quotes, indicating they are case-sensitive. The case insensitive column named `CaseInsensitiveCol` is lowercase as `caseinsensitivecol` when the relation is created. In the parameters, the `timestamp` for this relation is also specified; queries processing data using this relation as the source refer to the `timestamp` column `ViewTime` as the event's timestamp.
```sql
CREATE STREAM "DataBase"."Schema2"."CaseSensitivePV" (
"ViewTime" BIGINT,
"userID" VARCHAR,
"PageId" VARCHAR,
"CaseSensitiveCol" BIGINT,
CaseInsensitiveCol BIGINT
)
WITH (
'topic'='case_sensitive_pageviews',
'store'='OtherStore',
'value.format'='avro',
'key.format'='protobuf',
'key.type'='STRUCT',
'timestamp'='ViewTime'
);
```
#### Create a new stream with \`NOT NULL\` column
The following creates a new stream, `users`. Two columns in this stream are defined with the `NOT NULL` constraint:
1. `registertime`
2. `contactinfo`
This means in any valid record from this stream, these two columns are not allowed to contain null values.
```sql
CREATE STREAM "users" (
registertime BIGINT NOT NULL,
userid VARCHAR,
interests ARRAY,
contactinfo STRUCT NOT NULL
)
WITH (
'topic'='users',
'key.format'='json',
'key.type'='STRUCT',
'value.format'='json'
);
```
#### Create a new stream with format-specific properties for Avro
The following creates a new stream, `usersInfo,`. The key and value of the records in this stream are in `avro` format. The stream uses subjects from a store called `sr_store` as the base Avro schema to generate Avro schema for `usersInfo`'s key and value. `users_data-key` subject is used to generate the key's Avro schema; the `users_data-value` subject is used to generate the value's Avro schema for the records written into `usersInfo.`
```sql
CREATE STREAM "usersInfo" (
registertime BIGINT NOT NULL,
userid VARCHAR,
interests ARRAY,
contactinfo STRUCT NOT NULL
)
WITH (
'topic'='usersInfo',
'key.format'='avro',
'value.format'='avro',
'avro.base.store.name' = sr_store,
'avro.base.subject.key' = 'users_data-key',
'avro.base.subject.value' = 'users_data-value'
);
```
#### Create a new stream with data in S3
The following creates a pageviews stream with data backed by an S3 store. The connector runs discovery to find new S3 objects every 15 seconds and within each interval it processes up to 25000 files.
```sql
CREATE STREAM pageviews_s3 (
viewtime BIGINT,
userid VARCHAR,
pageid VARCHAR
) WITH (
'store' = 's3_store',
's3.uri'='s3://ctan-playground-data/jsonl/',
's3.discovery.interval.seconds'=15,
's3.max.splits.per.interval'=25000,
'value.format'='jsonl'
);
```
**Notes:**
* `s3.uri` is required
* `value.format`: options\[jsonl, json]
* `s3.discovery.interval.seconds`: optional. Default = 10
* `s3.max.splits.per.interval`: optional. Default = 20,000
#### Create a new stream with key columns
The following creates a pageviews stream. The key and value of records in this stream are both in `json` format. Value consists of 3 columns: `viewtime`, `userid` and `pageid`. Key is a `STRUCT` with two fields: `userid` and `pageid` whose values are coming from the corresponding columns.
```sql
CREATE STREAM pageviews (
viewtime BIGINT,
userid VARCHAR,
pageid VARCHAR
) WITH (
'store' = 'kafka_store',
'topic' = 'pageviews',
'value.format' = 'json',
'key.format' = 'json',
'key.columns'='userid,pageid'
);
```
#### Create a new stream with key columns and value exclude columns
The following creates a pageviews stream. The key and value of records in this stream are both in `json` format. Key is a `STRUCT` with two fields: `userid` and `pageid` whose values are coming from the corresponding columns. Given that `pageid` is set to be an excluded column from value, value in each record consists of 2 columns: `viewtime`, `userid` .
```sql
CREATE STREAM pageviews (
viewtime BIGINT,
userid VARCHAR,
pageid VARCHAR
) WITH (
'store' = 'kafka_store',
'topic' = 'pageviews',
'value.format' = 'json',
'key.format' = 'json',
'key.columns'='userid,pageid',
'value.columns.exclude'='pageid'
);
```
#### Create a new stream without column definitions for Protobuf data
The following creates a pageviews Stream. The value format of records in this Stream is Protobuf. The 'pageviews' topic should exist as an entity in Deltastream and it should have a Protobuf value [descriptor](https://docs.deltastream.io/reference/sql-syntax/ddl/create-descriptor_source).
```sql
CREATE STREAM pageviewsProto WITH (
'store' = 'kafka_store',
'topic' = 'pageviewsProto',
'value.format' = 'protobut'
);
```
---
# Agent Instructions: Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter:
```
GET https://docs.deltastream.io/reference/sql-syntax/ddl/create-stream.md?ask=
```
The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.