ksqlient

ksQlient (formerly ksqlDB-JS)

A native Node.js client for ksqlDB

About the Project

ksQlient is a lightweight Node.js client for ksqlDB, a database for streaming applications leveraging Kafka infrastructure under the hood.

With our client, you can deploy stream-processing workloads on ksqlDB from within JS applications using simple, declarative SQL statements.

Sample use cases:

1. Build applications that respond immediately to events.
2. Craft materialized views over streams.
3. Receive real-time push updates, or pull current state on demand.

Table of Contents

• About the project
• Getting Started
  • Usage
• Features
• Developers
• Contributions
• License

Getting Started

The client is available on Node package manager (npm) (link)

npm install ksqlient

Usage

Create a client in the application file:

const ksqldb = require("ksqlient");
const client = new ksqldb({ ksqldbURL: "<url to ksqlDB server>" });

To run tests, initiate Docker containers included in yaml file:

docker-compose up
npm test

Features

Create a pull query

client.pull("SELECT * FROM myTable;");

Create a push query (persistent query that subscribes to a stream)

client.push("SELECT * FROM myTable EMIT CHANGES;", (data) => {
  console.log(data);
});

Terminate persistent query (e.g. push query)

client.terminate(queryId);

Insert rows of data into a stream

client.insertStream("myTable", [
  { name: "jack", email: "[e-mail]", age: 25 },
  { name: "john", email: "[e-mail]", age: 20 },
]);

List streams/queries

client.ksql("LIST STREAMS;");

Create table/streams

client.createStream(
  "testStream",
  (columnsType = ["name VARCHAR", "email varchar", "age INTEGER"]),
  (topic = "testTopic"),
  (value_format = "json"),
  (partitions = 1)
);

For custom SQL statements including complex joins use the .ksql method

client.ksql("DROP STREAM IF EXISTS testStream;");

SQL Query builder

Feel free to use the built-in query builder to parametrize any SQL query to avoid SQL injection.

const builder = new queryBuilder();
const query = "SELECT * FROM table WHERE id = ? AND size = ?";
const finishedQuery = builder.build(query, 123, "middle");

client.ksql(finishedQuery);

Create a table (materialized view) from a source stream

client.createTableAs(
  "testTable",
  "sourceStream",
  (selectArray = ["name", "LATEST_BY_OFFSET(age) AS recentAge"]),
  (propertiesObj = { topic: "newTestTopic" }),
  (conditionsObj = { WHERE: "age >= 21", GROUP_BY: "name" })
);

Create a stream based on an existing stream

client.createStreamAs(
  "testStream",
  (selectColumns = ["name", "age"]),
  "sourceStream",
  (propertiesObj = {
    kafka_topic: "testTopic",
    value_format: "json",
    partitions: 1,
  }),
  (conditions = "age > 50")
);

Pull stream data between two timestamps

client.pullFromTo(
  "TESTSTREAM",
  "America/Los_Angeles",
  (from = ["2022-01-01", "00", "00", "00"]),
  (to = ["2022-01-01", "00", "00", "00"])
);

Troubleshooting methods to inspect server metrics

• inspectServerStatus
• inspectQueryStatus
• inspectClusterStatus

Use Case

We have built a demo app to showcase how ksQlient can be used to create a streaming application .

Developers

• Javan Ang - GitHub | LinkedIn
• Michael Snyder - GitHub | LinkedIn
• Jonathan Luu - GitHub | LinkedIn
• Matthew Xing - GitHub | LinkedIn
• Gerry Bong - GitHub | LinkedIn

Contributions

Contributions to the code, examples, documentation, etc. are very much appreciated.

• Please report issues and bugs directly in this GitHub project.

License

This product is licensed under the MIT License - see the LICENSE.md file for details.

This is an open source product.

This product is accelerated by OS Labs.

ksqlDB is licensed under the Confluent Community License.

Apache, Apache Kafka, Kafka, and associated open source project names are trademarks of the Apache Software Foundation.