关于 Apache Pulsar

Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性。<br>

GitHub 地址：http://github.com/apache/pulsar/

Pulsar Functions 是开源数据技术框架 Apache Pulsar 为轻量级计算提供的内置流处理器。在 [2020 年 Pulsar Summit 会议](https://pulsar-summit.org/en/event/virtual-conference-2020)上，我发表了一次关于 Pulsar Functions 的演讲。本文将深入讨论 Functions 的架构和实现细节。

Pulsar Functions 简介

Pulsar Functions 是 Pulsar 消息系统的核心计算基础结构。使用 Pulsar Functions，无需部署单独的系统（如 Apache Storm、Apache Heron），即可基于单条消息创建复杂的处理逻辑，简化事件流并引入无服务架构。

轻量级计算 function 从一个或多个 Pulsar topic 消费消息，将用户提供的处理逻辑应用于每条消息，并发布计算结果到其他 topic。由于不需要外部处理系统，Pulsar Functions 不仅使应用程序开发更便捷，还简化了故障排除操作，减轻了运维负担。

另外，开发人员可以直接使用 Pulsar Functions 的 API。了解 Java 语言的程序员可以直接使用 Java SDK 编写 function。示例如下：

import java.util.function.Function; public class ExclamationFunction implements Function<String, String>  @Override   public String apply(String input) {       return input + "!";   }}

Pulsar Functions 旨在借助简单的 API 和执行框架处理常见的流使用场景（如过滤、路由、扩充），而不是替换重量级流处理引擎（如 Spark、Flink）。

用户可以自行编写 Pulsar function，提交到 Pulsar 集群后，即可与 Pulsar Functions 的内置管理部件一起使用。使用基于 CRUD 的 REST API，用户可以从任一工作流程提交 function，提交后即可运行。

提交流程

提交 function 的流程称为 Function Representation。提交流程的结构称为 FunctionConfig，包括租户、命名空间和名称。Function 通过提交 JAR 或 Python 文件消费输入和输出数据、用户配置、私密管理支持等。另外，用户可以同时运行多个 function。

public class FunctionConfig {        private String tenant;        private String namespace;        private String name;        private String className;        private Collection<String> inputs;        private String output;        private ProcessingGuarantees processingGuarantees;        private Map<String, Object> userConfig;        private Map<String, Object> secrets;        private Integer parallelism;        private Resources resources;        ... }

用户提交 function 后，系统将会对 function 进行“提交检查”或“验证”，确保用户有权限提交此 function 到特定的命名空间和租户。如果使用 Java 语言，提交时就会加载这些类，确保指定的类在 JAR 文件中。所有的检查都会在提交时立刻进行，因此出现错误时，用户可以很快收到提示消息，而不用自行查看错误日志。

下一步是复制代码到 BookKeeper。将提交代码中的所有参数以协议缓冲结构表示为 FunctionMetaData，示例如下：

message FunctionMetaData {    FunctionDetails functionDetails ;    PackageLocationMetaData packageLocation;    uint64 version ;    uint64 createTime;    map<int32 , FunctionState> instanceStates ;    FunctionAuthenticationSpec functionAuthSpec ;}

Function MetaData Manager 负责管理 FunctionMetaData 结构。从 worker 的角度来看，Function MetaData Manager 负责维护系统记录。Function MetaData Manager 将完全限定 Function 名称（Fully Qualified Function Name，FQFN）映射到由 Pulsar topic 命名空间和 function 信息构成的 Function MetaData；基于提交内容更新、管理计算机状态，当提交多个 worker 时，检查冲突并将元数据写入 topic。

调度流程

系统接收 function 提交后，使用可插拔调度程序进行调度。如果 function 在提交后仅由 leader 执行，就会激活调度程序。通过灾备方式订阅协调 topic 即可确定 leader，即 topic 中的 consumer。

Leader 将任务写入到任务 Topic（Assignment Topic）。任务 Topic 储存在 Pulsar 特定的命名空间中，被分配给各个 worker。Worker 有权限查看所有已压缩并包含所有系统逻辑的任务，如任务 Table（Assignment Table）中的 FQFN、实例 ID。

执行流程

更改任务 Table 即可触发执行流程。Worker 的部件（Function RunTime Manager）负责管理 function 周期任务，如使用 Spawner 启动或停止消息。

Java 实例和 Pulsar IO

Pulsar Java 实例被封装为 Source、function（实际逻辑），也可以说是 Sink ensemble。Source 从 input topic 读数据，而 Sink 从 topic 读数据。

在使用内置 function 时，“Source”即为 Pulsar Source，从 Pulsar 读数据；“Sink”即为 Pulsar Sink，向 Pulsar topic 写数据。

但如果提交类似于 Google 发布/订阅的 Source（非 Pulsar Source），借助 Pulsar IO 成为 connector，且实际功能类似于 Pulsar Function，则此 function 为特性 Function（Identity Function），可用于处理数据。提交后，Pulsar Sink 向 topic 写入此 function；非 Pulsar Sink 则向外部系统写入。正因为 Pulsar IO 基于 Pulsar Functions 编写，Pulsar 才得以从外部系统读数据。

开始使用 Pulsar Functions

由于不需要外部处理系统，Pulsar Functions 不仅使应用程序开发更便捷，还简化了故障排除操作，减轻了运维负担。通过简单、轻量级无 SDK 的 API 和执行框架，Pulsar Functions 可以过滤、路由、扩充大多数流使用场景。开发人员可以自行编写 Pulsar function，提交至 Pulsar 集群后即可与 Pulsar Functions 的其他内置功能一起使用。此外，使用 Pulsar IO 可以处理非 Pulsar Source 并将其写入外部系统。

想要了解更多信息，点击这里查看演讲 PPT。

原文作者 Sanjeev Kulkarni，翻译 Sijia@StreamNative，如需转载，请后台留言。

英文链接：https://streamnative.io/blog/tech/2020-10-06-pulsar-functions-deep-dive

欢迎关注 Apache Pulsar 中文社区公众号 ApachePulsar，获取最新 Apache Pulsar 动态与技术干货。