ksql-python-ng 0.29.0.1

ksql

This method can be used for some KSQL features which are not supported via other specific methods like query, create_stream or create_stream_as. The following example shows how to execute the show tables statement:

client.ksql('show tables')

• Example Response [{'tables': {'statementText': 'show tables;', 'tables': []}}]

query

It will execute sql query and keep listening streaming data.

client.query('select * from table1')

This command returns a generator. It can be printed e.g. by reading its values via next(query) or a for loop. Here is a complete example:

from ksql import KSQLAPI
client = KSQLAPI('http://localhost:8088')
query = client.query('select * from table1')
for item in query: print(item)

• Example Response

  {"queryId":"44d8413c-0018-423d-b58f-3f2064b9a312","columnNames":["ORDER_ID","TOTAL_AMOUNT","CUSTOMER_NAME"],"columnTypes":["INTEGER","DOUBLE","STRING"]}
  [3,43.0,"Palo Alto"]
  [3,43.0,"Palo Alto"]
  [3,43.0,"Palo Alto"]

Query with HTTP/2

Execute queries with the new /query-stream endpoint. Documented here

To execute a sql query use the same syntax as the regular query, with the additional use_http2=True parameter.

client.query('select * from table1', use_http2=True)

A generator is returned with the following example response

{"queryId":"44d8413c-0018-423d-b58f-3f2064b9a312","columnNames":["ORDER_ID","TOTAL_AMOUNT","CUSTOMER_NAME"],"columnTypes":["INTEGER","DOUBLE","STRING"]}
[3,43.0,"Palo Alto"]
[3,43.0,"Palo Alto"]
[3,43.0,"Palo Alto"]

To terminate the query above use the close_query call. Provide the queryId returned from the query call.

client.close_query("44d8413c-0018-423d-b58f-3f2064b9a312")

Insert rows into a Stream with HTTP/2

Uses the new /inserts-stream endpoint. See documentation

rows = [
        {"ORDER_ID": 1, "TOTAL_AMOUNT": 23.5, "CUSTOMER_NAME": "abc"},
        {"ORDER_ID": 2, "TOTAL_AMOUNT": 3.7, "CUSTOMER_NAME": "xyz"}
    ]

results = self.api_client.inserts_stream("my_stream_name", rows)

An array of object will be returned on success, with the status of each row inserted.

create_stream/ create_table

client.create_stream(table_name=table_name,
                     columns_type=columns_type,
                     topic=topic,
                     value_format=value_format)

Options

• Responses

If create table/stream succeed:

return True

If failed:

raise a CreateError(respose_from_ksql_server)

create_stream_as

a simplified api for creating stream as select

client.create_stream_as(table_name=table_name,
                        select_columns=select_columns,
                        src_table=src_table,
                        kafka_topic=kafka_topic,
                        value_format=value_format,
                        conditions=conditions,
                        partition_by=partition_by,
                        **kwargs)

CREATE STREAM <table_name>
[WITH ( kafka_topic=<kafka_topic>, value_format=<value_format>, property_name=expression ... )]
AS SELECT  <select_columns>
FROM <src_table>
[WHERE <conditions>]
PARTITION BY <partition_by>];

Options

upload

Run commands from a .ksql file. Can only support ksql commands and not streaming queries.

from ksql.upload import FileUpload
pointer = FileUpload('http://ksql-server:8080')
pointer.upload('rules.ksql')

Options

• Responses

If ksql-commands succesfully executed:

return (List of server response for all commands)

If failed:

raise the appropriate error

More Options

There are more properties (partitions, replicas, etc…) in the official document.

KSQL Syntax Reference

• Responses

If create table/stream succeed:

return True

If failed:

raise a CreateError(respose_from_ksql_server)