It is super easy to get started on querying data in Pulsar.

要求

  1. Pulsar distribution
  2. Pulsar 内置连接器

首先，启动Pulsar独立集群：

  1. ./bin/pulsar standalone

下一步，启动一个Pulsar SQL worker：

  1. ./bin/pulsar sql-worker run

在Pulsar独立集群和SQL worker完成初始化之后，运行SQL CLI：

  1. ./bin/pulsar sql

您现在可以开始输入一些SQL命令：

  1. presto> show catalogs;
  2.  Catalog 
  3. ---------
  4.  pulsar  
  5.  system  
  6. (2 rows)
  8. Query 20180829_211752_00004_7qpwh, FINISHED, 1 node
  9. Splits: 19 total, 19 done (100.00%)
  10. 0:00 [0 rows, 0B] [0 rows/s, 0B/s]
  13. presto> show schemas in pulsar;
  14.         Schema         
  15. -----------------------
  16.  information_schema    
  17.  public/default        
  18.  public/functions      
  19.  sample/standalone/ns1 
  20. (4 rows)
  22. Query 20180829_211818_00005_7qpwh, FINISHED, 1 node
  23. Splits: 19 total, 19 done (100.00%)
  24. 0:00 [4 rows, 89B] [21 rows/s, 471B/s]
  27. presto> show tables in pulsar."public/default";
  28.  Table 
  29. -------
  30. (0 rows)
  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]

Currently, there is no data in Pulsar that we can query. Lets start the built-in connector DataGeneratorSource to ingest some mock data for us to query:

  1. ./bin/pulsar-admin source create --name generator --destinationTopicName generator_test --source-type data-generator

然后，将有一个可以在”public/default”命名空间查询的主题：

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

我们现在可以查询主题“generator_test”中的数据：

  1. presto> select * from pulsar."public/default".generator_test;
  3.   firstname  | middlename  |  lastname   |              email               |   username   | password | telephonenumber | age |                 companyemail                  | nationalidentitycardnumber | 
  4. -------------+-------------+-------------+----------------------------------+--------------+----------+-----------------+-----+-----------------------------------------------+----------------------------+
  5.  Genesis     | Katherine   | Wiley       | genesis.wiley@gmail.com          | genesisw     | y9D2dtU3 | 959-197-1860    |  71 | genesis.wiley@interdemconsulting.eu           | 880-58-9247                |   
  6.  Brayden     |             | Stanton     | brayden.stanton@yahoo.com        | braydens     | ZnjmhXik | 220-027-867     |  81 | brayden.stanton@supermemo.eu                  | 604-60-7069                |   
  7.  Benjamin    | Julian      | Velasquez   | benjamin.velasquez@yahoo.com     | benjaminv    | 8Bc7m3eb | 298-377-0062    |  21 | benjamin.velasquez@hostesltd.biz              | 213-32-5882                |   
  8.  Michael     | Thomas      | Donovan     | donovan@mail.com                 | michaeld     | OqBm9MLs | 078-134-4685    |  55 | michael.donovan@memortech.eu                  | 443-30-3442                |   
  9.  Brooklyn    | Avery       | Roach       | brooklynroach@yahoo.com          | broach       | IxtBLafO | 387-786-2998    |  68 | brooklyn.roach@warst.biz                      | 085-88-3973                |   
  10.  Skylar      |             | Bradshaw    | skylarbradshaw@yahoo.com         | skylarb      | p6eC6cKy | 210-872-608     |  96 | skylar.bradshaw@flyhigh.eu                    | 453-46-0334                |    
  11. .
  12. .
  13. .

现在，您有一些模拟数据可供查询和使用！

如果您想尝试获取一些自己的数据，可以编写一个简单的生产者将自定义数据写入Pulsar。

例如:

  1. public class Test {
  3.      public static class Foo {
  4.         private int field1 = 1;
  5.         private String field2;
  6.         private long field3;
  7.      }
  9.      public static void main(String[] args) throws Exception {
  10.         PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
  11.         Producer<Foo> producer = pulsarClient.newProducer(AvroSchema.of(Foo.class)).topic("test_topic").create();
  13.         for (int i = 0; i < 1000; i++) {
  14.             Foo foo = new Foo();
  15.             foo.setField1(i);
  16.             foo.setField2("foo" + i);
  17.             foo.setField3(System.currentTimeMillis());
  18.             producer.newMessage().value(foo).send();
  19.         }
  20.         producer.close();
  21.         pulsarClient.close();
  22.      }
  23. }

接下来，你应该能够查询到刚写入的数据。