Query data with Pulsar SQL

    1. Install Pulsar.
    2. Install Pulsar .

    To query data in Pulsar with Pulsar SQL, complete the following steps.

    1. Start a Pulsar standalone cluster.
    1. Start a Pulsar SQL worker.
    2. ./bin/pulsar sql-worker run
    1. After initializing Pulsar standalone cluster and the SQL worker, run SQL CLI.
    1. Test with SQL commands.
    2. presto> show catalogs;
    3.  Catalog 
    4. ---------
    5.  pulsar  
    6.  system  
    7. (2 rows)
    9. Query 20180829_211752_00004_7qpwh, FINISHED, 1 node
    10. Splits: 19 total, 19 done (100.00%)
    11. 0:00 [0 rows, 0B] [0 rows/s, 0B/s]
    14. presto> show schemas in pulsar;
    15.         Schema         
    16. -----------------------
    17.  information_schema    
    18.  public/default        
    19.  public/functions      
    20.  sample/standalone/ns1 
    21. (4 rows)
    23. Query 20180829_211818_00005_7qpwh, FINISHED, 1 node
    24. Splits: 19 total, 19 done (100.00%)
    27. presto> show tables in pulsar."public/default";
    28.  Table 
    29. -------
    32. Query 20180829_211839_00006_7qpwh, FINISHED, 1 node
    33. Splits: 19 total, 19 done (100.00%)
    34. 0:00 [0 rows, 0B] [0 rows/s, 0B/s]
    1. Start the built-in connector DataGeneratorSource and ingest some mock data.

    And then you can query a topic in the namespace “public/default”.

    2. presto> show tables in pulsar."public/default";
    3.      Table      
    4. ----------------
    5.  generator_test 
    6. (1 row)
    8. Query 20180829_213202_00000_csyeu, FINISHED, 1 node
    9. Splits: 19 total, 19 done (100.00%)
    10. 0:02 [1 rows, 38B] [0 rows/s, 17B/s]

    You can query the mock data.

    2. public class TestProducer {
    4.     public static class Foo {
    5.         private int field1 = 1;
    6.         private String field2;
    7.         private long field3;
    9.         public Foo() {
    10.         }
    12.         public int getField1() {
    13.             return field1;
    14.         }
    15.         public void setField1(int field1) {
    16.             this.field1 = field1;
    19.         public String getField2() {
    20.             return field2;
    21.         }
    23.         public void setField2(String field2) {
    24.             this.field2 = field2;
    25.         }
    27.         public long getField3() {
    28.             return field3;
    29.         }
    31.         public void setField3(long field3) {
    32.             this.field3 = field3;
    33.         }
    34.     }
    36.     public static void main(String[] args) throws Exception {
    37.         PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
    38.         Producer<Foo> producer = pulsarClient.newProducer(AvroSchema.of(Foo.class)).topic("test_topic").create();
    40.         for (int i = 0; i < 1000; i++) {
    41.             Foo foo = new Foo();
    42.             foo.setField1(i);
    43.             foo.setField2("foo" + i);
    44.             foo.setField3(System.currentTimeMillis());
    45.             producer.newMessage().value(foo).send();
    46.         }
    47.         producer.close();
    48.         pulsarClient.close();