如何让Apache Pulsar/Pulsar SQL实现Push-Queries推送查询？

一、利用Pulsar原生特性实现数据推送 #

Pulsar本身是流式消息系统，天然支持实时数据推送，完全可以替代轮询模式：

• 客户端直接订阅主题：如果你的查询结果是持续生成的，可将计算后的数据写入Pulsar主题，客户端通过Pulsar官方SDK（Java/Go/Python等）订阅该主题，实时接收新数据。客户端还支持通过WebSocket协议连接Pulsar，直接把数据推送到前端，无需额外服务中转。
• Pulsar Functions做实时计算+推送：如果需要先对原始数据做类SQL的查询处理，用Pulsar Functions编写轻量级逻辑，将计算结果输出到目标主题再由客户端订阅。示例代码：

  @FunctionAnnotation(outputTopic = "persistent://public/default/filtered-results")
  public String process(String input) {
      // 模拟SQL过滤：仅保留包含"valid"的记录
      if (input.contains("valid")) {
          return input;
      }
      return null;
  }
  

二、Pulsar SQL的实时查询推送方案 #

Pulsar SQL基于Presto构建，默认是交互式查询，但可以通过以下方式实现类Continuous Query的推送效果：

• 将查询结果写入Pulsar主题：用Pulsar SQL的INSERT INTO语句，把持续生成的查询结果写入指定主题，客户端订阅该主题即可实时获取数据：

  INSERT INTO pulsar.public.default.query-results
  SELECT column1, column2 FROM pulsar.public.default.raw-data
  WHERE condition = true;
  

  这条SQL会持续把符合条件的新数据写入目标主题，只有当原始数据变化时才会产生新结果，完全避免无效查询。
• 借助Presto持续查询扩展：部分版本的Presto支持CONTINUOUS QUERY语法，可以将查询结果持续输出到外部系统，再通过HTTP2/WebSocket中转到客户端。

三、第三方集成实现服务端点推送 #

如果需要直接将结果推送到HTTP2/WebSocket服务端点，可通过以下方式：

• Pulsar HTTP Sink Connector：配置HTTP Sink Connector，将主题中的查询结果推送到指定的HTTP服务端点，服务端点再通过WebSocket转发给客户端。配置示例（config.yaml）：

  configs:
    url: "http://your-service-endpoint/push"
    method: "POST"
    batchSize: 1
    batchTimeMs: 100
  

  这样每条新的查询结果都会立即推送到你的服务端点。
• 自定义推送服务：编写轻量级服务，订阅Pulsar主题获取实时结果，再通过HTTP2或WebSocket主动推送给客户端。比如用Netty实现WebSocket服务，订阅Pulsar主题后，有新数据就推送给在线客户端。

四、关键注意事项 #

• 所有方案均为数据驱动，只有当原始数据发生变化时才会触发计算和推送，彻底避免轮询的无效查询。
• 若数据量较大，建议在Pulsar Functions或Pulsar SQL中做前置过滤，减少推送的数据量；同时合理配置客户端的批量接收参数，平衡实时性与性能。

内容的提问来源于stack exchange，提问作者feder