A python wrapper for the KSQL REST API. Easily interact with the KSQL REST API using this library.
Supported KSQLDB version: 0.10.1+ Supported Python version: 3.5+
.. image:: https://travis-ci.org/bryanyang0528/ksql-python.svg?branch=master :target: https://travis-ci.org/bryanyang0528/ksql-python
.. image:: https://codecov.io/gh/bryanyang0528/ksql-python/branch/master/graph/badge.svg :target: https://codecov.io/gh/bryanyang0528/ksql-python
.. image:: https://pepy.tech/badge/ksql :target: https://pepy.tech/project/ksql
.. image:: https://pepy.tech/badge/ksql/month :target: https://pepy.tech/project/ksql/month
.. image:: https://img.shields.io/badge/license-MIT-yellow.svg :target: https://github.com/bryanyang0528/ksql-python/blob/master/LICENSE
.. code:: bash
pip install ksql
Or
.. code:: bash
git clone https://github.com/bryanyang0528/ksql-python
cd ksql-python
python setup.py install
Setup for KSQL
This is the GITHUB page of KSQL. https://github.com/confluentinc/ksql
If you have installed open source Confluent CLI (e.g. by installing Confluent Open Source or Enterprise Platform), you can start KSQL and its dependencies with one single command:
.. code:: bash
confluent start ksql-server
Setup for ksql-python API
.. code:: python
from ksql import KSQLAPI
client = KSQLAPI('http://ksql-server:8088')
.. code:: python
import logging
from ksql import KSQLAPI
logging.basicConfig(level=logging.DEBUG)
client = KSQLAPI('http://ksql-server:8088')
.. code:: python
from ksql import KSQLAPI
client = KSQLAPI('http://ksql-server:8088', api_key="your_key", secret="your_secret")
Options
+---------------+-----------+------------+--------------------------------------------------------------+
| Option | Type | Required | Description |
+===============+===========+============+==============================================================+
| ``url`` | string | yes | Your ksql-server url. Example: ``http://ksql-server:8080`` |
+---------------+-----------+------------+--------------------------------------------------------------+
| ``timeout`` | integer | no | Timout for Requests. Default: ``5`` |
+---------------+-----------+------------+--------------------------------------------------------------+
| ``api_key`` | string | no | API Key to use on the requests |
+---------------+-----------+------------+--------------------------------------------------------------+
| ``secret`` | string | no | Secret to use on the requests |
+---------------+-----------+------------+--------------------------------------------------------------+
Main Methods
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:
.. code:: python
client.ksql('show tables')
[{'tables': {'statementText': 'show tables;', 'tables': []}}]
query ^^^^^
It will execute sql query and keep listening streaming data.
.. code:: python
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:
.. code:: python
from ksql import KSQLAPI client = KSQLAPI('http://localhost:8088') query = client.query('select * from table1') for item in query: print(item)
Example Response
::
{"row":{"columns":[1512787743388,"key1",1,2,3]},"errorMessage":null} {"row":{"columns":[1512787753200,"key1",1,2,3]},"errorMessage":null} {"row":{"columns":[1512787753488,"key1",1,2,3]},"errorMessage":null} {"row":{"columns":[1512787753888,"key1",1,2,3]},"errorMessage":null}
Query with HTTP/2
^^^^^^^^^^^^^^^^^
Execute queries with the new /query-stream
endpoint. Documented here <https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-rest-api/streaming-endpoint/#executing-pull-or-push-queries>
_
To execute a sql query use the same syntax as the regular query, with the additional use_http2=True
parameter.
.. code:: python
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.
.. code:: python
client.close_query("44d8413c-0018-423d-b58f-3f2064b9a312")
Insert rows into a Stream with HTTP/2 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Uses the new /inserts-stream
endpoint. See documentation <https://docs.ksqldb.io/en/0.10.0-ksqldb/developer-guide/ksqldb-rest-api/streaming-endpoint/#inserting-rows-into-an-existing-stream>
_
.. code:: python
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.
Simplified API
create_stream/ create_table
^^^^^^^^^^^^^^^^^^^^^^^^^^^
.. code:: python
client.create_stream(table_name=table_name,
columns_type=columns_type,
topic=topic,
value_format=value_format)
Options
^^^^^^^
+-----------------+-----------+----------+--------------------------------------------------------------+
| Option | Type | Required | Description |
+=================+===========+==========+==============================================================+
| ``table_name`` | string | yes | name of stream/table |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``columns_type``| list | yes | ex:``['viewtime bigint','userid varchar','pageid varchar']`` |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``topic`` | string | yes | Kafka topic |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``value_format``| string | no | ``JSON`` (Default) or ``DELIMITED`` or ``AVRO`` |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``key`` | string | for Table| Key (used for JOINs) |
+-----------------+-----------+----------+--------------------------------------------------------------+
- 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
.. code:: python
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)
.. code:: sql
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
^^^^^^^
+-------------------+-----------+----------+--------------------------------------------------------------+
| Option | Type | Required | Description |
+===================+===========+==========+==============================================================+
| ``table_name`` | string | yes | name of stream/table |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``select_columns``| list | yes | you can select ``[*]`` or ``['columnA', 'columnB']`` |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``src_table`` | string | yes | name of source table |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``kafka_topic`` | string | no | The name of the Kafka topic of this new stream(table). |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``value_format`` | string | no | ``DELIMITED``, ``JSON``(Default) or ``AVRO`` |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``conditions`` | string | no | The conditions in the where clause. |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``partition_by`` | string | no | Data will be distributed across partitions by this column. |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``kwargs`` | pair | no | please provide ``key=value`` pairs. Please see more options. |
+-------------------+-----------+----------+--------------------------------------------------------------+
KSQL JOINs
KSQL JOINs between Streams and Tables are not supported yet via explicit methods, but you can use the ksql
method for this like the following:
.. code:: python
client.ksql("CREATE STREAM join_per_user WITH (VALUE_FORMAT='AVRO', KAFKA_TOPIC='join_per_user') AS SELECT Time, Amount FROM source c INNER JOIN users u on c.user = u.userid WHERE u.USERID = 1")
FileUpload
upload
^^^^^^^^^^^^^^^^^^^^^^^^^^^
Run commands from a .ksql file. Can only support ksql commands and not streaming queries.
.. code:: python
from ksql.upload import FileUpload
pointer = FileUpload('http://ksql-server:8080')
pointer.upload('rules.ksql')
Options
^^^^^^^
+-----------------+-----------+----------+--------------------------------------------------------------+
| Option | Type | Required | Description |
+=================+===========+==========+==============================================================+
| ``ksqlfile`` | string | yes | name of file containing the rules |
+-----------------+-----------+----------+--------------------------------------------------------------+
- 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 <https://github.com/confluentinc/ksql/blob/0.1.x/docs/syntax-reference.md#syntax-reference>`_
- Responses
:If create table/stream succeed:
return True
:If failed:
raise a CreatError(respose_from_ksql_server)