消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载Apache Pulsar 提供了面向 Apache Spark Streaming 的官方适配器pulsar-spark通过自定义SparkStreamingPulsarReceiver让 Spark Streaming 应用能够直接以 Resilient Distributed Dataset (RDD)完整讲解依赖引入、接收器配置、认证接入与完整可运行示例并对照当前仓库源码揭示ConsumerConfigurationData、认证体系等底层实现细节帮助你快速上手用 Spark Streaming 消费 Pulsar 数据。背景为什么需要 Pulsar 的 Spark 适配器Apache Spark Streaming 是 Spark 生态中处理实时数据流的经典计算模型它以微批次micro-batch的方式将连续的数据流切分为离散的批次并复用 Spark 的 RDD 抽象与丰富算子进行处理。Pulsar 作为分布式发布-订阅消息系统天然需要与 Spark Streaming 这类大数据计算框架打通数据通道。pulsar-spark适配器正是为此而生它提供了一个Spark Streaming 自定义接收器custom receiver使 Spark Streaming 作业可以像接收 Kafka、Socket 数据一样直接从 Pulsar 主题拉取原始数据。应用通过该接收器获得RDD 格式的数据后即可使用 Spark 的变换、聚合、SQL、机器学习等能力进行任意处理例如在官方示例中统计收到的消息里包含 Pulsar 字符串的消息数量。该适配器在 2.4.0 版本文档中与 Pulsar 与 Apache Storm 的适配器提供PulsarSpout与PulsarBolt并列共同构成 Pulsar 对主流流处理框架的对接能力。前置条件引入 pulsar-spark 依赖使用接收器前需要先在 Java 项目中声明pulsar-spark库的依赖。使用 Maven如果使用 Maven 构建在pom.xml的properties块中加入版本属性并在dependencies块中加入依赖!-- in your properties block -- pulsar.versionpulsar:version/pulsar.version !-- in your dependencies block -- dependency groupIdorg.apache.pulsar/groupId artifactIdpulsar-spark/artifactId version${pulsar.version}/version /dependency其中pulsar:version是文档中的版本占位符实际使用时请替换为你所部署的 Pulsar 版本号与 Broker 端版本保持一致。使用 Gradle如果使用 Gradle 构建在build.gradle中加入def pulsarVersion pulsar:version dependencies { compile group: org.apache.pulsar, name: pulsar-spark, version: pulsarVersion }同样需要将pulsar:version替换为真实版本号。说明当前仓库为 Pulsar 2.4.0 时代的版本化文档快照pulsar-spark属于独立于本仓库的适配器模块官方示例代码位于 pulsar-adapters 项目仓库中因此本文聚焦于文档给出的接收器配置方式与仓库内可验证的客户端底层实现。使用方式配置 SparkStreamingPulsarReceiver适配器的核心使用方式是把SparkStreamingPulsarReceiver实例传给JavaStreamingContext的receiverStream方法从而得到一个JavaReceiverInputDStreambyte[]输入流。最小可运行示例String serviceUrl pulsar://localhost:6650/; String topic persistent://public/default/test_src; String subs test_sub; SparkConf sparkConf new SparkConf().setMaster(local[*]).setAppName(Pulsar Spark Example); JavaStreamingContext jsc new JavaStreamingContext(sparkConf, Durations.seconds(60)); ConsumerConfigurationDatabyte[] pulsarConf new ConsumerConfigurationData(); SetString set new HashSet(); set.add(topic); pulsarConf.setTopicNames(set); pulsarConf.setSubscriptionName(subs); SparkStreamingPulsarReceiver pulsarReceiver new SparkStreamingPulsarReceiver( serviceUrl, pulsarConf, new AuthenticationDisabled()); JavaReceiverInputDStreambyte[] lineDStream jsc.receiverStream(pulsarReceiver);这段代码的关键要素serviceUrlPulsar Broker 的接入地址。本地 standalone 模式默认是pulsar://localhost:6650/与 Broker 默认配置 中的brokerServicePort6650一致。topic要消费的 Pulsar 主题的完整名称例如持久化主题persistent://public/default/test_src。subs消费所用的订阅名称Subscription Name。ConsumerConfigurationDatabyte[]Pulsar 客户端的消费者配置对象这里通过setTopicNames指定主题集合、通过setSubscriptionName指定订阅名。AuthenticationDisabled关闭认证的认证实现。文档中的完整示例统计收到的消息中包含 Pulsar 字符串的消息数量位于 pulsar-adapters 仓库的examples/spark模块SparkStreamingPulsarReceiverExample.java。结合源码理解 ConsumerConfigurationDataConsumerConfigurationDataT是 Pulsar 客户端消费配置的真实载体其源码位于 pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java。从源码结构看它实现了Serializable与Cloneable内部通过DataLombok自动生成 getter/setter因此可以被 Spark 的序列化机制与setXxx链式配置使用。该配置类除文档示例用到的topicNames、subscriptionName外还提供了一批可供深入调优的默认值以当前仓库源码为准配置项默认值含义subscriptionTypeExclusive订阅类型Exclusive / Shared / Failover / Key_SharedsubscriptionModeDurable订阅模式Durable / NonDurablereceiverQueueSize1000接收队列大小acknowledgementsGroupTimeMicros100ms以微秒计确认消息的批量聚合时间窗口negativeAckRedeliveryDelayMicros1min以微秒计负确认nack后的重投递延迟maxTotalReceiverQueueSizeAcrossPartitions50000所有分区上的接收队列总大小上限consumerNamenull消费者名称ackTimeoutMillis0消息确认超时时间0 表示关闭tickDurationMillis1000定时器 tick 间隔毫秒priorityLevel0消费者优先级maxPendingChunkedMessage10最大挂起的分块chunked消息数autoAckOldestChunkedMessageOnQueueFullfalse队列满时是否自动确认最旧的分块消息expireTimeOfIncompleteChunkedMessageMillis1min不完整分块消息的过期时间readCompactedfalse是否读取已压缩compacted的主题subscriptionInitialPositionLatest订阅的初始位置Latest / EarliestpatternAutoDiscoveryPeriod60正则订阅模式的主题自动发现周期秒regexSubscriptionModePersistentOnly正则订阅只匹配持久化主题autoUpdatePartitionstrue分区数变化时是否自动更新autoUpdatePartitionsIntervalSeconds60自动更新分区的间隔秒replicateSubscriptionStatefalse跨集群复制订阅状态startPausedfalse是否以暂停状态启动消费者此外该类通过JsonIgnore标记了messageListener、consumerEventListener、messageCrypto、keySharedPolicy等无法序列化的瞬态字段以及properties用户自定义属性、deadLetterPolicy、batchReceivePolicy、retryEnable等高级配置。这意味着在 Spark 接收器场景下可以通过直接构造并设置该配置对象为接收器提供比示例更细粒度的消费行为控制。结合源码理解认证体系示例中SparkStreamingPulsarReceiver构造函数的第三个参数是认证对象new AuthenticationDisabled()表示不启用认证new AuthenticationToken(token:secret-JWT-token)表示使用 JWT Token 认证。在 Pulsar 客户端 API 中认证统一抽象为Authentication接口。虽然 2.4.0 版本中AuthenticationDisabled与AuthenticationToken的默认实现位于客户端 impl 包例如 pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDisabled.java但公共 API 层提供了更推荐的统一入口AuthenticationFactory见 pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationFactory.java包含AuthenticationFactory.token(String token)与AuthenticationFactory.token(SupplierString tokenSupplier)基于固定或动态提供的 JWT Token 创建认证AuthenticationFactory.create(String authPluginClassName, String authParamsString)与AuthenticationFactory.create(String authPluginClassName, MapString, String authParams)按插件类名与参数创建自定义认证实现。因此在需要更灵活认证如 TLS 客户端证书、Athenz、SASL 等的场景可从当前仓库的认证插件中选择例如 Pulsar 与 JWT Token 安全、TLS 认证 等文档所介绍的实现。使用 Token 认证的接收器配置当 Pulsar 集群启用了 JWT 认证时需要为接收器传入 Token 认证对象SparkStreamingPulsarReceiver pulsarReceiver new SparkStreamingPulsarReceiver( serviceUrl, pulsarConf, new AuthenticationToken(token:secret-JWT-token));其中secret-JWT-token是实际签发得到的 JWT 密文。文档明确说明如有需要可以使用其他 Pulsar 认证类即构造函数的第三个参数是可替换的认证抽象便于接入不同的安全体系。实际部署中的注意事项结合文档与仓库配置可以总结出以下实践要点版本匹配pulsar-spark的版本应与 Pulsar Broker/客户端版本保持一致避免二进制协议或 API 不兼容。服务地址与端口本地 standalone 默认pulsar://localhost:6650/生产环境请替换为实际的 Broker 或 Proxy 地址TLS 场景使用pulsarssl://相关配置可参考 TLS 传输加密文档。主题命名使用完整主题名如persistent://tenant/namespace/topicPulsar 支持多租户租户与命名空间划分可参考 多租户概念。订阅语义setSubscriptionName指定的订阅配合默认Durable订阅模式可在应用重启后从上次确认的位置继续消费如需从最早消息开始可配置subscriptionInitialPosition Earliest。认证安全Token 属于敏感信息生产环境中应通过安全的密钥管理或AuthenticationFactory.token(Supplier)动态供应避免硬编码在代码中。小结pulsar-spark适配器通过SparkStreamingPulsarReceiver把 Pulsar 无缝接入 Spark Streaming只需三步——引入依赖、构造ConsumerConfigurationData并指定主题与订阅、把接收器注册进receiverStream即可获得以 RDD 表示的消息流。文档示例同时展示了无认证与 JWT Token 两种接入方式而仓库中 ConsumerConfigurationData.java 与 AuthenticationFactory.java 的实现为接收器的参数扩展与认证定制提供了充分依据。你可以在 version-2.4.0 版本文档目录 中查阅更多 Pulsar 概念与安全配置将本适配器与完整的 Pulsar 能力结合使用。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 与 Apache Spark Streaming 集成指南使用 pulsar-spark 接收器消费消息Apache Pulsar 与 Apache Spark Streaming 集成指南使用 pulsar spark 接收器消费消息 Apache Pulsa消息队列后端流处理Apache Pulsar 与 Spark 集成指南使用 Spark Streaming Receiver 消费 Pulsar 消息Apache Pulsar 与 Spark 集成指南使用 Spark Streaming Receiver 消费 Pulsar 消息 本指南讲解 Apache消息队列后端流处理Apache Pulsar 与 Spark Streaming 集成指南pulsar-spark 接收器接入与实战配置Apache Pulsar 与 Spark Streaming 集成指南pulsar spark 接收器接入与实战配置 本篇文章基于 Apache Pulsa消息队列后端流处理上一篇Edge 卸载不彻底、一更新就复活EdgeRemover 三步根治卸载难题下一篇k-skill 实战指南基于 korean-holiday-calendar 技能与 k-skill-proxy 的韩国法定假日/24 节气一站式查询方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考