search
ProductsDownload Community Edition

right-leftWebsocket

This page describes how to use Lenses Websocket endpoints

hashtag
Query Data via SQL

A WebSocket allows the client to request data using SQL as input.

Use the following URL to open a WebSocket connection. Depending on deployment, use ws (insecure) or wss (secure) connection:

WS | WSS ${LENSES_URL}/environment/${YOUR_ENVIRONMENT_NAME}/proxy/api/ws/v2/sql/execute
circle-info

The query is per environment, you need to add the environment name.

hashtag
Establishing a connection

To authenticate, create a service account and set the token in the header for the Authorization Bearer.

Once the connection is established, the server expects the client to send a JSON message with the following structure:

{
  "sql": String,
  "live": Optional[Boolean],
  "stats": Optional[Int]
}

For example, in Python:

import os, websockets, json, asyncio

import ssl
#Optional if using self-signed cert with the websocket:
# ssl_context = ssl._create_unverified_context()

async def execute_SQL(environmentName: str, sql: str) -> dict:

    #FOR Secure Websocket use:
    #  url = str("wss://") + os.getenv("HQ_HOST") + "/api/v1/environments/" + environmentName + "/proxy/api/ws/v2/sql/execute"
    url = str("ws://") + os.getenv("HQ_HOST") + "/api/v1/environments/" + environmentName + "/proxy/api/ws/v2/sql/execute"
    headers = {"Authorization": "Bearer " + os.getenv('HQ_TOKEN')}

    data = []
    try:
        request = {
            "sql": sql,
            "live": False,
            "stats": 2
        }

        #FOR WSS use:
        #async with websockets.connect(url, ssl=ssl_context, additional_headers=headers) as websocket:
        async with websockets.connect(url, additional_headers=headers) as websocket:
            await websocket.send(json.dumps(request))
            async for message in websocket:
                response = json.loads(message)
                if response.get('type') == 'RECORD':
                    data.append(response)

    except websockets.exceptions.WebSocketException as e:

        print(f"An error occurred: {e}")
        return None
    print(data)

asyncio.run(execute_SQL("demo", "SELECT * FROM input LIMIT 100;"))

hashtag
Query Request

Field
Description
Type

sql

The SQL query used to fetch data

string

live

If set, enables partial results for aggregation queries.

optional[boolean]

stats

Millis interval to receive query stats. If not provided the stats information is not sent

optional[int]

hashtag
Protocol

The client should wait for incoming messages and stop when EOF is received. The server sends JSON-encoded messages, the payload structure is:

Since there are multiple types of messages the server sends, the type attribute determines the data structure for the data payload. Below is the list of possible type values:

Value
Description

HEARTBEAT

To keep the connection alive, the server injects an empty message.

RECORD

The message represents a data record.

ERROR

The message contains information about an error that occurred.

STATS

The message contains information about the current execution.Only activated if the stats field in the request is present.

METADATA

The message contains a list of metadata fields each record message will contain. For example: offset, partition,timestamp, __keysize, __valueSize, valueSchemaId.

SCHEMA

Reserved.

PAGE_END

Reserved.

END

The message signals the end of the execution. The server will close the socket after this message is sent.

BADRECORD

The message contains information about the Kafka message which cannot be read. For example, if the topic expects Avro and the payload is not a valid Avro.

SENTINEL

The message contains information about the execution termination reason when query thresholds are reached.

hashtag
Heartbeat

The data the attribute is null. The client should ignore these messages.

hashtag
Error

In this case, the payload structure is:

hashtag
Stats

When requested the payload format is:

A possible payload can be:

hashtag
Metadata

The payload format is:

hashtag
End

The payload format is:

hashtag
Bad Record

The payload format is:

hashtag
Sentinel

The payload format is:

hashtag
Record

Given the type RECORD the payloads contains the data returned to the client, and it is dependent on the SQL input. The payload format is:

Value
Description

data.key

Returns the underlying Kafka message key value. Only applied when * projection is used. When a key projection is used (i.e. _key.MMSI), the value is returned as data.value.MMSI.

data.data

Will contain the output generated by the SQL projections. If * is used for the SQL projection, it will return the Kafka message value structure.

data.metadata

Returned when no aggregations are involved. It contains the Kafka message partition, offset, byte size information

data.rownum

Reserved optional long value.

hashtag
Examples

For a topic which contains nested data for both key and value.

The payload format is:

If the Kafka message key is null the response would be:

Projecting the key will not return the key attribute:

produces the following output:

Aggregating data will not return the key and metadata attributes:

PreviousAuthenticationNextEnvironments

Was this helpful?