Repository files navigation

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+

pip install ksql

Or

git clone https://github.com/bryanyang0528/ksql-pythoncd ksql-pythonpython setup.py install

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:

confluent start ksql-server

• Setup for the KSQL API:

fromksqlimportKSQLAPIclient=KSQLAPI('http://ksql-server:8088')

• Setup for KSQl API with logging enabled:

importloggingfromksqlimportKSQLAPIlogging.basicConfig(level=logging.DEBUG)client=KSQLAPI('http://ksql-server:8088')

• Setup for KSQL API with Basic Authentication

fromksqlimportKSQLAPIclient=KSQLAPI('http://ksql-server:8088',api_key="your_key",secret="your_secret")

This method can be used for some KSQL features which are not supported via other specific methods likequery,create_stream orcreate_stream_as.The following example shows how to execute theshow tables statement:

client.ksql('show tables')

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

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:

fromksqlimportKSQLAPIclient=KSQLAPI('http://localhost:8088')query=client.query('select * from table1')foriteminquery: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}

Execute queries with the new/query-stream endpoint. Documentedhere

To execute a sql query use the same syntax as the regular query, with the additionaluse_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 theclose_query call.Provide thequeryId returned from thequery call.

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

Uses the new/inserts-stream endpoint. Seedocumentation

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.

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

• Responses

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 ... )]ASSELECT<select_columns>FROM<src_table>[WHERE<conditions>]PARTITION BY<partition_by>];

KSQL JOINs between Streams and Tables are not supported yet via explicit methods, but you can use theksql method for this like the following:

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")

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

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

• Responses

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

KSQL Syntax Reference

• Responses