CREATE STREAM - Kinetica Docs

Creates a new data stream (natively, a table monitor), which publishes changes in a given table to a target.

CREATE STREAM Syntax

CREATE STREAM [<stream schema name>.]<stream name>
ON
<
    [TABLE] [<table schema name>.]<table name>
    |
    [QUERY] (<query>)
>
[
    REFRESH
    <
        ON CHANGE |
        EVERY <number> <SECOND[S] | MINUTE[S] | HOUR[S] | DAY[S]> [STARTING AT '<YYYY-MM-DD [HH:MM[:SS]]>']
    >
]
[WHERE <filter expression>]
WITH OPTIONS
(
    < DATASINK_NAME = '<data sink name>' | DESTINATION = '<destination>' >,
    <option name> = '<option value>'[,...]
)

The monitored table can be one of:

a table (external tables not supported)
a materialized view (insert monitoring only)

The target can be one of:

an external Apache Kafka broker
an external webhook
a local database table

By default, a stream will publish inserted records to the target. A stream can alternatively be configured to monitor for updates or deletes and publish the corresponding record counts. Streams that monitor for inserts can have an expression applied to only have specific inserts of interest published. An existing data sink can be referenced in creating a stream, as the target of the streamed data. Only local tables or unauthenticated external targets may be used if not using an existing data sink; however, in these cases, a data sink will automatically be created to support the stream. When the source table of a stream is altered or dropped, the stream will also be dropped.

Parameters

Name of the schema that will contain the created stream; if no schema is specified, the stream will be created in the user’s default schema

Name of the stream to create; must adhere to the supported naming criteria

TABLE

Optional keyword for clarity, in creating a stream on a table.

Name of the schema containing the table or materialized view to monitor.

Name of the table or materialized view to monitor for changes.

QUERY

Optional keyword for clarity, in creating a stream on a join.

<query>

Join query to which the stream will be applied, using the following form:{{< code “sql” “linenos=true” “Query Syntax” >}} SELECT * FROM <delta_table> dt LEFT SEMI JOIN <lookup_table> lt ON <join_clause> {{< /code >}}Here, <delta_table> is the data table being monitored and <lookup_table> contains the criteria for whether a <delta_table> record should be streamed or not, as determined by the given <join_clause>.

The marked delta table must be a regular table—it cannot be any of the following:

The join must also meet sharding requirements for joins—either both tables need to have shard keys on their equality-based joined columns, or at least one of the tables needs to be replicated.See CREATE TABLE for syntax for creating a replicated table or a sharded table with a shard key column.

See below for a table-driven geofence example using a query-based stream.

REFRESH

Specifies the reporting scheme for monitored changes. The following schemes are available:

Constant	Description
`ON CHANGE`	Will cause notifications to be streamed any time a record is added, modified, or deleted from the monitored table
`EVERY`	Allows specification of an interval in seconds, minutes, hours, or days, with the optional specification of a starting time at which the first monitor interval will run; if no start time is specified, the default will be an interval’s worth of time from the point at which the stream was created

Boolean expression that can be used to monitor for only a specific set of inserts.

WITH OPTIONS

Indicator that a comma-delimited list of configuration option/value assignments will follow. See Options for the full list of options.

One of either DATASINK_NAME or DESTINATION is required.

Examples

To create a stream, kin_stream, that publishes inserts into the order_stream table of orders over $10,000 via the kin_dsink data sink:

CREATE STREAM Example

CREATE STREAM example.kin_stream ON example.order_stream
WHERE cost > 10000.00
WITH OPTIONS (DATASINK_NAME = 'example.kin_dsink')

To create a stream, kin_stream, that publishes inserts into the order_stream table of orders, placed within a given string-literal geofence, via the kin_dsink data sink:

CREATE STREAM Single Geometry Geofence Example

CREATE STREAM example.kin_stream ON example.order_stream
REFRESH ON CHANGE
WHERE STXY_INTERSECTS(lon, lat, 'POLYGON((
    -108.97203757374318 40.945055425026034,
    -101.72106101124318 40.945055425026034,
    -101.72106101124318 36.98324941198095,
    -108.97203757374318 36.98324941198095,
    -108.97203757374318 40.945055425026034
    ))')
WITH OPTIONS (DATASINK_NAME = 'example.kin_dsink')

To create a stream, kin_stream, that publishes inserts into the order_stream table of orders, placed within a table-driven geofence defined in order_location, via the kin_dsink data sink; noting that the order_location table in this example is replicated:

CREATE STREAM Table-Driven Geofence Example

CREATE STREAM example.kin_stream
ON QUERY
(
	SELECT *
	FROM example.order_stream os
	LEFT SEMI JOIN example.order_location ol
		ON STXY_INTERSECTS(os.lon, os.lat, ol.geom)
)
REFRESH ON CHANGE
WITH OPTIONS (DATASINK_NAME = 'example.kin_dsink')

To create a stream, kin_stream, that inserts into the local order_target table any orders over $10,000 inserted into the order_stream table:

CREATE STREAM to Local Table Example

CREATE STREAM example.kin_stream ON example.order_stream
WHERE cost > 10000.00
WITH OPTIONS (DESTINATION = 'TABLE://example.order_target')

Options

DATASINK_NAME

Consumer: AnyData sink object to use to identify the consumer of this stream; mutually exclusive with destination

DESTINATION

Consumer: AnyLocation of the target when not using a data sink; the URL of the consumer to connect to in this format:

<consumer>://<target>[:<port>]

Supported consumers include:

Consumer	Description
`http`	Unsecured webhook
`https`	Secured webhook
`kafka`	Apache Kafka broker
`table`	Local Kinetica table, where `target` is the name of the target table in the following format: `[<schema name>.]<table name>`

EVENT

Consumer: AnySpecifies the type of event to monitor

Event Type	Description
`insert`	(default) Monitor inserts and publish them to the target
`update`	Monitor updates and publish their counts to the target
`delete`	Monitor deletes and publish their counts to the target

INCREASING_COLUMN

Consumer: AnyColumn of the monitored table that contains ever-increasing values for new data; e.g., a timestamp or sequence ID column

If such a column can be identified, the performance of the stream will improve.

MAX_CONSECUTIVE_FAILURES

Consumer: AnyMaximum number of failed notification attempts before automatically suspending the the stream. The default, -1, disables auto-suspension.

FAILED_NOTIFICATIONS_TABLE_NAME

Consumer: http https kafkaName of the table to which failed stream notifications are written and from which they are replayed when the stream is re-enabled from a suspended state. The default is no table for storing/replaying failed stream notifications.

KAFKA_TOPIC_NAME

Consumer: kafkaKafka topic to write to, if the target is a Kafka broker

Consumer-Specific Syntax

Several configurations across multiple consumers are supported.

CREATE STREAM [<stream schema name>.]<stream name>
ON [TABLE] [<table schema name>.]<table name>
[WHERE <filter expression>]
WITH OPTIONS (DATASINK_NAME = '[<data sink schema name>.]<data sink name>')

CREATE STREAM [<stream schema name>.]<stream name>
ON [TABLE] [<table schema name>.]<table name>
[WHERE <filter expression>]
WITH OPTIONS (DESTINATION = 'http://<host>[:<port>]')

CREATE STREAM [<stream schema name>.]<stream name>
ON [TABLE] [<table schema name>.]<table name>
[WHERE <filter expression>]
WITH OPTIONS (DESTINATION = 'https://<host>[:<port>]')

CREATE STREAM [<stream schema name>.]<stream name>
ON [TABLE] [<table schema name>.]<table name>
[WHERE <filter expression>]
WITH OPTIONS
(
	DESTINATION = 'kafka://<host>:<port>',
	KAFKA_TOPIC_NAME = '<kafka topic name>'
)

CREATE STREAM [<stream schema name>.]<stream name>
ON [TABLE] [<source table schema name>.]<source table name>
[WHERE <filter expression>]
WITH OPTIONS (DESTINATION = 'table://[<target table schema name>.]<target table name>')

​Parameters

​Examples

​Options

​Consumer-Specific Syntax

Parameters

Examples

Options

Consumer-Specific Syntax