CREATE STREAM
Syntax
Description
A Stream is a sequence of immutable, partitioned, and partially ordered events (we use events and records synonymously). A Stream is a relational representation of data in streaming Store, such as the data in a Kafka topic or a Kinesis stream. The records in a Stream are independent of each other, meaning there is no correlation between two records in a Stream. A Stream declares the schema of the records, which includes the column name along with the column type and optional constraints. A Stream is a type of #_relation. Each Relation belongs to a #_schema in a Database, so the fully qualified name of the Relation would be <database>.<schema>.<relation>
.
Arguments
stream_name
Specifies the name of the new Stream. For case-sensitive names, the name must be wrapped in double quotes, otherwise, the lowercased name will be used.
column_name
The name of a column to be created in the new Stream. For case-sensitive names, the name must be wrapped in double quotes, otherwise, the lowercased name will be used.
data_type
The data type of the column. This can include array specifiers. For more information on the data types supported by DeltaStream, refer to the Data Types 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
Parameter Name | Description |
---|---|
| |
| |
| |
| 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 |
| The format to use for Required: No
Default value: |
Kafka Specific Parameters
Parameters to be used if the associated Store is type KAFKA
:
Parameter Name | Description |
---|---|
| The number of partitions to use when creating the Topic, if applicable. If the topic already exist, then this value must be equal to the number of partitions in the existing Kafka Topic. Required: Yes, unless topic already exists Default value: Leftmost source Relation topic's partition count Type: Integer Valid values: [1, ...] |
| 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 Topic. Required: Yes, unless topic already exists Default values: Leftmost source Relation topic's replica count Type: Integer Valid values: [1, ...] |
| |
| Required: No, unless |
| The fault tolerance guarantees applied when producing to this Stream. Required: No
Default value:
|
Kinesis Specific Parameters
Parameters to be used if the associated Store is type KINESIS
:
Parameter Name | Description |
---|---|
| 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: |
Examples
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
:
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
:
Create a new Stream without an existing Topic
The following creates a new Stream visit_count
, and since its corresponding topic doesn't exist in the store kinesis_store
, it requires an additional topic parameter, i.e. topic.shards
to create the new Kinesis Data Stream pv_count
in the store:
Create a new Stream for an existing Topic
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. Since the key.format
is included, it is required that key.type
is also provided and the value in this example is STRUCT<pageid VARCHAR>
. Note that many of the columns are in quotes, indicating they are case-sensitive. The case insensitive column named CaseInsensitiveCol
will be lowercase as caseinsensitivecol
when the Relation is created. In the parameters, the timestamp
for this Relation is also specified so queries processing data using this Relation as the source will refer to the timestamp
column ViewTime
as the event's timestamp.
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: registertime
and contactinfo
. This means in any valid record from this Stream, these two columns are not allowed to contain null values.
Last updated