作者 | 冉小龙

编辑 | Irene

导读：本文是 Pulsar Functions 的第一篇，主要向大家普及 Pulsar Functions 相关的概念，架构设计，以及运行的逻辑。

更多干货请关注微信公众号“StreamNative” 。

当我们进行流式处理的时候，很多情况下，我们的需求可能只是下面这些简单的操作：简单的 ETL 操作\Real-time 的聚合操作\event 路由\基于 event 的服务。

但为了实现这些功能，我们不得不去部署一整套 SPE 服务。部署成功后才发现需要的仅是 SPE 服务中的一小部分功能，部署 SPE 的成本可能比用户开发这个功能本身更困难。由于SPE 本身 API 的复杂性，我们需要了解这些算子的使用场景，明白不同算子之间有哪些区别，什么情况下，应该使用什么算子来处理相应的逻辑。

基于以上原因，我们设计并实现了 Pulsar Functions，在 Pulsar Functions 中，用户只需关心计算逻辑本身，而不需要去了解或者部署 SPE 的相关服务，当然你也可以将 pulsar-function 与现有的 SPE 服务一起使用。也就是说，在 Pulsar Functions 中，无需部署 SPE 的整套服务，就可以达到与 SPE 服务同样的优势。

什么是 Pulsar Functions?

Pulsar Functions 是一个轻量级的计算框架，像 AWS 的 lambda、Google Cloud 的 Functions 一样，Pulsar Functions 可以给用户提供一个部署简单、运维简单、API 简单的 FASS（Function as a service）平台。

设计架构-Instance

image

在 Pulsar 中，我们把需要处理的操作单元抽象为主题。如上图所示，整个 instance 的设计中，我们可以将其拆分为三类主题：

● input topic ● output topic ● log topic

input topic 是数据的来源，在 Pulsar Functions 中，所有的数据均来自 input topic。当数据进入input topic 中，Pulsar Functions 充当消费者的角色，去 input topic 中消费消息；当从 input topic 中拿到需要处理的消息时，Pulsar Functions 充当生产者的角色往 output topic 或者 log topic 中生产消息。

output topic 和 log topic 都可以看作是 Pulsar Functions 的输出。从是否会有 output 这个点来看，我们可以将 Pulsar Functions 分为两类，当有输出的时候 Pulsar Functions 会将相应的 output 输出到 output topic 中。log topic 主要存储用户的日志信息，当 Pulsar Functions 出现问题时，方便用户定位错误并调试。

综上所述：我们不难看出 Pulsar Functions 充当了一个消息处理和转运的角色。

image

设计架构-Pulsar Functions Worker

image

上述处理过程总结如下：

1. 用户给 REST server 发送一个请求

2. REST server 响应用户的请求

3. Function Metadata Manage 将更新写到 FMT

4. Function Metadata Manager 从 FMT 中读取更新

5. Scheduler Manager 将更新写到 assignment 主题

6. Function Runtime Manager 从 assignment topic 中读取更新

7. Membership Manager 配合 Coordination Topic 来做 leader 的选举

8. Membership Manager 配合 Coordination Topic 保证 active membership 的资格

下面是一个典型的 Pulsar Functions worker 运行的例子:

1. 用户提交一个请求到 REST sever 来执行一个 function 的实例。因为 function 的分配不依赖于任何 worker，所以这个请求可能被提交到任意的一个 worker 来处理。

2. REST server 将请求传递给 Function Metadata Manager，Function Metadata Manager 将该请求写入 Function Metadata Topic（FMT）。

3. Function Metadata Manager 会监听所有新进入 FMT 的 message，当 Function Metadata Manager 从 FMT 收到一个新的 messgae 时，首先会去检查消息是否过期。如果过期了，就直接丢弃；没过期的话，Function Metadata Manager 使用该消息更新其内部状态，其中包含正在运行的所有 function 的全局视图。由于每个工作程序都运行一个 Function Metadata Manager，因此每个 worker 都有一个最终一致的全局视图，其中包含所有正在运行的函数的状态。因此，可以将读取请求（例如获取正在运行中函数的状态请求）提交给任何工作者。

4. 当 Function Metadata Manager 更新其内部状态时，会去触发 Scheduler Manager。因为这个时候系统有新的更新进来，肯定需要去对这个新的更新进行计算。这时候，处于leader 状态的 worker 执行调度策略，看这一次的计算分配给谁执行比较合适，然后将新的分配写入到 Assignment Topic。 Membership Manager 用于维护集群中的 leader 以及所有处于 active 状态成员的列表

5. Function Runtime Manager 会监听 Assignment Topic，看是否有新的更新。当有更新进来时，Function Runtime Manager 将更新其内部状态，其中包含所有 worker 的全局视图。如果有更新，Function Runtime Worker 会根据这个更新，判断是否需要 start 或者stop function 的实例。

Processing guarantees

Pulsar Functions 提供了以下三种处理语义，用户可根据具体的场景作出选择：

● At most once (默认) ● At least once ● Effectively once

1. At most once 是指消息最多会被处理一次。从 input topics 中接收到之后，在真正处理消息之前去执行, at-most-once 模式下，不管 function 是否执行成功，这个 message 都会被确认（ack），而且只发送一次，无论是否发送成功，都不会重发。

2. At least once 是指消息至少发送一次。如果消息未能接受成功，可能会重发，直到接收成功。在整个 Pulsar Functions 处理消息的过程中，如果失败，都需要对该 message 执行 nack（重发） 操作，来保证 At least once 语义的正确性。

3. Effectively once 是指消息会被有效执行一次。上述两种语意都没办法保证系统 crash 之后数据的一致性问题，Effectively once 可以保证只会对结果产生一次影响。Effectively once 本身更像是一个事务的处理过程，首先我们在 setup 生产者的时候需要保证生产者的幂等性；其次在处理消息的过程中，如果出现错误，我们需要让整个 function 停止操作，这点不同于 At least once。

Pulsar Functions subscribeType

为了同时兼容 queue 和 stream 的消费方式，Pulsar 在消费者之间抽象了一层订阅层，在 Pulasr 中，订阅的方式主要分为如下三种：

● exclusive ● failover ● share

但是 Pulsar Functions 中并没有支持 exclusive 的订阅方式。这是为什么呢？

在大部分 functions 的特定场景下，exclusive 的订阅类型没多大用，我们分为两种情况来讨论：

1.如果只有一个 instance，那么 failover 就相当于独占的类型。

2.如果有多个 instance，exclusive 类型的订阅会不断的 crash、 restart，而 failover 的订阅是通过 failover 的方式来进行切换，保证有一个 active 的 worker。（这个是本质原因）

Pulsar Functions runtime

在 Pulsar Functions 中，我们把每个运行的实例称作 instance，一个 instance 执行的是一个 function 的副本。pulsar-function 支持同时并行执行多个 instance，具体 instance 执行的数量可以通过配置文件来指定。为了最大程度提高部署的灵活性，我们支持以下三种 runtime 的形式，用户可以根据需求选择。

● thread runtime ● process runtime ● kubernetes runtime

不同的运行时提供的是不同的隔离级别和成本。这个是一个相对的关系，成本越低，隔离效果越低；反之亦然。thread runtime 是 java 框架开发的，所以目前只有 Java 的 instance 支持 thread runtime。

如何部署 Pulsar Functions

Pulsar Functions 目前支持 java、 python 和 go 三种语言（稍后将支持更多的语言），大家可以选择自己熟悉的语言编写相应的处理函数。目前，Pulsar Functions 支持以下两种部署方式：

local run

在本地运行或者集群外运行一个 Pulsar Functions，适用于开发者。

cluster

在集群内运行 Pulsar Functions。在该模式下部署 function 时，Apache BookKeeper 将自动处理状态存储，目前 go 版本的 function 暂时不支持状态存储。

本文转载自 StreamNative 微信公众号。获取更多干货，请关注 StreamNative 微信公众号。