CREATE CHANGELOG
Syntax
Description
A Changelog is a sequence of partitioned and partially ordered events (we use events and records synonymously). A Changelog is a relational representation of data in the streaming Stores, such as the data in a Apache Kafka topic or an Amazon Kinesis stream.
A Changelog defines a PRIMARY KEY
for records that is used to represent the change over time for records with the same PRIMARY KEY
. Records in a Changelog correlate with each other based on the PRIMARY KEY
. This means that a record in a Changelog either is an insert record if it’s the first time the record with the given PRIMARY KEY
is appended to the Changelog or upsert records if a previous record with the same PRIMARY KEY
has been inserted into the Changelog.
A Changelog 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
changelog_name
This specifies the name of the new Changelog. For case-sensitive names, the name must be wrapped in double quotes; otherwise, the lowercase name will be used.
column_name
This is the name of a column to be created in the new Changelog. For case-sensitive names, the name must be wrapped in double quotes; otherwise, the lowercase name will be used.
data_type
This refers to the data type of the column, which 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.
PRIMARY KEY (column_name [, …])
The PRIMARY KEY
constraint specifies that column(s) of a table can contain only unique (non-duplicate), non-null values.
WITH (changelog_parameter = value [, … ])
This clause specifies Changelog Parameters.
Changelog Parameters
Parameter Name | Description |
---|---|
| Name of the Entity that has the data for this Changelog. If the Entity doesn’t exist, an Entity with this name is created in the corresponding Required: No
Default value: Lowercase |
| Name of the store that hosts the Entity for this Changelog. Required: No Default value: User’s default Store name Type: String Valid values: See LIST STORES. |
| Format of message value in the Entity. See Data Formats (Serialization) for more information regarding serialization formats. Required: Yes
Type: String
Valid values: |
| Name of the column in the Changelog 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 Required: No
Default value: Record’s timestamp
Type: String
Valid values: Must be of type |
| 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 Entity, if applicable. If the Entity already exists, then this value must be equal to the number of partitions in the existing Kafkan Entity. Required: Yes, unless Entity already exists Default value: Leftmost source Relation Entity’s partition count Type: Integer Valid values: [1, ...] |
| The number of replicas to use when creating the Entity, if applicable. If the Entity already exists, then this value must be equal to the number of replicas in the existing Kafkan Entity. Required: Yes, unless Entity already exists Default values: Leftmost source Relation Entity's replica count Type: Integer Valid values: [1, ...] |
| Format of the message key in the Entity. This value can be the same as or different from the Required: No, unless |
| Required: No, unless |
| The fault tolerance guarantees applied when producing to this Changelog. 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 Entity, if applicable. If the Entity already exists, then this value must be equal to the number of shards in the existing Kinesis Data Stream. Required: Yes, unless Entity already exists
Default values: Leftmost source Relation topic’s shard count
Type: Integer
Valid values: [1, ...]
Alias: |
Kinesis stores provide a delivery guarantee of at_least_once
when producing events into a sink Entity.
Format Specific Parameters
Avro
Parameters to be used when writing records into a Changelog if associated key.format
or value.format
is avro
and the default Avro schema generation needs to 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 the column's, then the field's definition from the base is used in the generated schema. This includes retaining base schema's
doc
andlogicalType
for the field.if the base schema has a field with the same name as the column's, but 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.
Currently supported Schema Registries are Confluent Cloud and Confluent Platform.
Known Limitation: Confluent Schema Registry must use the default TopicNameStrategy for creating subject names.
Check CREATE SCHEMA_REGISTRY for more details.
Parameter Name | Description |
---|---|
| 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 Changelog's key and/or value. Required: No Default values: Current session's store name Type: Identifier Valid values: See LIST STORES. |
| Name of the subject in the Schema Registry to obtain the base schema for generating Avro schema for Changelog's key.
Required: No, unless |
| Name of the subject in the Schema Registry to obtain the base schema for generating Avro schema for Changelog's value columns.
Required: No, unless |
Examples
Create a new Changelog
The following creates a new Changelog, user_last_page
. This Changelog reads from a topic named pageviews
and has a value.format
of JSON
. Note that this query also specifies userid
as the PRIMARY KEY
for the Changelog:
Create a new Changelog for an existing Entity
The following creates a new users
Changelog for the existing users
Entity in the current Store. This DDL implies that the name of the Changelog should be used as the name of the Entity that hosts the records. This DDL also implies the original structure for the users
Entity with a PRIMARY KEY
for updates:
Create a new Changelog with a multi-column Primary Key
The following creates a new Changelog, pagevisits
. This Changelog reads from an Entity named pageviews
and has a value.format
of JSON
. Note that this query also specifies (userid, pageid)
as the PRIMARY KEY
for the Changelog:
Create a new Changelog with specifying key and timestamp
The following creates a new Changelog, LatestPageVisitor
, in the Database, DataBase
, and Schema, Schema2
. This Changelog reads from a topic named case_sensitive_pageviews
from the store OtherStore
and has a value.format
of Avro and a key.format
of PROTOBUF
. Since the key.format
is included, it is required that the key.type
is also provided and the value in this example is STRUCT<pageid VARCHAR>
. Note that this query also specifies PageId
as the PRIMARY KEY
for the Changelog, and 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 Changelog specifying Kafka delivery guarantee
The following creates a new Changelog, user_exactly_once
. This Changelog reads from an Entity named users
and has a delivery.guarantee
of exactly_once
. By specifying the delivery.guarantee
, we are overriding the default value of at_least_once
. A user may want to use this configuration if their application can tolerate higher latencies but cannot tolerate duplicate records. When this Changelog is used as the sink in an INSERT INTO query, the query will use the delivery.guarantee
specified here.
Create a new Changelog with `NOT NULL` column
The following creates a new Changelog, users_log
. Two columns in this Changelog are defined with the NOT NULL
constraint: registertime
and contactinfo
. This means in any valid record from this Changelog, these two columns are not allowed to contain null values.
Create a new Changelog with format specific properties for Avro
The following creates a new Changelog, usersInfo,
whose records' key and value are in avro
format. It 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 key's Avro schema and users_data-value
subject is used to generate value's Avro schema for the records written into usersInfo.
Last updated