# Stream Data Source API (/docs/symetryml-rest-client/rest-documentation/stream-data-source-api) Data source described in the [Data Source API](/docs/symetryml-rest-client/rest-documentation/data-source-api) section all are *finite* type of data source int he sense that they all have a *finite* number of rows. SymetryML also support *Stream Data Source* that in theory can contain an infinite number of rows. Currently, both Kafka and NATS Stream Data Sources are supported. ## Kafka Streams [#kafka-streams] To use a Kafka stream data source, create a JSON data structure described in [DSInfo](/docs/symetryml-rest-client/rest-documentation/appendix-a-json-data-structure-schema#dsinfo-json) that contains the fields in the table [below](/docs/symetryml-rest-client/rest-documentation/stream-data-source-api#fields-required-to-create-a-stream-dsinfo-data-structure). Please note that for SymetryML to be able to read the data in your topic it needs to use both the `KafkaAvroDeserializer` as well as the Kafka Registry. ### Fields Required to Create a Stream DSInfo Data Structure [#fields-required-to-create-a-stream-dsinfo-data-structure] | Field | Description | | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **type** | **Type of Data Source** - `kafka` for Kafka streams or `nats` for NATS streams | | **name** | Name of the data source. | | **path** | path of the file / entity | | **info** | **Hash Map** Containing Information needed to create a Stream. Please consult [Kafka Stream Additional Information](/docs/symetryml-rest-client/rest-documentation/stream-data-source-api#kafka-stream-additional-information) or [NATS Stream Additional Information](#nats-stream-additional-information) for details | ## Kafka Stream Additional Information [#kafka-stream-additional-information] | Key | Required / Optional | Description | | ----------------------------------------- | ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | **bootstrap.servers** | Required | Kafka Stream configuration parameter. | | **schema.registry.url** | Required | Kafka Stream configuration parameter. | | **kafka.topic** | Required | Kafka Stream configuration parameter. | | **kafka.partitions** | Optional | Kafka Stream configuration parameter. Default to all partitions. The list of partition to use must be defined as a list of comma separated integers. For instance `0,2,4,6,8` or `0,1,2,6,7,10,12`. | | **kafka\_stream\_time\_between\_persist** | Optional | Default to 300 seconds, that is 5 minutes. How long to wait between when the Symetry Project will persist its state. To disable this pass -1. | | **enable.auto.commit** | Optional | Kafka Stream configuration parameter. Default to true | | **auto.commit.interval.ms** | Optional | Kafka Stream configuration parameter. Default to 1000 ms. | | **auto.offset.reset** | Optional | Kafka Stream configuration parameter. Default to *earliest* | | **Any other kafka parameter** | Optional | Any Kafka parameters can be used as well. One needs to prefix them with `sml.kafka.` e.g. `sml.kafka.client.dns.lookup` or `sml.kafka.fetch.min.bytes` | ## NATS Streams [#nats-streams] NATS is a lightweight, high-performance messaging system that provides both publish-subscribe and distributed queueing capabilities. SymetryML can connect to NATS servers to consume streaming data in real-time. ### NATS Stream Additional Information [#nats-stream-additional-information] To use a NATS stream data source, create a JSON data structure described in [DSInfo](/docs/symetryml-rest-client/rest-documentation/appendix-a-json-data-structure-schema#dsinfo-json) with `type` set to `nats` and include the following fields in the `info` map: | Key | Required / Optional | Description | | ----------------------------- | ------------------- | ---------------------------------------------------------------------------------------------------------------------------------- | | **nats\_hosts** | Required | NATS server URL(s) to connect to. For example: `nats://localhost:4222`. Multiple hosts can be specified as a comma-separated list. | | **nats.subject** | Required | NATS subject to subscribe to. The subject determines which messages the stream will receive. | | **data.format** | Required | Format of the data in the NATS messages. Supported values: `json`, `csv`, or `protobuf` | | **protobuf.schema** | Required (protobuf) | Protobuf schema definition. Required only when `data.format` is set to `protobuf`. | | **protobuf.msg.type.name** | Required (protobuf) | Protobuf message type name to deserialize. Required only when `data.format` is set to `protobuf`. | | **nats\_sec\_user** | Optional | Username for NATS authentication. Use for user/password authentication. | | **nats\_sec\_passwd** | Optional | Password for NATS authentication. Use in combination with `nats_sec_user`. | | **nats\_sec\_token** | Optional | Token for NATS authentication. Use for token-based authentication. | | **nats\_sec\_nkey\_seed** | Optional | NKey seed for NATS authentication. Use for NKey-based authentication. | | **nats\_req\_timeout** | Optional | Request timeout in seconds. Default is `30` seconds. | | **nats\_stream\_max\_memory** | Optional | Maximum memory (in bytes) for the NATS stream buffer. Default is `1073741824` (1 GB). | ### NATS Data Formats [#nats-data-formats] NATS streams support three data formats: 1. **JSON** - Messages are expected to be in JSON format with attribute names matching the project schema 2. **CSV** - Messages are expected to be comma-separated values (or using custom delimiters defined with additional CSV options) 3. **Protobuf** - Messages are serialized using Protocol Buffers. Requires both `protobuf.schema` and `protobuf.msg.type.name` to be specified ### Example NATS DSInfo (JSON Format) [#example-nats-dsinfo-json-format] ```json { "type": "nats", "name": "my-nats-stream", "info": { "nats_hosts": "nats://localhost:4222", "nats.subject": "sensor.data", "data.format": "json" } } ``` ### Example NATS DSInfo (Protobuf Format) [#example-nats-dsinfo-protobuf-format] ```json { "type": "nats", "name": "my-protobuf-stream", "info": { "nats_hosts": "nats://nats-server:4222", "nats.subject": "telemetry.metrics", "data.format": "protobuf", "protobuf.schema": "syntax = \"proto3\"; message Metrics { double temperature = 1; double pressure = 2; }", "protobuf.msg.type.name": "Metrics" } } ``` ## Stream Data Source Encryption [#stream-data-source-encryption] Same as one need to encrypt normal data source, stream data source information needs to be encrypted. Please consult the [Data Source Encryption](/docs/symetryml-rest-client/rest-documentation/data-source-api#data-source-encryption) for details. ## Stream Data Source Create [#stream-data-source-create] This API function creates a new stream data source and attach it to a SymetryML project - the owner project. Once created the new stream data source will continuously pull data from Kafka and then push the new data tuple into SymetryML Project in a streaming fashion. ### URL [#url] ``` POST /{cid}/projects/{pid}/streams/create [Body=DSInfo (encrypted)] ``` | Parameter | Required / Optional | Description | | ----------------- | ------------------- | --------------------------------------------------------------- | | **fromBeginning** | Optional | if true then start streaming data from beginning of the stream. | ### HTTP Responses [#http-responses] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ------------------------------------------------------------ | | **202** | CREATED | Success. | | **409** | CONFLICT | A stream data source with the specified name already exists. | ## Stream Data Source Browse [#stream-data-source-browse] This methods allows you to browse available stream on your stream server. For Kafka this means listing topic that are available. ### URL [#url-1] ``` POST /{cid}/streams/browse [Body=DSInfo (encrypted)] ``` ### HTTP Responses [#http-responses-1] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success. | ### HTTP Response Entity [#http-response-entity] | HTTP Response Entity | Description | | ------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------- | | [**DSListingResponse**](/docs/symetryml-rest-client/rest-documentation/appendix-a-json-data-structure-schema#dslistingresponse-json) | Contains listing information from the streaming server. | ## Stream Data Source Preview [#stream-data-source-preview] This methods allows you to preview a sample of the data available on a given stream. ### URL [#url-2] ``` POST /{cid}/streams/preview [Body=DSInfo (encrypted)] ``` ### HTTP Responses [#http-responses-2] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success. | ### HTTP Response Entity [#http-response-entity-1] | HTTP Response Entity | Description | | -------------------------------------------------------------------------------------------------------------------- | ------------------------------------------ | | [**DataFrame**](/docs/symetryml-rest-client/rest-documentation/appendix-a-json-data-structure-schema#dataframe-json) | Dataframe containing a preview of the data | ## Stream Data Source Metrics [#stream-data-source-metrics] This rest endpoint return information about a stream. Number of rows processed, tuples / secs processed, etc… Information varies with the type of the stream. ### URL [#url-3] ``` GET /{cid}/projects/{pid}/streams/{sid}/metrics ``` ### HTTP Responses [#http-responses-3] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success. | ### HTTP Response Entity [#http-response-entity-2] | HTTP Response Entity | Description | | ----------------------------------------------------------------------------------------------------------- | ------------------------------------ | | [**KSVSMap**](/docs/symetryml-rest-client/rest-documentation/appendix-a-json-data-structure-schema#ksvsmap) | A map with key, value as string pair | ## Stream Data Source Start [#stream-data-source-start] This rest endpoint start / resume a stream data source. That is start pulling data and push it into the owner SymetryML project. ### URL [#url-4] ``` GET /{cid}/projects/{pid}/streams/{sid}/start ``` | Parameter | Required/Optional? | Description | | ----------------- | ------------------ | --------------------------------------------------------------- | | **fromBeginning** | Optional | if true then start streaming data from beginning of the stream. | ### HTTP Responses [#http-responses-4] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success. | ## Stream Data Source Stop [#stream-data-source-stop] This rest endpoint stop a stream data source. Data will not be pushed anymore to the owner SymetryML project. ### URL [#url-5] ``` GET /{cid}/projects/{pid}/streams/{sid}/stop ``` ### HTTP Responses [#http-responses-5] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success. | ## Delete a Stream [#delete-a-stream] Delete a stream from a project. If the stream is running it will first be stopped. ### URL [#url-6] ``` DELETE /symetry/rest/{cid}/projects/{pid}/streams/{sid} ``` ### Canonical URL Parameters [#canonical-url-parameters] | Parameter | Description | | --------- | ----------- | | **cid** | Customer ID | | **pid** | Project ID | | **sid** | Stream ID | ### HTTP Responses [#http-responses-6] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success | ## Stream Data Source Error Log [#stream-data-source-error-log] This rest endpoint return a list of error for a given stream. Since stream happen asynchronously in the background, it allows to check for any problems with a given stream ### URL [#url-7] ``` GET /{cid}/projects/{pid}/streams/{sid}/errorLog ``` ### HTTP Responses [#http-responses-7] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success. | ### HTTP Response Entity [#http-response-entity-3] | HTTP Response Entity | Description | | ----------------------------------------------------------------------------------------------------------------- | ------------------------------- | | [**StringList**](/docs/symetryml-rest-client/rest-documentation/appendix-a-json-data-structure-schema#stringlist) | A list of error for that stream | ## Stream Data Source List [#stream-data-source-list] This rest endpoint return a list of streams name that belong to a given project for a given user. ### URL [#url-8] ``` GET /{cid}/projects/{pid}/streams/list ``` ### HTTP Responses [#http-responses-8] | HTTP Status Code | HTTP Status Message | Description | | ---------------- | ------------------- | ----------- | | **200** | OK | Success. | ### HTTP Response Entity [#http-response-entity-4] | HTTP Response Entity | Description | | ----------------------------------------------------------------------------------------------------------------- | --------------------------------- | | [**StringList**](/docs/symetryml-rest-client/rest-documentation/appendix-a-json-data-structure-schema#stringlist) | A list of stream data source name |