Stream specs

Data types

In Kuiper, each column or an expression has a related data type. A data type describes (and constrains) the set of values that a column of that type can hold or an expression of that type can produce.

Below is the list of data types supported.

#Data typeDescription
1bigint
2float
3string
4datetime
5boolean
6byteaA sequence of bytes to store binary data. If the stream format is “JSON”, the bytea field must be a base64 encoded string
7arrayThe array type, can be any simple types or array and type.
8structThe complex type.

Language definitions

  1. CREATE STREAM
  2. stream_name
  3. ( column_name <data_type> [ ,...n ] )
  4. WITH ( property_name = expression [, ...] );

The supported property names.

Property nameOptionalDescription
DATASOURCEfalseThe value is determined by source type. The topic names list if it’s a MQTT data source. Please refer to related document for other sources.
FORMATtrueThe data format, currently the value can be “JSON” and “BINARY”. The default is “JSON”. Check Binary Stream for more detail.
KEYtrueReserved key, currently the field is not used. It will be used for GROUP BY statements.
TYPEtrueThe source type, if not specified, the value is “mqtt”.
StrictValidationtrueTo control validation behavior of message field against stream schema. See Strict Validation for more info.
CONF_KEYtrueIf additional configuration items are requied to be configured, then specify the config key here. See MQTT stream for more info.

Example 1,

  1. my_stream
  2. (id bigint, name string, score float)
  3. WITH ( datasource = "topic/temperature", FORMAT = "json", KEY = "id");

The stream will subscribe to MQTT topic topic/temperature, the server connection uses servers key of default section in configuration file $kuiper/etc/mqtt_source.yaml.

Example 2,

  1. demo (
  2. USERID BIGINT,
  3. FIRST_NAME STRING,
  4. LAST_NAME STRING,
  5. NICKNAMES ARRAY(STRING),
  6. Gender BOOLEAN,
  7. ADDRESS STRUCT(STREET_NAME STRING, NUMBER BIGINT),
  8. ) WITH (DATASOURCE="test/", FORMAT="JSON", KEY="USERID", CONF_KEY="demo");

The stream will subscribe to MQTT topic test/, the server connection uses settings of demo section in configuration file $kuiper/etc/mqtt_source.yaml.

Strict Validation

  1. The value of StrictValidation can be true or false.
  2. 1) True: Drop the message if the message is not satisfy with the stream definition.
  3. 2) False: Keep the message, but fill the missing field with default empty value.
  4. bigint: 0
  5. float: 0.0
  6. string: ""
  7. datetime: the current time
  8. boolean: false
  9. bytea: nil
  10. array: zero length array
  11. struct: null value

Schema-less stream

If the data type of the stream is unknown or varying, we can define it without the fields. This is called schema-less. It is defined by leaving the fields empty.

  1. schemaless_stream
  2. ()
  3. WITH ( datasource = "topic/temperature", FORMAT = "json", KEY = "id");

Schema-less stream field data type will be determined at runtime. If the field is used in an incompatible clause, a runtime error will be thrown and send to the sink. For example, where temperature > 30. Once a temperature is not a number, an error will be sent to the sink.

See Query languange element for more inforamtion of SQL language.

Binary Stream

Specify “BINARY” format for streams of binary data such as image or video streams. The payload of such streams is a block of binary data without fields. So it is required to define the stream as only one field of bytea. In the below example, the payload will be parsed into image field of demoBin stream.

  1. demoBin (
  2. image BYTEA
  3. ) WITH (DATASOURCE="test/", FORMAT="BINARY");

If “BINARY” format stream is defined as schemaless, a default field named self will be assigned for the binary payload.