新功能详解+试用征集（Pulsar|新功能详解+试用征集：Pulsar Function Mesh）新功能详解+试用征集：Pulsar

导读：在 3 月 6 日 TGIP-CN 直播活动上，我们邀请到 StreamNative 高级工程师吕能，他为大家分享了 Pulsar Function Mesh 的功能与特性。下面是吕能分享视频的简洁文字整理版本，供大家参考。

很高兴今天能跟大家分享 StreamNative 基于 Pulsar Function 做的新工作：Function Mesh，它整个核心的想法是把一些复杂的、分离的、单独管理的 Function 进行统一化管理，基于原生地整合到 Kubernetes 中，并能充分利用其多方面的功能和调度算法。
【新功能详解+试用征集（Pulsar|新功能详解+试用征集：Pulsar Function Mesh）】

新功能详解+试用征集（Pulsar|新功能详解+试用征集：Pulsar Function Mesh）

文章图片

Pulsar 中的数据处理

文章图片

首先看一下 Pulsar 中所支持的各种数据处理的模块和方式，主要分为三个方面。第一，是基于 Presto 的交互式查询，Pulsar 中有自己的 Pulsar SQL，基于 Presto 对 Pulsar 整个集群查询；有和 Presto 相关的 Connector，可以直接通过 Presto 集群来查询 topic。
第二，作为消息队列、消息处理数据的核心，Pulsar 可以对接各种不同的流数据或者批数据处理的框架，比如 Flink、Spark、Hive。后续我们会发布 Pulsar 和 Flink SQL 整合的完整解决方案。
最后是 Pulsar 内建了 Pulsar Function，核心思想在于提供一个最简单的 API，让用户能够方便地处理在 Pulsar 中流动的数据。总结来说，Pulsar Function 是一个轻量级数据处理的进程，主要进行如下操作：

消费来自一个或多个 Pulsar topic 的消息；
将用户提供的处理逻辑应用于每个消息；
将结果发布到一个 Pulsar topic。

Pulsar Function 何为 Pulsar Function

文章图片

在上面提及的轻量级数据处理进程 Pulsar Function 如图所示。用户可以输入多个 topic，每输入一个 topic 都可以向用户自定义的 Pulsar Function 发送数据，Pulsar Function 的处理单元处理完成之后把结果发送到唯一一个 Output Pulsar topic，其中一些辅助性的 topic 可以进行日志或消息的收集。
Pulsar Function 并不是一个完全的流处理框架，不像 Flink 一样提供很多保证，也并不是一个计算的抽象层，其主要与 Pulsar 紧密结合在一起进行计算任务的处理。它的部署很简单，不需要再额外搭建管理任何集群，只需要在 Pulsar 配置文件中打开关于 Function 的支持，就可以向已有的 Pulsar 集群提交 Function。用户可以直接在集群中进行数据处理，而并不需要额外维护另一套集群去对接去处理。
Pulsar Function 常见的应用场景，比如集中在 ETL 数据清理的任务、实时数据的聚合......由于 Function 本身其实是一个非常通用的抽象，它只是一个应用函数，所以也可以应用于微服务场景。在 Function 应用的函数中可以调用任何的 API 进行操作，比如事件路由，用户可以使用 Pulsar Function 将数据分发到不同的集群。
如何实现 Pulsar Function

文章图片

上图为 Pulsar Function 的 API，Pulsar Function 支持三种语言来进行数据的处理：Java、Python、Golang。
Pulsar Function 支持的语义有三种：

最多一次（At-most once）：不关心消息是否发送成功、不需要消息发送的返回值；
至少一次（At-least once）：发送到的消息未接受到返回值会重新发送，保证消息不丢，可能会造成消息的重复，在消费的时候需要对消息进行幂等性操作；
精确一次（Exactly once）：保证消息不丢且不会重复。

Pulsar Function 自带简单的内件状态管理分为三种：

提供 Context 对象支持用户可访问状态；
将状态存储在 BookKeeper 中；
支持服务端操作（如计数器）。

之前介绍过的 API 除了 Input 之外，还会带有 Context 的参数，很多状态的管理是在 Context 中。

publicclassWordCountFunctionimplementsFunction { @Override publicVoid process(String input, Context context) throwsException{ Arrays.aslist(input.split(“\.”)).forEach(word -> context.incrCounter(word, 1)); } }

如何部署 Function Pulsar Function 的 CLI 可进行创建、删除、更新、获取、重启、关闭、开启等等一系列的操作。

$ ./pulsar—admin functions Usage: pulsar—admin functions [options] [command] [command options] Commands: localrunRun a PulsarFunction locally, rather than deploy to a Pulsar cluster) createCreate a PulsarFunctionin cluster mode (deploy it on a Pulsar cluster) deleteDelete a PulsarFunction that is running on a Pulsar cluster updateUpdate a PulsarFunction that has been deployedto a Pulsar cluster getFetch information about a PulsarFunction restartRestartfunction instance stopStopsfunction instance startStarts a stopped function instance statusCheck the current status of a PulsarFunction statsGet the current stats of a PulsarFunction listList all PulsarFunctions running under a specific tenant andnamespace querystateFetch the current state associated with a PulsarFunction putstatePut the state associated with a PulsarFunction triggerTrigger the associated specified Pulsarwith a supplied value

Pulsar Function 的特点 Pulsar Function 有下面的几个特点：

高效开发：简单的 API，不需要耗费大量精力学习，并且支持多种语言；
运维方便：与 Pulsar 完全集成，无需额外的系统/服务设置；
易于故障排除：本地运行时很方便，log topic 易于使用。

详解 Function Mesh 上面对 Pulsar Function 做了介绍，本部分内容将为大家深入展开介绍 Function Mesh。
何为 Function Mesh Function Mesh 是一组 function 的集合，它能让多个 function 在一起协调完成数据处理目标，并且每个 function 有各自明确任务和被定义好的 stage。特别强调的是，Function Mesh初衷不是代替 Flink 或者成为 Flink 的竞争对手，而是一个对现有流数据处理引擎的补充与支持。
下图是 Function Mesh 的经典视图：

文章图片

如上图所示，在 Function Mesh 之前，我们所使用的都是单个 Pulsar Function。引入 Function Mesh 之后，多个 function 就有了关联和数据联络，最后产生想要的结果，可以映射成微服务的不同场景。
Function Mesh 实现方案 Function Mesh 设计实现方案一：基于 Pulsar
目前 Pulsar 提供了命令行工具，可用来管理单个 function，如上图所示例，则需要在 Pulsar 命令行工具启动 function 1 到 function 6 才行，如此会带来管理的重复和复杂度；同时 Pulsar 会将上述多个 function 当作单个 function 来处理，使得很难追踪 function，无法将它们当作组合处理；也无法很好地了解各自 function 的上下游和处理顺序。
面对上述提到的几个问题，我们针对性提出解决方案，详情可参见 PIP-66[[1]](#)。PIP-66 主要的思路是在 Pulsar 中提供对 Function Mesh 的原生支持，即通过 Pulsar 命令行提交 Function Mesh，并在 Function Mesh YAML 配置文件中定义所包含的每个 function 参数及组织关系、输入和输出来源等。

bin/pulsar-admin function-mesh create -f mesh.yaml //创建 Function Mesh 示例命令

// YAML 配置演示 # Metadata name: PIP_Mesh namespace: PIP_Namespace tenant: PIP_Tenant # Function Mesh configs jarFile: /local/jar/files/example.jar # Functions functionInfos: - name: Func1 classname: org.apache.pulsar.functions.api.examples.ExclamationFunction replicas: 1 inputs: - pulsar_topic_sourcce output: - pulsar_topic_1 - name: Func2 classname: org.apache.pulsar.functions.api.examples.ExclamationFunction replicas: 1 inputs: - pulsar_topic_1 output: - pulsar_topic_result

下图是根据上述思路产生的一个 Function Mesh 调度方案，最大化利用已有的 Pulsar Function 调度机制实现 Function Mesh 的设计目标，当然也引入了 FunctionMeshManager 来管理 Function Mesh 的元数据。

文章图片

Function Mesh 设计实现方案二：基于 Kubernetes
随着整体项目和相关云端项目的推进，我们发现基于 Kubernetes 实现 Function Mesh 是非常有意义和价值的事情。用户要创建 Function Mesh，可以直接利用 Kubernetes 命令行工具创建（如下演示命令），而我们要做的是开发 CRD，类型就是 FunctionMesh，相关关系与方案一中的设定基本一致：

$ kubectl apply -f function-mesh.yaml …

apiVersion: cloud.streamnative.io/v1alpha1 kind: FunctionMesh metadata: name: functionmesh-sample spec: functions: - name: f1 … - name: f2 … - name: f3 … - name: f4 … - name: f5 … - name: f6 …

在该模式下，Function Mesh 不再是运行在 Pulsar 上，而是运行在 Kubernetes 云平台之上。在该模式下我们可以定义一系列资源，如单个 Function、Mesh（一整套 function 组合）、Source 和 Sink，其中 Source 和 Sink 更多是 Pulsar connector 中的概念，Source 是将第三方系统数据导入到 Pulsar topic 中，sink 则是相反的动作，这样也方便处理数据湖等场景。
基于 Kubernetes 的 Function Mesh 调度方案，如下图所示：

文章图片

方案对比：Pulsar vs Kubernetes
对比基于 Pulsar 和基于 Kubernetes 的 Function Mesh 实现方案，我们有几点思考：

如果能够利用 Kubernetes 的调度能力是十分有利的事情，对不同任务的调度是 Kubernetes 的专项能力，并且也能提供高可用、容错性保障。
在云环境中，Function 成为一等公民，地位与 Pulsar 提供的服务一样。
我们如果能将 Function 从 Pulsar 当中抽离出来，那它则具备与其他消息系统进行数据对接、处理的潜力，类似 AWS 提供的 Lamba Function，也方便我们进行事件驱动型的模式设计。

演示视频（点击链接查看原推文视频）
后续规划目前我们已经进行了大量前期 Function Mesh 的工作，对 Function Mesh 我们也有更多规划: