一、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
网友评论