Create topic sensor-data in the public/default namespace:

pulsar-admin topics create persistent://public/default/sensor-data

Display the topics in the public/default namespace:

pulsar-admin topics list public/default

Configure retention to be unlimited both in size and time for the public/default namespace:

pulsar-admin namespaces set-retention public/default \
 --size -1 \
 --time -1

Start a Pulsar SQL coordinator/worker:

pulsar sql-worker run

If the coordinator/worker starts successfully, there will be a SERVER STARTED message displayed.

Do not worry if you see, at the beginning of the output, ERROR: ld.so: object [...] from LD_PRELOAD cannot be preloaded. You can safely ignore it.

Start a Pulsar SQL client:

pulsar sql

List the available catalogs:

SHOW CATALOGS;

List the schemas in pulsar:

SHOW SCHEMAS FROM pulsar;

List the tables in schema public/default:

SHOW TABLES FROM pulsar."public/default";

Count the number of rows in table sensor-data:

SELECT COUNT(*) AS number_of_messages 
FROM pulsar."public/default"."sensor-data";

1. Check out the code in file SensorDataProducer.java:

sensor-data-producer/src/main/java/com/example/pulsar/SensorDataProducer.java

2. Build the producer:

mvn -f sensor-data-producer/pom.xml clean compile assembly:single

3. Run the producer:

java -cp sensor-data-producer/target/sensor-data-producer-1.0-SNAPSHOT-jar-with-dependencies.jar com.example.pulsar.SensorDataProducer

Go back to the SQL Client Terminal for the following commands.

Note that, after executing a query that returns a larger result set, arrow keys can be used to browse the result and :q can be used to quit the result browsing mode.

1. Set the current working schema:

USE pulsar."public/default";

2. Find the number of measurements:

SELECT COUNT(*) AS number_of_messages 
FROM "sensor-data";

You may find that the last measurement produced by the producer is missing. This is expected behaviour: related to how Pulsar SQL assumes it is working in a continuous-streaming regime.

3. Find distinct sensors ids (note the case and compare to the Java class definition):

SELECT DISTINCT sensorid  
FROM "sensor-data"
ORDER BY sensorid ASC;

4. Find all measurements whose values are greater or equal to 100:

SELECT sensorid, timestamp, value  
FROM "sensor-data"
WHERE value >= 100;

Press q to end the query.

5. Find the number of measurements and average measurement value for each sensor:

SELECT sensorid, 
       COUNT(*) AS total,
       AVG(value) AS avg  
FROM "sensor-data"
GROUP BY sensorid
ORDER BY sensorid ASC;

6. Compare measurements with the same timestamps reported by two different sensors:

SELECT timestamp, 
       S1.value AS "sensor-1 value", 
       S2.value AS "sensor-2 value"
FROM "sensor-data" AS S1 
     INNER JOIN "sensor-data" AS S2
     USING (timestamp)
WHERE S1.sensorid = 'sensor-1'
  AND S2.sensorid = 'sensor-2'
ORDER BY timestamp DESC;

Press q to end the query.