Using Pulsar SQL with Luna Streaming

Add the DataStax Helm chart repo to your Helm store.

helm repo add datastax-pulsar https://datastax.github.io/pulsar-helm-chart

Install the Helm chart using a minimalist values file.
This command creates a Helm release named "my-pulsar-cluster" using the DataStax Luna Helm chart, within the K8s namespace "datastax-pulsar". The minimal cluster creates only the essential components and has no ingress or load balanced services.

VALUES_URL="https://raw.githubusercontent.com/datastaxdevs/luna-streaming-examples/main/pulsar-sql/values.yaml"
helm install \
  --namespace datastax-pulsar \
  --create-namespace \
  --values $VALUES_URL \
  --version 3.0.4 \
  my-pulsar-cluster \
  datastax-pulsar/pulsar

Wait for the broker pod to be in a running state. You might see a few restarts as your components start up.

kubectl -n datastax-pulsar wait --for=condition=Ready pod/pulsar-broker-0 --timeout=120s

In a browser, navigate to http://localhost:8081.
You will be greeted by Trino’s login.

There is no authentication enabled, so input whatever user name you prefer and select “Log In”.
You will be greeted by a blank dashboard. This confirms Trino is up and running.

Download the minimalist Pulsar client. This "client.conf" assumes the port forwarding addresses we will perform in the next step.

wget https://raw.githubusercontent.com/datastaxdevs/luna-streaming-examples/main/client.conf

Set the client environment variable. Replace the value with the absolute path to the "client.conf" you just downloaded.

# the client conf path must be an absolute path to the downloaded conf file
export PULSAR_CLIENT_CONF=<REPLACE_WITH /path/to/client.conf>

Navigate to the Pulsar home folder and run the following command. The CLI will use the environment variable’s value as configuration for interacting with the Pulsar cluster.

./bin/pulsar-admin sources create --name generator --destination-topic-name public/default/mytopic --source-type data-generator

Confirm the topic was created.

Open a Trino shell session using the Trino CLI. The server’s port number should match the forwarded port number from the previous step. The user can match the name you used to login earlier in this guide, but doesn’t have to.

./trino --user test --server localhost:8081

Once inside the shell, let’s have a look around. List the catalogs Trino has loaded.

trino> show catalogs;

 Catalog
---------
 pulsar
 system
(2 rows)

Query 20230103_163242_00000_zvk84, FINISHED, 2 nodes

List the tables within the pulsar.”public/default” catalog/schema.

trino> show tables in pulsar."public/default";

  Table
---------
 mytopic
(1 row)

Query 20230103_163355_00001_zvk84, FINISHED, 2 nodes

Query the table to see the first 10 rows of data.

trino> select * from pulsar."public/default".mytopic limit 10;

Exit the trino shell.

trino> exit

Create a file named “PrestoExample.java” with the following contents. Remember to update the JDBC connection with the correct values for your environment.

import java.sql.*;

class PrestoExample {
  public static void main(String[] args) {
    try {
      Connection conn = DriverManager.getConnection("jdbc:presto://localhost:8090/pulsar?user=test");
      System.out.println("Connection established......");

      Statement stmt = conn.createStatement();

      try {
        ResultSet rs = stmt.executeQuery("select * from pulsar.\"public/default\".mytopic limit 10");
        while(rs.next()) {
            String str = rs.getString(1);
            System.out.println(String.format("%s", str));
        }
      }
      finally {
        stmt.close();
        conn.close();
      }
    }
    catch (Exception e) {
      e.printStackTrace();
    }
  }
}

Build the Java class.

javac PrestoExample.java

Download the version 443 Trino JDBC driver to the same folder as the Java class.

Run the client and observe the output. -cp tells Java to use the downloaded trino-jdbc jar in your classpath to connect.

java -cp .:trino-jdbc-443.jar PrestoExample