消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载导读Pulsar Functions 是 Apache Pulsar 内置的轻量级计算框架它允许开发者直接以“函数”的形态编写消息处理逻辑无需再部署独立的流处理系统如 Storm、Flink。本文将基于 Pulsar 2.2.1 版本文档与仓库源码系统讲解 Pulsar Functions 的编程模型、两类原生/SDK 编程接口、命令行部署与配置方式、本地/集群两种运行模式以及日志、用户配置、触发、处理保证、状态存储等核心机制帮助你在 Pulsar 消息总线之上快速构建可投产的流式处理逻辑。Pulsar Functions 是什么核心目标与设计动机Pulsar Functions 是“轻量级计算进程”它做三件事从一个或多个 Pulsar topic消费消息对每条消息应用用户提供的处理逻辑把计算结果发布到另一个 topic。Pulsar 官方文档将其定位为让用户无需部署旁路系统如 Apache Storm、Apache Heron、Apache Flink即可在 Pulsar 消息体系内直接构建任意复杂度的处理逻辑——本质上Pulsar Functions 是随消息系统一起交付的“现成计算基础设施”。围绕这一核心目标还派生出一系列子目标开发者生产力支持语言原生函数Language-native与 Pulsar Functions SDK 两类写法上手成本低易排查函数可向日志 topic 输出日志、发布指标便于定位问题运维简单不需要维护额外的流处理集群降低整体运维负担。从设计渊源看Pulsar Functions 借鉴了两类系统的思路流处理引擎Apache Storm、Apache Heron、Apache Flink的拓扑/算子思想“Serverless / 函数即服务FaaS”云平台AWS Lambda、Google Cloud Functions、Azure Functions的按需计算模型。可以把它概括为“Lambda 风格的函数 以 Pulsar 作为消息总线”。每当你需要“对消息做点什么”时都可以考虑用 Pulsar Functions 就地完成。编程模型输入、处理与三路输出Pulsar Functions 的编程模型非常简洁函数从一个或多个输入 topic接收消息每当收到一条消息函数可以对输入应用处理逻辑并把结果写入Pulsar 中的输出 topic将结果写入Apache BookKeeper状态存储向日志 topic输出日志常用于调试累加counter 计数器。从仓库源码看这一模型由 Function 接口 直接落地FunctionI, O的唯一抽象方法O process(I input, Context context) throws Exception会针对输入 topic 的每条消息被调用一次接口还提供了default void initialize(Context context)函数实例启动时初始化资源与default void close()实例停止时关闭资源两个生命周期钩子它们分别对应函数实例的“启动”与“停止”阶段是编写有状态/有资源开销函数时的扩展点。示例一Exclamation 函数Java 原生接口文档给出的 Java 原生接口native interface示例用java.util.Function实现字符串追加感叹号import java.util.Function; public class ExclamationFunction implements FunctionString, String { Override public String apply(String input) { return String.format(%s!, input); } }Python 的等价实现同样使用原生接口直接定义一个process函数def process(input): return {0}!.format(input)函数每次在输入 topic 收到消息时执行一次。例如函数监听tweet-streamtopic那么每当有消息发布到该 topic函数就会被触发运行一次。示例二词频统计Word count与状态计数器经典的词频统计可以用 Pulsar Functions 很自然地实现输入 topic 提供句子函数对句子分词并利用内置的计数器状态由 BookKeeper 持久化为每个单词累加计数。使用 Java SDK 的写法如下仓库中的完整实现见 WordCountFunction.javapackage 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 { // 每有一条消息发布到输入 topic该函数即被调用一次 Override public Void process(String input, Context context) { Arrays.asList(input.split( )).forEach(word - { String counterKey word.toLowerCase(); context.incrCounter(counterKey, 1) }); return null; } }注意文档中的示例对每个单词做了toLowerCase()归一化the与The视为同一个词仓库源码版本则直接以原始单词为 keycontext.incrCounter(word, 1)你可以按业务需要自由选择。context.incrCounter背后就是基于 Apache BookKeeper 的持久化计数器因此即使函数实例重启计数也不会丢失——这是它与普通局部变量计数器的本质区别。用pulsar-admin在集群中部署该函数集群运行模式见下文$ bin/pulsar-admin functions create \ --jar target/my-jar-with-dependencies.jar \ --classname org.example.functions.WordCountFunction \ --tenant public \ --namespace default \ --name word-count \ --inputs persistent://public/default/sentences \ --output persistent://public/default/count示例三基于内容的路由Python内容路由是更复杂的典型场景函数根据输入内容把消息发布到不同的输出 topic。下图展示了“水果/蔬菜/都不是”三类路由Python SDK 实现如下from pulsar import Function class RoutingFunction(Function): def __init__(self): self.fruits_topic persistent://public/default/fruits self.vegetables_topic persistent://public/default/vegetables staticmethod def is_fruit(item): return item in [apple, orange, pear, other fruits...] staticmethod def is_vegetable(item): return item in [carrot, lettuce, radish, other vegetables...] def process(self, item, context): if self.is_fruit(item): context.publish(self.fruits_topic, item) elif self.is_vegetable(item): context.publish(self.vegetables_topic, item) else: warning The item {0} is neither a fruit nor a vegetable.format(item) context.get_logger().warn(warning)这里用到的context.publish(topic, item)会经 SerDe 序列化后发布到指定 topiccontext.get_logger().warn(...)则把警告写入函数配置的日志 topic。命令行接口pulsar-admin 与 pulsar-functionsPulsar Functions 通过pulsar-adminCLI 工具管理核心子命令是functions。一个在本地运行模式下启动函数的示例$ bin/pulsar-functions localrun \ --inputs persistent://public/default/test_src \ --output persistent://public/default/test_result \ --jar examples/api-examples.jar \ --classname org.apache.pulsar.functions.api.examples.ExclamationFunction这里的org.apache.pulsar.functions.api.examples.ExclamationFunction是仓库中真实存在的示例类见 ExclamationFunction.java其process实现正是String.format(%s!, input)。相关pulsar-admin functions子命令还包括create集群模式部署、trigger触发函数、list列出函数等。全限定函数名FQFN每个 Pulsar Function 都有一个全限定函数名Fully Qualified Function NameFQFN由三部分组成tenant/namespace/name例如public/default/word-count。FQFN 使得不同命名空间下可以存在同名函数而互不冲突。结合 Pulsar 的多租户模型概念详见 concepts-multi-tenancy.mdFQFN 是函数在集群中的唯一寻址标识。配置方式CLI 参数与 YAML 文件Pulsar Functions 支持两种配置方式且可以混用命令行参数通过pulsar-admin functions接口传入YAML 配置文件通过--function-config-file指定路径。YAML 方式示例$ bin/pulsar-admin functions create \ --function-config-file ./my-function.yaml对应的my-function.yamlname: my-function tenant: public namespace: default jar: ./target/my-functions.jar className: org.example.pulsar.functions.MyFunction inputs: - persistent://public/default/test_src output: persistent://public/default/test_result你完全可以“混搭”一部分属性走 CLI、另一部分走 YAML。此外部署与管理工作负载时还支持默认参数机制——许多参数缺省时会被自动推导例如函数名缺省取类名--classname org.example.MyFunction的函数名即为MyFunctiontenant/namespace 缺省时从输入 topic 名推导输出 topic 缺省为{input topic}-{function name}-output处理保证缺省为ATLEAST_ONCE服务地址缺省为pulsar://localhost:6650。支持的编程语言与两类 APIPulsar Functions 目前支持Java与Python两种语言。对应的 API 分两类Pulsar Functions APISDKAPI 面向类型安全与 SerDe类型安全函数既可以处理原始字节也可以处理应用自定义的复杂类型基于 SerDe序列化/反序列化多种内置类型开箱即用也支持自定义 SerDe。SDK 的核心类型定义在 pulsar-functions/api-java 目录下包括FunctionI, O核心函数接口、Context函数上下文、SerDe序列化接口、Record消息记录、StateStore状态存储接口等。函数上下文Function context使用 SDK 创建的每个函数都能访问一个context 对象它提供两类能力函数自身信息函数名、tenant、namespace、用户配置user configuration等特殊功能向指定日志 topic 输出日志、发布指标metrics。从源码看Context 接口 还提供getInputTopics()所有输入 topic、getOutputTopic()输出 topic、getUserConfigMap()/getUserConfigValue(key)/getUserConfigValueOrDefault(key, defaultValue)读取用户配置、publish(topicName, object)发布消息、newOutputMessage(topicName, schema)带 Schema 的发布、getPulsarAdmin()Pulsar Admin 客户端等能力。语言原生函数Language-native functionsJava 与 Python 都支持编写无任何依赖的“原生”函数如前面的 Exclamation 示例。其优点是零外部依赖、即写即用缺点是无法访问 context因而无法使用日志、用户配置、计数器等高级能力。需要这些能力时请使用 SDK 版本。SDK 示例读取上下文的函数Java使用 SDK 访问上下文信息import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; import org.slf4j.Logger; public class ContextAwareFunction implements FunctionString, Void { Override public Void process(String input, Context, context) { Logger LOG context.getLogger(); String functionTenant context.getTenant(); String functionNamespace context.getNamespace(); String functionName context.getName(); LOG.info(Function tenant/namespace/name: {}/{}/{}, functionTenant, functionNamespace, functionName); return null; } }Python 等价实现from pulsar import Function class ContextAwareFunction(Function): def process(self, input, context): log context.get_logger() function_tenant context.get_function_tenant() function_namespace context.get_function_namespace() function_name context.get_function_name() log.info(Function tenant/namespace/name: {0}/{1}/{2}.format(function_tenant, function_namespace, function_name))部署模式本地运行与集群运行Pulsar Functions 支持两种部署模式部署模式说明本地运行模式函数运行在本地环境例如你的笔记本上集群运行模式函数运行在 Pulsar 集群内部、与 Pulsar broker 同机本地运行模式Local run mode本地运行模式下函数运行在执行命令的那台机器上可能是笔记本、EC2 实例等$ bin/pulsar-admin functions localrun \ --py myfunc.py \ --classname myfunc.SomeFunction \ --inputs persistent://public/default/input-1 \ --output persistent://public/default/output-1默认情况下函数会连接本机上的 Pulsar 集群broker 服务地址pulsar://localhost:6650。若要连接非本地的 Pulsar 集群可用--broker-service-url指定$ bin/pulsar-admin functions localrun \ --broker-service-url pulsar://my-cluster-host:6650 \ # 其他函数参数集群运行模式Cluster run mode集群模式下函数代码被上传到 Pulsar broker并与 broker 一起运行而不是跑在你的本地环境。用create命令部署$ bin/pulsar-admin functions create \ --py myfunc.py \ --classname myfunc.SomeFunction \ --inputs persistent://public/default/input-1 \ --output persistent://public/default/output-1该命令会把myfunc.py上传到 Pulsar随后 Pulsar 依据代码启动一个或按需多个函数实例。并行度Parallelism集群模式下默认只启动1 个函数实例但你可以通过并行度参数运行多个实例。例如创建并行度为 5即 5 个实例的函数$ bin/pulsar-admin functions create \ --name parallel-fun \ --tenant public \ --namespace default \ --py func.py \ --classname func.ParallelFunction \ --parallelism 5并行度既可以在创建时指定也可以事后更新已有的单实例函数。函数实例资源Function instance resources集群运行模式下可以为每个函数实例分配资源资源指定方式运行时CPU核数Docker即将支持RAM字节数进程、Docker磁盘空间字节数Docker示例为函数分配 8 核、8 GB 内存、10 GB 磁盘$ bin/pulsar-admin functions create \ --jar target/my-functions.jar \ --classname org.example.functions.MyFunction \ --cpu 8 \ --ram 8589934592 \ --disk 10737418240更多资源相关说明可参考 Deploying and Managing Pulsar Functions。日志机制使用 SDK 创建的 Pulsar Functions 可以把日志发送到函数配置中指定的日志 topic。例如下面的命令会把该函数的所有日志输出到persistent://public/default/my-func-1-log$ bin/pulsar-admin functions create \ --name my-func-1 \ --log-topic persistent://public/default/my-func-1-log \ # 其他配置Java 函数中按输入内容在不同日志级别输出仓库中有对应的 LoggingFunction.java 示例public class LoggerFunction implements FunctionString, Void { Override public Void process(String input, Context context) { Logger LOG context.getLogger(); if (input.length() 100) { LOG.info(This string has a length of {}, input); } else { LOG.warn(This string is getting too long! It has {} characters, input); } } }把日志集中到 Pulsar topic意味着你可以用既有的消息消费链路甚至另一个 Pulsar Function统一收集和分析函数日志而不必各自接入日志系统。用户配置User configurationPulsar Functions 支持通过命令行传入任意 key-value键和值都必须是字符串这组键值对称为函数的用户配置配置内容必须是 JSON 字符串。示例$ bin/pulsar-admin functions create \ --user-config {key-1:value-1,key-2,value-2} \ # 其他配置函数内通过 context 读取配置public class ConfigMapFunction implements FunctionString, Void { Override public Void process(String input, Context context) { String val1 context.getUserConfigValue(key1).get(); String val2 context.getUserConfigValue(key2).get(); context.getLogger().info(The user-supplied values are {} and {}, val1, val2); return null; } }从 Context 接口 的实现看读取配置有三种方式getUserConfigMap()全量 Map、getUserConfigValue(key)返回Optional需要.get()取值以及getUserConfigValueOrDefault(key, defaultValue)带默认值兜底。这意味着无需修改函数代码即可通过 CLI 动态改变函数行为如阈值、目标 topic 名等是一种典型的“配置与代码分离”实践。仓库示例 UserConfigFunction.java 展示了完整用法。触发函数Triggering集群模式下运行的函数可以被触发trigger——即通过 CLI 直接向函数传入一个值并拿到返回值无需创建客户端、向输入 topic 发消息等繁琐步骤。触发特别适合测试与调试。从机制上讲触发一个函数与“向函数输入 topic 生产一条消息”并无本质区别pulsar-admin functions trigger命令本质上只是“向函数发消息”的便捷机制省去了使用pulsar-client或语言客户端库的步骤。例如下面这个反转字符串的 Python 原生函数def process(input): return input[::-1]在集群中运行时可以这样触发$ bin/pulsar-admin functions trigger \ --tenant public \ --namespace default \ --name reverse-func \ --trigger-value snoitcnuf raslup ot emoclew控制台输出应为welcome to pulsar functions。除了命令行传字符串也可以用--triggerFile指定文件内容作为触发输入。处理保证Processing guaranteesPulsar Functions 提供三种消息语义可应用到任意函数投递语义说明At-most-once至多一次发到函数的每条消息“很可能被处理”但也可能不被处理所以是“至多”At-least-once至少一次发到函数的每条消息可能被处理多次所以是“至少”Effectively-once恰好一次发到函数的每条消息对应一个输出结果例如以 effectively-once 语义在集群模式运行函数$ bin/pulsar-admin functions create \ --name my-effectively-once-function \ --processing-guarantees EFFECTIVELY_ONCE \ # 其他函数配置三种语义的具体原理与取舍可进一步参考 Processing guarantees for Pulsar Functions 文档。指标Metrics与状态存储State storage指标使用 Pulsar Functions SDK 的函数可以向 Pulsar 发布指标如函数处理的消息总数、处理时长等供监控系统采集分析。详见 Metrics for Pulsar Functions。状态存储Pulsar Functions 使用Apache BookKeeper作为状态存储接口incrCounter之类的计数器即持久化于此。所有 Pulsar 安装包括本地 standalone 单机安装都自带 BookKeeper bookie 部署因此无需额外引入外部状态系统即可获得持久化、一致性的函数状态。更完整的 API 说明见 Pulsar Functions API状态相关深入内容见 Pulsar Functions state storage。快速上手路径与延伸阅读如果你想立刻动手阅读 Functions quick start在一个 standalone 集群上跑通第一个函数系统学习 Pulsar Functions APIJava/Python 的 SDK 与原生函数写法深入 Deploying and managing Pulsar Functions 掌握资源分配、触发、更新/删除等运维操作仓库中 pulsar-functions/java-examples 提供了 38 个可直接阅读的示例函数词频统计、日志、用户配置、发布、路由、窗口等是学习不同用法的最佳范本相关 Python 示例见 pulsar-functions/python-examples 目录如exclamation_function.py、wordcount_function.py等。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar Functions 部署与管理实战本地运行、集群模式、并行度与触发机制Apache Pulsar Functions 部署与管理实战本地运行、集群模式、并行度与触发机制 导读 Pulsar Functions 是 Apache消息队列后端流处理Apache Pulsar Functions 深度解析编程模型、部署命令与处理保证机制Apache Pulsar Functions 深度解析编程模型、部署命令与处理保证机制 Pulsar Functions 是构建在 Pulsar 之上的轻量消息队列后端流处理Apache Pulsar Functions 部署实战从本地运行到集群模式的完整指南Apache Pulsar Functions 部署实战从本地运行到集群模式的完整指南 本文基于 Apache Pulsar 官方文档 functions d消息队列后端流处理上一篇为什么选择vscode-python深度解析其核心功能与优势下一篇Strata开源项目推荐创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
