Apache Pulsar——Function 轻量级计算框架

一、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 添加依赖

    org.apache.pulsar
    pulsar-client
    2.10.0



    org.apache.pulsar
    pulsar-functions-api
    2.10.0
       

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 {

    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