Apache Pulsar Functions 打包实战指南Java、Python、Go 三种语言的完整打包与部署流程【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址: https://gitcode.com/gh_mirrors/pulsar28/pulsar本文围绕 Apache Pulsar 中 Pulsar Functions 的打包方法展开系统讲解 JavaMaven 装配式 JAR、Python单文件 / ZIP / PIP 三种格式与 GoSDK 反射校验三种语言的函数打包流程以及localrun本地运行与create集群模式下pulsar-admin functions命令的完整用法。读完本文你将能够从零编写、打包、部署一个可运行的 Pulsar Function并理解函数运行时ProcessRuntime如何加载你打出的制品。Pulsar Functions 是 Apache Pulsar 内置的轻量级流处理框架允许用户用 Java、Python 或 Go 编写处理函数消费一个或多个输入 topic 中的消息并把处理结果写入输出 topic。无论函数最终运行在本地进程、函数工作节点Function Worker还是 Kubernetes Pod 中第一步都是把函数代码打包成运行时能够加载的制品对 Java 函数来说通常是包含全部依赖的 fat JAR对 Python 函数来说是单.py文件、ZIP 包或whl包对 Go 函数来说则是一个独立的可执行二进制。本文内容以官方文档 functions-package.md 为核心骨架并结合本仓库源码给出底层实现依据。前置条件先启动一个可用的 Pulsar 实例在运行任何 Pulsar Function 之前需要先启动 Pulsar 服务。官方提供两种常用方式在 Docker 中运行 Standalone 模式的 Pulsar参见 getting-started-docker.md在 Kubernetes 中通过 Helm 部署 Pulsar参见 getting-started-helm.md。启动后可以使用docker ps命令确认 Docker 镜像是否正常启动docker ps另外运行本文后续命令前请确认pulsar-admin命令可用位于 Pulsar 安装目录的./bin/pulsar-admin函数的完整命令行参数说明可参考 reference-pulsar-admin.md#functions 或直接执行./bin/pulsar-admin functions打包 Java 函数Maven fat JAR 全流程1. 创建 Maven 项目与 POM 配置新建一个 Maven 项目并创建pom.xml。下面的示例中mainClass的值需要替换为你自己的包名这里是org.example.test.ExclamationFunction?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdjava-function/groupId artifactIdjava-function/artifactId version1.0-SNAPSHOT/version dependencies dependency groupIdorg.apache.pulsar/groupId artifactIdpulsar-functions-api/artifactId version2.6.0/version /dependency /dependencies build plugins plugin artifactIdmaven-assembly-plugin/artifactId configuration appendAssemblyIdfalse/appendAssemblyId descriptorRefs descriptorRefjar-with-dependencies/descriptorRef /descriptorRefs archive manifest mainClassorg.example.test.ExclamationFunction/mainClass /manifest /archive /configuration executions execution idmake-assembly/id phasepackage/phase goals goalassembly/goal /goals /execution /executions /plugin plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-compiler-plugin/artifactId configuration source8/source target8/target /configuration /plugin /plugins /build /project关键配置说明pulsar-functions-api依赖提供 Pulsar Function 编程接口对应本仓库的 pulsar-functions/api-java 模块其中包含org.apache.pulsar.functions.api.Function与Context等核心接口。maven-assembly-plugin的jar-with-dependencies将项目自身的类与所有第三方依赖合并打包成一个可独立运行的 fat JAR这是 Pulsar Functions 运行时能够直接加载的前提——运行时通过 JAR 的Main-Class/类路径信息定位用户函数类。appendAssemblyIdfalse使最终产物的文件名不带jar-with-dependencies后缀例如直接生成java-function-1.0-SNAPSHOT.jar。maven-compiler-plugin指定source/target为 8保证字节码兼容性Pulsar Functions 运行环境通常基于 Java 8 构建。2. 编写 Java 函数两种 Function 接口Java 函数可以使用两种函数接口编写Java 8 原生函数式接口java.util.function.FunctionPulsar 自带的函数接口org.apache.pulsar.functions.api.Function使用 Java 8 接口的示例package org.example.test; import java.util.function.Function; public class ExclamationFunction implements FunctionString, String { Override public String apply(String s) { return This is my function!; } }两种接口最核心的区别在于org.apache.pulsar.functions.api.Function接口额外提供了Context上下文。当你需要在函数内与 Pulsar 交互时可以通过context获取 Pulsar Functions 提供的丰富信息与能力例如计数器counter、日志、用户配置、发布消息到其他 topic 等。使用 Pulsar 接口并携带Context的示例一个经典单词计数函数package org.example.functions; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; import java.util.Arrays; public class WordCountFunction implements FunctionString, Void { // This function is invoked every time a message is published to the input topic Override public Void process(String input, Context context) throws Exception { Arrays.asList(input.split( )).forEach(word - { String counterKey word.toLowerCase(); context.incrCounter(counterKey, 1); }); return null; } }该示例中context.incrCounter(counterKey, 1)会对每个单词的 key 累加计数计数状态由 Pulsar Functions 的状态存储State Store持久化这正是 Pulsar 函数接口相对 Java 8 原生接口多出来的能力。仓库中提供了丰富的 Java 函数示例可供参考例如 pulsar-functions/java-examples 目录下的 50 个示例类。3. 执行打包在项目根目录执行mvn package打包完成后项目目录下会自动生成target目录。进入target目录检查是否生成了形如java-function-1.0-SNAPSHOT.jar的 JAR 包该包就是后续要提交给运行时的制品。4. 运行 Java 函数1将打包好的 JAR 复制到 Pulsar 容器内docker exec -it [CONTAINER ID] /bin/bash docker cp path of java-function-1.0-SNAPSHOT.jar CONTAINER ID:/pulsar2使用pulsar-admin functions localrun以本地模式运行函数./bin/pulsar-admin functions localrun \ --classname org.example.test.ExclamationFunction \ --jar java-function-1.0-SNAPSHOT.jar \ --inputs persistent://public/default/my-topic-1 \ --output persistent://public/default/test-1 \ --tenant public \ --namespace default \ --name JavaFunction参数含义--classname函数类的全限定名--jarfat JAR 的路径--inputs输入 topic支持多个逗号分隔--output输出 topic--tenant/--namespace函数所属租户与命名空间默认分别为public与default--name函数的唯一名称。看到如下日志即表示 Java 函数启动成功... 07:55:03.724 [main] INFO org.apache.pulsar.functions.runtime.ProcessRuntime - Started process successfully ...这条日志来自org.apache.pulsar.functions.runtime.ProcessRuntime它表明函数被以独立进程Process Runtime的方式拉起并成功启动。Pulsar Functions 支持多种运行时实现localrun默认使用进程运行时在本地直接拉起函数进程无需额外的容器环境。打包 Python 函数三种格式详解Python Function 支持以下三种打包格式单个 Python 文件one python fileZIP 压缩包ZIP filePIP 包whl 文件仅支持 Kubernetes 运行时注意窗口函数Window Function目前不支持 Python 与 Go仅支持 JavaJava 窗口函数的打包方式与普通 Java 函数完全一致。格式一单个 Python 文件1. 编写 Python 函数。需要继承pulsar.Function类并实现process()方法from pulsar import Function // import the Function module from Pulsar # The classic ExclamationFunction that appends an exclamation at the end # of the input class ExclamationFunction(Function): def __init__(self): pass def process(self, input, context): return input !process()方法主要有两个参数input函数的输入消息contextPulsar Function 暴露的上下文接口基于该 context 对象可以在 Python 函数中获取各种属性消息元数据、用户配置、计数器等。本仓库 pulsar-functions/python-examples/exclamation_function.py 中保存着与该示例完全一致的实现此外该目录下还有wordcount_function.py、logging_function.py、publish_function.py等 11 个可直接参考的官方示例。2. 安装 Python 客户端。Python 函数的实现依赖 Python 客户端库因此在部署前需要安装对应版本的客户端pip install pulsar-client2.6.03. 运行 Python 函数。1将 Python 函数文件复制到 Pulsar 容器docker exec -it [CONTAINER ID] /bin/bash docker cp path of Python function file CONTAINER ID:/pulsar2使用--py参数运行./bin/pulsar-admin functions localrun \ --classname Python Function file name.Python Function class name \ --py path of Python Function file \ --inputs persistent://public/default/my-topic-1 \ --output persistent://public/default/test-1 \ --tenant public \ --namespace default \ --name PythonFunction注意此时--classname的写法是“文件名.类名”例如文件名为exclamation.py、类名为ExclamationFunction则应写为exclamation.ExclamationFunction。同样看到ProcessRuntime - Started process successfully日志即表示启动成功。格式二ZIP 文件1. 准备 ZIP 文件。假设 ZIP 名为func.zip解压后必须包含以下结构func/src func/requirements.txt func/depssrc存放函数源码requirements.txt声明 Python 依赖deps存放预打包的依赖如whl文件在函数启动时被安装/加载。仓库tests/docker-images下的python-examples目录提供了一份exclamation.zip官方示例其内部结构如下. ├── deps │ └── sh-1.12.14-py2.py3-none-any.whl └── src └── exclamation.py2. 运行 Python 函数。1将 ZIP 文件复制到 Pulsar 容器docker exec -it [CONTAINER ID] /bin/bash docker cp path of ZIP file CONTAINER ID:/pulsar2运行命令注意此时--classname只写模块名不写类名./bin/pulsar-admin functions localrun \ --classname exclamation \ --py path of ZIP file \ --inputs persistent://public/default/in-topic \ --output persistent://public/default/out-topic \ --tenant public \ --namespace default \ --name PythonFunction看到ProcessRuntime - Started process successfully日志即表示启动成功。格式三PIP仅 Kubernetes 运行时PIP 方式只在 Kubernetes 运行时下受支持。完整步骤如下1. 配置functions_worker.yml。在 Function Worker 配置文件的 Kubernetes Runtime 段中开启用户代码依赖安装#### Kubernetes Runtime #### installUserCodeDependencies: true该配置项同样存在于本仓库的 conf/functions_worker.yml 中注释中明确说明其作用是 The flag indicates to install user code dependencies. (applied to python package)。当 Python Function 检测到提交的制品是whl且installUserCodeDependencies被置为true时系统会使用pip install命令安装函数所需的全部依赖。2. 编写 Python 函数。除了 Pulsar 客户端外可以引入任意额外依赖from pulsar import Function import js2xml # The classic ExclamationFunction that appends an exclamation at the end # of the input class ExclamationFunction(Function): def __init__(self): pass def process(self, input, context): // add your logic return input !3. 生成whl文件。使用仓库自带的打包脚本$ cd $PULSAR_HOME/pulsar-functions/scripts/python $ chmod x generate.sh $ ./generate.sh path of your Python Function path of the whl output dir the version of whl # e.g: ./generate.sh /path/to/python /path/to/python/output 1.0.0输出会写入指定的输出目录-rw-r--r-- 1 root staff 1.8K 8 27 14:29 pulsarfunction-1.0.0-py2-none-any.whl -rw-r--r-- 1 root staff 1.4K 8 27 14:29 pulsarfunction-1.0.0.tar.gz -rw-r--r-- 1 root staff 0B 8 27 14:29 pulsarfunction.whl从源码看generate.sh 的实际逻辑是把path of your Python Function目录下的所有*.py复制到临时目录的pulsarfunction包中通过sed将 setup.py.template 中的VERSION占位符替换为第三个参数指定的版本号并把requirements.txt一并复制进去随后依次执行setup.py sdist与setup.py bdist_wheel生成tar.gz源码包与whlwheel 包最后在输出目录中touch一个空的pulsarfunction.whl作为 PIP 格式的标识文件。这也解释了为何需要提前在函数目录中准备好requirements.txt——它是 wheel 包install_requires元数据的来源。打包 Go 函数SDK 反射签名校验1. 编写 Go 函数Go 函数只能通过官方 SDK 实现函数接口以 SDK 形式暴露。使用前需要导入github.com/apache/pulsar/pulsar-function-go/pf包对应本仓库 pulsar-function-go/pf 目录import ( context fmt github.com/apache/pulsar/pulsar-function-go/pf ) func HandleRequest(ctx context.Context, input []byte) error { fmt.Println(string(input) !) return nil } func main() { pf.Start(HandleRequest) }可以使用pf.FromContext从 context 中取出函数上下文并访问函数元信息if fc, ok : pf.FromContext(ctx); ok { fmt.Printf(function ID is:%s, , fc.GetFuncID()) fmt.Printf(function version is:%s\n, fc.GetFuncVersion()) }从源码看pf.FromContext定义于 pulsar-function-go/pf/context.go#L192而pf.Start定义于 pulsar-function-go/pf/function.go#L168它负责启动函数实例并与 Function Worker 建立 gRPC 通信。编写 Go 函数时需要注意两个关键约束在main()中只需将函数名注册给Start()且Start()只接收一个函数名参数Go 函数基于传入的函数名使用Go 反射reflection校验参数列表与返回值列表是否合法。参数与返回值组合必须是下列 9 种形式之一func () func () error func (input) error func () (output, error) func (input) (output, error) func (context.Context) error func (context.Context, input) error func (context.Context) (output, error) func (context.Context, input) (output, error)这 9 种签名是 Go 函数运行时反射校验的完整白名单输入可选无或input必须的context.Context可放在最前返回值必须包含error输出output可选。2. 构建 Go 函数在函数目录下直接编译生成可执行二进制go build your Go Function filename.go3. 运行 Go 函数1将编译出的二进制复制到 Pulsar 容器docker exec -it [CONTAINER ID] /bin/bash docker cp your go function path CONTAINER ID:/pulsar2使用--go参数运行./bin/pulsar-admin functions localrun \ --go [your go function path] --inputs [input topics] \ --output [output topic] \ --tenant [default:public] \ --namespace [default:default] \ --name [custom unique go function name]看到如下日志即表示 Go 函数启动成功... 07:55:03.724 [main] INFO org.apache.pulsar.functions.runtime.ProcessRuntime - Started process successfully ...在集群模式下启动函数如果希望以集群模式cluster mode启动函数只需把前面所有命令中的localrun替换为create即可。例如./bin/pulsar-admin functions create \ --classname org.example.test.ExclamationFunction \ --jar java-function-1.0-SNAPSHOT.jar \ --inputs persistent://public/default/my-topic-1 \ --output persistent://public/default/test-1 \ --tenant public \ --namespace default \ --name JavaFunction与localrun不同的是create会把函数提交给 Function Worker 进行调度部署而不是在当前机器上拉起进程返回以下日志表示创建成功Created successfullylocalrun与create的主要区别总结模式命令运行位置适用场景本地模式localrun当前进程环境直接拉起函数进程本地调试、快速验证集群模式create提交至 Function Worker / Kubernetes 调度生产环境部署关于--classname、--jar、--py、--go、--inputs等参数的完整说明可运行./bin/pulsar-admin functions查看帮助或参阅 reference-pulsar-admin.md#functions。小结本文完整覆盖了 Pulsar Functions 三种语言的打包与部署路径Java 函数通过maven-assembly-plugin打出带全部依赖的 fat JARPython 函数可根据运行环境在单文件、ZIP携带src/deps/requirements.txt与 PIPinstallUserCodeDependencies: truegenerate.sh生成 wheel三种格式中选择Go 函数则编译为二进制后通过 SDK 注册并遵循 9 种反射签名白名单。无论哪种语言最终都通过pulsar-admin functions以localrun本地或create集群两种方式提交给 Pulsar Functions 运行时由ProcessRuntime完成进程拉起。掌握这套打包流程后你可以在此基础上进一步探索Context的高级用法计数器、状态存储、用户配置以及窗口函数仅 Java等更复杂的能力。【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址: https://gitcode.com/gh_mirrors/pulsar28/pulsar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考