1. 安装 。
  2. 安装 Pulsar 内置连接器。

要使用 Pulsar SQL 查询数据，请完成以下步骤。

  1. 启动 Pulsar 独立集群。
  1. 启动 Pulsar SQL 工作器。
  1. 初始化 Pulsar 独立集群和 SQL 工人后，运行 SQL CLI 。
  1. 使用 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.  public/default        
  17.  public/functions      
  18.  sample/standalone/ns1 
  19. (4 rows)
  22. Splits: 19 total, 19 done (100.00%)
  23. 0:00 [4 rows, 89B] [21 rows/s, 471B/s]
  26. presto> show tables in pulsar."public/default";
  27.  Table 
  28. -------
  29. (0 rows)
  31. Query 20180829_211839_00006_7qpwh, FINISHED, 1 node
  32. Splits: 19 total, 19 done (100.00%)
  33. 0:00 [0 rows, 0B] [0 rows/s, 0B/s]
  1. Start the built-in connector DataGeneratorSource and ingest some mock data.

然后可以在命名空间“公共/默认”中查询一个主题。

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

可以查询模拟数据。

  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();