消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本指南基于 Apache Pulsar 开源仓库系统讲解 Java SDK 为窗口函数Window Function提供的窗口上下文对象WindowContext。窗口函数与普通函数的最大区别在于它接收的是一个CollectionRecordT一批消息组成的窗口因此需要一个专门定制的上下文对象来暴露函数元信息、日志、用户配置、路由发布、指标上报与状态存储能力。读完本文你将掌握WindowContext全部核心 API 的用法并理解其底层在 WindowContextImpl.java 与 WindowFunctionExecutor.java 中的实现机制。窗口上下文对象能做什么Java SDK 为窗口函数提供了一个窗口上下文对象window context object它是对普通函数Context的面向窗口场景的封装提供以下几大类信息与功能Spec函数规格函数关联的全部输入 Topic 名称列表、输出 Topic 名称、所属租户tenant与命名空间namespace、窗口函数名称/ID/版本、运行窗口函数的实例 ID、调用窗口函数的实例总数、输出 Schema 的内置类型或自定义类名。Logger窗口函数使用的日志对象用于产生日志消息。User config访问任意的用户自定义配置键值对。Routing窗口函数支持路由——通过publish接口向任意 Topic 发送消息。Metrics记录指标metric的接口。State storage在状态存储中存取状态的接口。从源码角度看WindowContext 是一个标注为InterfaceAudience.Public、InterfaceStability.Stable的公开稳定接口全部能力都通过方法签名暴露而 WindowContextImpl 是其默认实现内部直接委托给普通函数的Context对象完成实际逻辑——也就是说窗口上下文的每一项能力最终都复用了 Pulsar Functions 运行时为普通函数提供的基础设施。Spec读取窗口函数的规格信息Spec 包含一个函数的基础信息。WindowContext中提供了一组getXxx方法分别获取输入/输出 Topic、租户、命名空间、函数名、函数 ID、函数版本、实例 ID、实例总数与输出 Schema 类型。获取输入 Topic 列表getInputTopics()方法返回所有输入 Topic 的名称列表public class GetInputTopicsWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { CollectionString inputTopics context.getInputTopics(); System.out.println(inputTopics); return null; } }获取输出 TopicgetOutputTopic()方法获取消息发送目标 Topic 的名称public class GetOutputTopicWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { String outputTopic context.getOutputTopic(); System.out.println(outputTopic); return null; } }获取租户与命名空间getTenant()返回窗口函数所属的租户名称getNamespace()返回其所属的命名空间public class GetTenantWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { String tenant context.getTenant(); System.out.println(tenant); return null; } } public class GetNamespaceWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { String ns context.getNamespace(); System.out.println(ns); return null; } }获取函数名称与 IDgetFunctionName()获取窗口函数名称getFunctionId()获取窗口函数 IDpublic class GetNameOfWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { String functionName context.getFunctionName(); System.out.println(functionName); return null; } } public class GetFunctionIDWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { String functionID context.getFunctionId(); System.out.println(functionID); return null; } }获取函数版本getFunctionVersion()获取窗口函数的版本号函数每次更新后版本都会变化可用于识别运行中的函数版本public class GetVersionOfWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { String functionVersion context.getFunctionVersion(); System.out.println(functionVersion); return null; } }获取实例 ID 与实例总数getInstanceId()返回当前运行窗口函数的实例 IDgetNumInstances()返回调用该窗口函数的实例总数。这两个方法配合使用可以判断当前实例在全部并行实例中的位置public class GetInstanceIDWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { int instanceId context.getInstanceId(); System.out.println(instanceId); return null; } } public class GetNumInstancesWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { int numInstances context.getNumInstances(); System.out.println(numInstances); return null; } }获取输出 Schema 类型getOutputSchemaType()返回输出 Schema 的内置类型名或自定义类名例如avro、json、protobuf等内置类型或自定义 Schema 类的全限定类名public class GetOutputSchemaTypeWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { String schemaType context.getOutputSchemaType(); System.out.println(schemaType); return null; } }实现细节以上所有 getter 在 WindowContextImpl 中都是一行式的委托调用直接转发给底层普通函数的Context对象例如getTenant()实现为return this.context.getTenant();。这说明窗口上下文与普通函数上下文共享同一套运行时元数据只是以窗口语义重新暴露给用户。Logger在窗口函数中打日志使用 Java SDK 的 Pulsar 窗口函数可以获取一个 SLF4JLogger对象用于在指定日志级别输出日志。下面的例子遍历窗口内每条记录并输出INFO级别日志import java.util.Collection; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.functions.api.WindowContext; import org.apache.pulsar.functions.api.WindowFunction; import org.slf4j.Logger; public class LoggingWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { Logger log context.getLogger(); for (RecordString record : inputs) { log.info(record -window-log); } return null; } }如果希望函数产生可被外部订阅消费的日志需要在创建或运行函数时指定日志 Topiclog topicbin/pulsar-admin functions create \ --jar my-functions.jar \ --classname my.package.LoggingFunction \ --log-topic persistent://public/default/logging-function-logs \ # Other function configs创建之后LoggingFunction产生的全部日志都可以通过persistent://public/default/logging-function-logs这个 Topic 访问。Metrics按 key 记录用户指标Pulsar 窗口函数可以将任意指标发布到可查询的 metrics 接口。:::note 语言原生接口的局限 如果窗口函数使用 Java 的语言原生接口language-native interface则该函数无法向 Pulsar 发布指标与统计信息。也就是说要使用recordMetric上报指标必须通过 SDK 接口实现函数。 :::指标可以基于key键粒度记录。下面的例子在每次处理消息时将消息的事件时间记录到名为MessageEventTime的指标中import java.util.Collection; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.functions.api.WindowContext; import org.apache.pulsar.functions.api.WindowFunction; /** * Example function that wants to keep track of * the event time of each message sent. */ public class UserMetricWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { for (RecordString record : inputs) { if (record.getEventTime().isPresent()) { context.recordMetric(MessageEventTime, record.getEventTime().get().doubleValue()); } } return null; } }recordMetric(String metricName, double value)是 WindowContext 中声明的方法指标值统一使用double类型指标最终由函数运行时聚合可通过 Prometheus 等监控体系查询。User config读取用户自定义配置当使用 SDK 创建或更新 Pulsar Functions 时可以通过--user-config标志传入任意键值对键值对必须以 JSON 格式指定bin/pulsar-admin functions create \ --name word-filter \ --user-config {forbidden-word:rosebud} \ # Other function configs三个用户配置 API窗口上下文为读取用户自定义信息提供了三个 APIgetUserConfigMap获取函数全部用户自定义键值对的 Map/** * Get a map of all user-defined key/value configs for the function. * * return The full map of user-defined config values */ MapString, Object getUserConfigMap();getUserConfigValue按 key 获取单个用户自定义键值对返回Optional包装/** * Get any user-defined key/value. * * param key The key * return The Optional value specified by the user for that key. */ OptionalObject getUserConfigValue(String key);getUserConfigValueOrDefault按 key 获取用户自定义值若不存在则返回默认值/** * Get any user-defined key/value or a default value if none is present. * * param key * param defaultValue * return Either the user config value associated with a given key or a supplied default value */ Object getUserConfigValueOrDefault(String key, Object defaultValue);实战示例Java SDK 的上下文对象支持通过命令行以 JSON 形式访问传入窗口函数的键值对。先创建一个带用户配置的函数bin/pulsar-admin functions create \ --user-config {word-of-the-day:verdure} \ # Other function configs:::tip 类型约定 对于传入 Java 窗口函数的所有键值对key 和 value 都是String类型。如果需要把 value 作为其他类型使用需要自行从String类型反序列化。 :::下面这个UserConfigWindowFunction每次被调用即每条消息到达时都会读取WhatToWrite键对应的值如果存在则返回该值否则返回固定字符串。用户配置只有在通过命令行工具或 REST API 等途径更新函数时才会改变import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; import org.slf4j.Logger; import java.util.Optional; public class UserConfigWindowFunction implements WindowFunctionString, String { Override public String process(CollectionRecordString input, WindowContext context) throws Exception { OptionalObject whatToWrite context.getUserConfigValue(WhatToWrite); if (whatToWrite.get() ! null) { return (String)whatToWrite.get(); } else { return Not a nice way; } } }如果没有提供配置值可以获取完整的用户配置 Map或者为读取操作指定一个默认值// Get the whole config map MapString, String allConfigs context.getUserConfigMap(); // Get value or resort to default String wotd context.getUserConfigValueOrDefault(word-of-the-day, perspicacious);实现细节从 WindowFunctionExecutor.getWindowConfigs 可以看到窗口函数自身的窗口参数窗口长度、滑动间隔等也是以WindowConfig.WINDOW_CONFIG_KEY为 key 存放在用户配置中的运行时通过 Gson 反序列化为 WindowConfig含windowLengthCount、windowLengthDurationMs、slidingIntervalCount、slidingIntervalDurationMs、lateDataTopic、maxLagMs、watermarkEmitIntervalMs、timestampExtractorClassName、actualWindowFunctionClassName等字段。这与用户自定义配置走的是同一条通路因此getUserConfigMap()返回的 Map 中同样包含窗口配置项。Routing通过 publish 接口向任意 Topic 发送消息窗口函数可以使用context.publish()接口发布任意数量的结果。publish(String topicName, O object)使用默认 Schema 向指定 Topic 发布对象publish(String topicName, O object, String schemaOrSerdeClassName)则允许显式指定内置 Schema 类型如avro、json、protobuf或自定义 Schema 类名见 WindowContext。下面的PublishWindowFunction从用户配置中读取目标 Topic 名默认值为publishtopic把每个输入窗口的内容拼接后发布到该 Topicpublic class PublishWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString input, WindowContext context) throws Exception { String publishTopic (String) context.getUserConfigValueOrDefault(publish-topic, publishtopic); String output String.format(%s!, input); context.publish(publishTopic, output); return null; } }实现细节在 WindowFunctionExecutor.processWindow 中可以看到窗口函数的非空返回值也会被运行时自动发布到函数的输出 Topiccontext.publish(context.getOutputTopic(), output, context.getOutputSchemaType())。也就是说窗口函数有两种产生输出的方式——返回非空结果自动发布到输出 Topic或通过context.publish()显式路由到任意 Topic。State storage基于 BookKeeper 的状态存储Pulsar 窗口函数使用 Apache BookKeeper 作为状态存储接口。Apache Pulsar 的安装包括 standalone 单机安装本身就包含 BookKeeper bookie 的部署。Pulsar 通过与 Apache BookKeepertable service的集成来为函数存储state。例如WordCount函数可以通过 Pulsar Functions 状态 API 将其counters状态存储到 BookKeeper table service 中。状态以键值对key-value pairs形式组织key 是字符串value 是任意二进制数据——计数器counter以64 位大端big-endian二进制数值存储。key 的作用域是单个 Pulsar 函数并在该函数的所有实例之间共享。目前Pulsar 窗口函数通过 Java API 暴露状态的访问、更新与管理能力这些 API 在使用 Java SDK 编写函数时均可从上下文对象中获得。核心 API 如下Java API说明incrCounter增加一个以 key 为标识的内置分布式计数器getCounter获取指定 key 的计数器值putState更新指定 key 的状态值incrCounterincrCounterAPI 按 key 增加内置分布式计数器。应用通过它将指定key的计数器增加amount如果该key尚不存在则会创建一个新的 key/** * Increment the builtin distributed counter referred by key * param key The name of the key * param amount The amount to be incremented */ void incrCounter(String key, long amount);getCountergetCounterAPI 获取指定 key 的计数器值通常用于读取之前通过incrCounter修改过的计数器/** * Retrieve the counter value for the key. * * param key name of the key * return the amount of the counter value for this key */ long getCounter(String key);除了getCounterPulsar 还提供了通用的键值 APIputState供函数存储通用的键值状态。putStateputStateAPI 更新指定 key 的状态值/** * Update the state value for the key. * * param key name of the key * param value state value of the key */ void putState(String key, ByteBuffer value);实战示例WordCount 窗口函数下面的WordCountWindowFunction演示了应用如何在窗口函数中存储状态其逻辑非常简单清晰函数先用正则\\.把收到的字符串拆分成多个单词对每个word通过incrCounter(key, amount)将对应计数器加 1。import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; import java.util.Arrays; public class WordCountWindowFunction implements WindowFunctionString, Void { Override public Void process(CollectionRecordString inputs, WindowContext context) throws Exception { for (RecordString input : inputs) { Arrays.asList(input.getValue().split(\\.)).forEach(word - context.incrCounter(word, 1)); } return null; } }由于计数器的 key 在函数的所有实例之间共享即使 WordCount 窗口函数被水平扩展为多个实例并行处理每个单词的计数依然全局一致这正是状态存储对分布式窗口函数的核心价值。窗口上下文底层工作机制为了更透彻地理解WindowContext值得看一下它被创建与使用的时机。在 WindowFunctionExecutor 中initialize(Context context)从用户配置中解析WindowConfig根据配置创建WindowManager、EvictionPolicy驱逐策略与TriggerPolicy触发策略并可选地启动WaterMarkEventGenerator水位线事件生成器用于事件时间窗口每条消息到达时process(I input, Context context)将记录按处理时间或事件时间通过TimestampExtractor提取加入窗口管理器当窗口被激活满足触发条件时onActivation回调把窗口内的记录列表交给processWindow后者调用用户函数process(Window, WindowContext)——此时传入的正是新建的WindowContextImpl(context)窗口过期evict时onExpiry回调对窗口内记录逐条执行ack()保证消息被正确处理。从源码结构可以推断WindowContextImpl是窗口函数用户代码与 Pulsar Functions 运行时之间的桥梁它本身不维护任何状态所有读写都委托给底层Context因此窗口函数与普通函数在元数据、配置、指标、日志和状态方面共享同一套实现学习窗口上下文 API 的经验可以平滑迁移到普通函数开发中。总结WindowContext是 Pulsar 窗口函数 Java SDK 的核心编程入口它围绕窗口场景封装了六大能力Spec输入/输出 Topic、租户命名空间、函数身份与实例信息、输出 Schema、LoggerSLF4J 日志、MetricsrecordMetric用户指标、User configgetUserConfigMap/getUserConfigValue/getUserConfigValueOrDefault三件套、Routingpublish自由路由以及State storage基于 BookKeeper table service 的计数器与通用键值状态。其默认实现 WindowContextImpl 通过委托普通函数Context复用运行时基础设施而 WindowFunctionExecutor 则负责窗口的攒批、触发与过期处理让开发者可以专注于用一组简单的getXxx与recordMetric/incrCounter/publish调用构建有状态的流式窗口计算逻辑。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar Window Functions Context 详解从 Spec 元数据到状态存储的完整 Java SDK 指南Apache Pulsar Window Functions Context 详解从 Spec 元数据到状态存储的完整 Java SDK 指南 本篇技术指南系消息队列后端流处理Apache Pulsar Functions 开发指南Java / Python / Go 多语言 API、SerDe、上下文与状态存储实战Apache Pulsar Functions 开发指南Java / Python / Go 多语言 API、SerDe、上下文与状态存储实战 本篇技术指南围消息队列后端流处理Apache Pulsar Functions 状态存储State Storage开发指南基于 BookKeeper Table Service 的有状态函数实战Apache Pulsar Functions 状态存储State Storage开发指南基于 BookKeeper Table Service 的有状态消息队列后端流处理上一篇探索LazySSH自动化SSH跳板机的未来下一篇从安装到部署PP-OCRv6_tiny_det_onnx完整入门指南3分钟实现文本检测创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考