Pulsar 如何应用到数据查询中

首先我们来简单回顾一下 Pulsar上周的新增内容：

• PR 7646 - Support Set Backlog Quota On Topic Level
  https://github.com/apache/pulsar/pull/7646
• PR 4955 - Introduce system topic and topic policies service
  https://github.com/apache/pulsar/pull/4955
• PIP 39 - Namespace Change Events
  https://github.com/apache/pulsar/wiki/PIP-39%3A-Namespace-Change-Events
• PR 7647 - Allow Ability To Specify Retain Key Ordering In Functions
  https://github.com/apache/pulsar/pull/7647
• PR 7655 - Fix Backward Compatibility Issues With Batch Index Acknowledgment
  https://github.com/apache/pulsar/pull/7655

---

Presto

在正式进入「基于 Apache Pulsar 进行的流数据查询」话题分享前，我们先来说一下关于 Presto 的相关介绍。

Presto 是一个开源分布式 SQL 查询引擎，主要用于交互式分析查询。Presto 采用「主从」的分析模式，主节点可以接收到用户的查询语法，生成子查询任务进行后续动态操作。计算结果采取分批模式返还给用户。

虽然 Presto 可以支持标准的 SQL 查询语法，也支持多种数据源，但它并不是传统意义上的数据库。它更多的实现意义在于数据仓库和数据分析。

Pulsar SQL

Pulsar SQL 是基于 Presto 产品的基础上进行开发的。右侧为 Presto 的一些接口，Pulsar 通过实现这些接口来打通与 Presto 的交互。

接下来我们就开始进行设置。

Namespace Retention

bin/pulsar-admin namespaces set-retention --size 100M --time 7d public/default
bin/pulsar-admin namespaces get-retention public/default

以上命令行，使用 Pulsar 的小伙伴可能都很熟悉，主要是设置 namespace 的保留策略（100m 或 保留 7 天），做这个设置的原因是为了防止自动生成的数据会被 Pulsar 删除。

Produce Messages To Pulsar

目前 Pulsar SQL 支持的数据类型很多，基本 Pulsar 里的数据都支持。

比如原始数据类型：BOOLEAN, INT16, DOUBLE, STRING；复杂的数据类型：KeyValue 和 Struct 等。

首先定义一个 JavaBean，包含名称、价格、更新时间等字段。

之后就可创建 producer 进行消息生产。使用 JsonSchema、设置批量最大消息、设置批量最大发布延迟、设置 compressionType、以异步方式发送等操作进行后续步骤（如下图）。

Query

消息发送之后，就可以使用 Pulsar SQL 进行消息查询。目前使用 Pulsar SQL 进行查询的方式有三种。

第一种方式是采用命令行方式进行，在 Pulsar 根目录下执行 `bin/pulsar sql`就可以进入交互式界面。

第二种方式是采用 Presto JDBC 的方式进行。首先需要引入 Presto JDBC 的依赖，然后再用 Java SQL 的语言进行查询。

第三种方式是采用 Presto HTTP API 模式进行。利用 `http://ip:port/v1/statement` 和 `http header: X-Presto-User` 命令进行。

第四种也可以利用 Metabase 这个工具。Metabase 是一个开源模式的数据探索工具，支持多种数据源，生成多种形式图表。主要操作步骤是在“设置”里进行相关配置的设定。

如上图示例，选择数据类型为 Presto，主机为 Pulsar SQL 的主机地址，接下来就可以通过 Metabase 输入语法进行数据查询提问。可以选择线性或者多种图表的呈现形式进行展示。

查询性能优化建议

在针对数据查询方面，可以通过以下几点进行效率的提高。

1. 增加 Topic Partition 数量

当我们需要查询的数据数量非常大的时候，就会很耗时。这里就可以借鉴传统数据库的分表思路，来增加 Topic Partition 数量，扩大「分区」数量。这样就可以专注于某个分区的数据，提高查询效率。

2. 在查询条件中加入 ` __publish_time__`

Pulsar SQL 在监测查询语法时，如果添加了这个条件，会在拆分任务前计算出需要扫描范围，这样也可以提高查询效率（如上图所示）。

未来 Pulsar SQL 也会添加对 Compaction data（压缩数据） 的读取，Compaction 数据主要是对相同 key 数据的压缩，只保留最新的数据。

压缩后可以较少数据量，也可以提高查询效率。目前 Pulsar 社区的小伙伴也在努力开发中，期待大佬们的成果！

总结

本周分享主要针对 Pulsar SQL 在数据查询方面的一些细节介绍，希望大家可以通过本期的分享，更好地使用 Pulsar SQL 进行数据分析等。

更多细节可以查看下方回放视频。

同时本周末将于张勇同学为大家带来关于 namespace 策略的相关讲解。大家可以扫描下方二维码或者点击「阅读原文」进行报名。