一、Function背景介绍
当我们进行流式处理的时候，很多情况下，我们的需求可能只是下面这些简单的操作：简单的ETL 操作\聚合计算操作等相关服务。
但为了实现这些功能，我们不得不去部署一整套 SPE 服务。部署成功后才发现需要的仅是SPE(流处理引擎)服务中的一小部分功能，部署 SPE 的成本可能比用户开发这个功能本身更困难。由于SPE 本身API 的复杂性，我们需要了解这些算子的使用场景，明白不同算子之间有哪些区别，什么情况下，应该使用什么算子来处理相应的逻辑。
基于以上原因，我们设计并实现了 Pulsar Functions，在 Pulsar Functions 中，用户只需关心计算逻辑本身，而不需要去了解或者部署 SPE 的相关服务，当然你也可以将pulsar-function 与现有的SPE 服务一起使用。也就是说，在 Pulsar Functions 中，无需部署SPE 的整套服务，就可以达到与 SPE 服务同样的优势。
二、什么是Functions
Pulsar Functions 是一个轻量级的计算框架，像 AWS 的 lambda、Google Cloud 的Functions 一样，Pulsar Functions 可以给用户提供一个部署简单、运维简单、API 简单的 FASS（Function as a service）平台。Pulsar Functions 的设计灵感来自于 Heron 这样的流处理引擎，Pulsar Functions 将会拓展Pulsar和整个消息领域的未来。使用 Pulsar Functions，用户可以轻松地部署和管理 function，通过function 从Pulsartopic 读取数据或者生产新数据到 Pulsar topic。
引入 Pulsar Functions 后，Pulsar 成为统一的消息投递/计算/存储平台。只需部署一套Pulsar 集群，便可以实现一个计算引擎，页面简单，操作便捷。
Input topic 是数据的来源，在 Pulsar Functions 中，所有的数据均来自 input topic。当数据进入inputtopic 中，Pulsar Functions 充当消费者的角色，去 input topic 中消费消息；当从input topic 中拿到需要处理的消息时，Pulsar Functions 充当生产者的角色往 output topic 或者 log topic 中生产消息。
Output topic 和 log topic 都可以看作是 Pulsar Functions 的输出。从是否会有output 这个点来看，我们可以将 Pulsar Functions 分为两类，当有输出的时候 Pulsar Functions 会将相应的output 输出到outputtopic中。log topic 主要存储用户的日志信息，当 Pulsar Functions 出现问题时，方便用户定位错误并调试。
综上所述：我们不难看出 Pulsar Functions 充当了一个消息处理和转运的角色。
在使用Pulsar Functions，可以使用不同的语言来编写，比如Python、Java、Go等。编写方式主要两种：
- 本地模式：集群外部，进行本地运行
- 集群模式：集群内部运行(支持独立模式和集成模式)
三、Pulsar Function的使用
3.1 Pulsar Function的启用
修改Pulsar集群所有服务器的conf/broker.conf，如下内容
functionsWorkerEnabled=true修改Pulsar集群所有服务器的conf/functions_worker.yml，如下内容
pulsarFunctionsCluster: pulsar-cluster然后重启broker服务
注意：三台节点都需要执行，依次都停止，然后依次启动
3.2 使用Pulsar Function
运行官网提供的example包，先在集群模式下创建Function，创建完成的Function是运行的
[root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions create \ > --jar examples/api-examples.jar \ > --classname org.apache.pulsar.functions.api.examples.ExclamationFunction \ > --inputs persistent://public/default/exclamation-input \ > --output persistent://public/default/exclamation-output \ > --tenant public \ > --namespace default \ > --name exclamation "Created successfully" [root@bigdata001 apache-pulsar-2.9.1]#然后触发Function运行，得到结果。原理是向exclamation-input这个topic发送消息，然后消费exclamation-output这个topic的消息
[root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions trigger --name exclamation --trigger-value "hello world" hello world! [root@bigdata001 apache-pulsar-2.9.1]#查看Function状态
[root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions status --name exclamation { "numInstances" : 1, "numRunning" : 1, "instances" : [ { "instanceId" : 0, "status" : { "running" : true, "error" : "", "numRestarts" : 0, "numReceived" : 0, "numSuccessfullyProcessed" : 0, "numUserExceptions" : 0, "latestUserExceptions" : [ ], "numSystemExceptions" : 0, "latestSystemExceptions" : [ ], "averageLatency" : 0.0, "lastInvocationTime" : 0, "workerId" : "c-pulsar-cluster-fw-bigdata003-8086" } } ] } [root@bigdata001 apache-pulsar-2.9.1]#stop Function
[root@bigdata001 apache-pulsar-2.9.1]# [root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions stop --name exclamation Stopped successfully [root@bigdata001 apache-pulsar-2.9.1]#start Function
[root@bigdata001 apache-pulsar-2.9.1]# [root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions start --name exclamation Started successfully [root@bigdata001 apache-pulsar-2.9.1]#delete Function
[root@bigdata001 apache-pulsar-2.9.1]# [root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions delete --name exclamation "Deleted successfully" [root@bigdata001 apache-pulsar-2.9.1]#其它Function使用说明
[root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions [command]属性说明
bin/pulsar-admin functions 属性说明: functions: 可选值: localrun: 创建本地function进行运行 create: 在集群模式下创建 delete: 删除在集群中运行的function get: 获取function的相关信息 restart: 重启 stop : 停止运行 start: 启动 status: 检查状态 stats: 查看状态 list: 查看特定租户和名称空间下的所有的function --classname: 设置function执行类 --jar 设置function对应的jar包 --inputs : 输入的topic --output : 输出的topic --tenant : 设置function运行在那个租户中 --namespace: 设置function运行在那个名称空间中 --name : 定义function的名称四、自己编写一个Function
需求：读取input topic，其中日期格式为yyyy/MM/dd HH/mm/ss，转换为格式yyyy-MM-dd HH:mm:ss，然后发送到output topic
4.1 添加依赖
<dependency> <groupId>org.apache.pulsar</groupId> <artifactId>pulsar-client</artifactId> <version>2.10.0</version> </dependency> <dependency> <groupId>org.apache.pulsar</groupId> <artifactId>pulsar-functions-api</artifactId> <version>2.10.0</version> </dependency>4.2 编写程序
import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; /** * @Author: huangyibo * @Date: 2022/5/28 18:15 * @Description: 读取input topic，其中日期格式为yyyy/MM/dd HH/mm/ss， * 转换为格式yyyy-MM-dd HH:mm:ss，然后发送到output topic */ public class FormatDateFunction implements Function<String, String> { private DateTimeFormatter formatter1 = DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss"); private DateTimeFormatter formatter2 = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); /** * 每来一条消息，都会调用process进行处理 * @param input 输入的消息数据 * @param context 表示上下文对象，用于执行一些相关的统计计算操作，以及获取相关的对象和元数据信息 * @return * @throws Exception */ @Override public String process(String input, Context context) throws Exception { LocalDateTime localDateTime = LocalDateTime.parse(input, formatter1); return localDateTime.format(formatter2); } }4.3 然后将程序进行打包，上传到Pulsar集群中的一台服务器
4.4 创建Function
[root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions create \ > --jar /opt/pulsar_dev-1.0-SNAPSHOT.jar \ > --classname DateTransfromFunction \ > --inputs persistent://public/default/dateTransfrom-input \ > --output persistent://public/default/dateTransfrom-output \ > --tenant public \ > --namespace default \ > --name dateTransfrom "Created successfully" [root@bigdata001 apache-pulsar-2.9.1]#4.5 触发Function
[root@bigdata001 apache-pulsar-2.9.1]# bin/pulsar-admin functions trigger --name dateTransfrom --trigger-value "2022/04/10 16/32/18" 2022-04-10 16:32:18 [root@bigdata001 apache-pulsar-2.9.1]#参考： https://blog.csdn.net/yy8623977/article/details/124072174