消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本文以 Apache Pulsar 2.2.1 版本官方文档sql-deployment-configurations为核心系统讲解 Pulsar SQL由 Trino即原 Presto SQL 引擎驱动的完整部署与配置流程从 Presto Pulsar Connector 的核心参数、多主机连接方式到如何将插件接入既有 Presto 集群、如何通过./bin/pulsar sql-worker启动全新集群、以及如何在多节点上搭建协调者coordinator 工作节点worker架构。读者阅读完本文后将能够独立完成 Pulsar SQL 的配置、启动、多节点部署与验证并理解其绕过 Broker 直连 BookKeeper 读取数据的底层机制。1. Pulsar SQL 架构与配置文件概览Pulsar SQL 允许你用标准 SQL 直接查询 Pulsar 中的 Topic 数据。它基于Trinoformerly Presto SQL构建因此其 worker 的部署与配置方式与 Presto 完全一致。与普通 Presto 集群最大的不同在于Pulsar SQL 通过Presto Pulsar Connector将每个 Topic 映射为一张数据表并通过 BookKeeper 客户端直接读取 Ledger 中的数据。Pulsar 发行版的 SQL 相关配置集中放置在conf/presto/目录下仓库中的真实结构如下配置文件作用conf/presto/catalog/pulsar.propertiesPresto Pulsar Connector 的全部连接与性能参数conf/presto/config.propertiesPresto 集群级配置coordinator/worker 角色、端口、内存等conf/presto/jvm.configworker 的 JVM 参数堆内存、GC 策略等conf/presto/log.properties日志级别配置2. 配置 Presto Pulsar ConnectorConnector 的配置集中在${project.root}/conf/presto/catalog/pulsar.properties属性文件中仓库中对应 conf/presto/catalog/pulsar.properties。文档给出的核心配置与默认值如下# name of the connector to be displayed in the catalog connector.namepulsar # the url of Pulsar broker service pulsar.web-service-urlhttp://localhost:8080 # URI of Zookeeper cluster pulsar.zookeeper-urilocalhost:2181 # minimum number of entries to read at a single time pulsar.entry-read-batch-size100 # default number of splits to use per query pulsar.target-num-splits4注意文档中的pulsar.entry-read-batch-size在仓库实际配置文件中对应的键名为pulsar.max-entry-read-batch-size两者含义一致均为单次读取的条目数量最小值。实际使用时以仓库发行版conf/presto/catalog/pulsar.properties中的键名为准。2.1 核心参数详解与源码印证对照 PulsarConnectorConfig.java 的Config注解实现各参数的底层行为如下connector.namepulsarPresto catalog 中显示的 connector 名称对应PulsarPlugin注册的 connector 标识。pulsar.web-service-urlPulsar broker 的 Web 服务地址HTTP 端口默认 8080。在 PulsarConnectorConfig.java 中webServiceUrl默认留空此时回退到brokerServiceUrl默认http://localhost:8080一旦配置了pulsar.web-service-urlgetBrokerServiceUrl()会优先返回 Web 服务地址。该地址最终被用于创建PulsarAdmin见getPulsarAdmin()方法源码第 427-452 行用于获取 Topic 的 schema 信息。pulsar.zookeeper-uriZooKeeper 集群地址用于 SQL worker 直连 ZooKeeper 获取 Ledger 元数据metadata service URI 形如zk://uri/ledgers。该参数在 PulsarConnectorCache.java 中被写入 BookKeeperClientConfiguration从而绕过 broker 直接读取数据。pulsar.entry-read-batch-size/pulsar.max-entry-read-batch-size默认 100单次读取的 entry 批量大小。该值在 PulsarRecordCursor.java 中被读取并作为读取游标的最大批大小直接决定读吞吐与内存占用之间的平衡。pulsar.target-num-splits文档默认 4仓库默认 2每个查询默认生成的 split 数量。在 PulsarSplitManager.java 中getSplits()首先读取该值作为 split 数量再结合 topic 的 entry 总数与每批读取大小计算实际的 split 切分从而决定查询的并行度。2.2 配置文件中的进阶参数仓库实测值仓库中的真实配置文件 conf/presto/catalog/pulsar.properties 还提供了以下进阶参数供生产调优参考# deprecated早期版本使用推荐改用 pulsar.web-service-url # pulsar.broker-service-urlhttp://localhost:8080 # max message queue size pulsar.max-split-message-queue-size10000 # max entry queue size pulsar.max-split-entry-queue-size1000 # 队列缓存字节上限-1 表示不限制前半用于 entry queue后半用于 message queue pulsar.max-split-queue-cache-size-1 # 是否重写 namespace 分隔符避免使用 namespace 允许的符号导致误重写 pulsar.namespace-delimiter-rewrite-enablefalse pulsar.rewrite-namespace-delimiter/ # 单个 batch message 最大字节数默认 5MB # pulsar.max-message-size5242880 ####### TIERED STORAGE OFFLOADER CONFIGS ####### # pulsar.managed-ledger-offload-driver aws-s3 # pulsar.offloaders-directory /pulsar/offloaders # pulsar.managed-ledger-offload-max-threads 2 # pulsar.offloader-properties {s3ManagedLedgerOffloadBucket: offload-bucket, ...} ####### AUTHENTICATION CONFIGS ####### # pulsar.auth-plugin # pulsar.auth-params # pulsar.tls-allow-insecure-connection # pulsar.tls-hostname-verification-enable # pulsar.tls-trust-cert-file-path ####### BOOKKEEPER CONFIGS ####### pulsar.bookkeeper-throttle-value 0 # pulsar.bookkeeper-num-io-threads # pulsar.bookkeeper-num-worker-threads pulsar.bookkeeper-use-v2-protocoltrue pulsar.bookkeeper-explicit-interval0 ####### MANAGED LEDGER CONFIGS ####### pulsar.managed-ledger-cache-size-MB 0 # pulsar.managed-ledger-num-worker-threads # pulsar.managed-ledger-num-scheduler-threads ####### PROMETHEUS CONFIGS ####### # pulsar.stats-providerorg.apache.bookkeeper.stats.prometheus.PrometheusMetricsProvider # pulsar.stats-provider-configs{httpServerEnabled:false, prometheusStatsHttpPort:9092, prometheusStatsHttpEnable:true}这些参数在源码中的对应关系均可在 PulsarConnectorConfig.java 中找到Config注解包括BookKeeper 客户端参数pulsar.bookkeeper-throttle-value每秒读取条数限流0 表示不限、pulsar.bookkeeper-use-v2-protocol默认 true使用 v2 协议携带 LAC、pulsar.bookkeeper-explicit-interval显式 LAC 间隔等均被 PulsarConnectorCache.java 灌入 BookKeeperClientConfiguration。Managed Ledger 参数pulsar.managed-ledger-cache-size-MB数据负载缓存从 JVM 直接内存分配0 表示禁用、pulsar.managed-ledger-num-worker-threads与pulsar.managed-ledger-num-scheduler-threads默认均为 CPU 核数。认证与 TLSpulsar.auth-plugin/pulsar.auth-params会被传入PulsarAdmin.builder().authentication(...)TLS 三个参数分别对应allowTlsInsecureConnection、enableTlsHostnameVerification、tlsTrustCertsFilePath见 PulsarConnectorConfig.java 的getPulsarAdmin()。命名空间分隔符重写pulsar.namespace-delimiter-rewrite-enable与pulsar.rewrite-namespace-delimiter用于处理命名空间分隔符兼容问题重写字符不得包含 namespace 名称允许的符号a-zA-Z_0-9 -:%否则 setRewriteNamespaceDelimiter 会抛出IllegalArgumentException。2.3 连接多主机 Pulsar 集群当 Pulsar 集群包含多个 broker 或 ZooKeeper 节点时可以在同一配置项中列出多个地址逗号分隔。文档示例pulsar.web-service-urlhttp://localhost:8080,localhost:8081,localhost:8082 pulsar.zookeeper-urilocalhost1,localhost2:2181底层处理同样有源码佐证在 PulsarConnectorCache.java 中ZooKeeper 地址列表中的逗号会被替换为分号构造出形如zk://localhost1;localhost2:2181/ledgers的 BookKeeper 元数据服务 URI实现多 ZooKeeper 节点的容错连接。3. 将 Connector 接入既有 Presto 集群如果你已经拥有一套 Presto 集群无需重新部署只需将 Presto Pulsar Connector 插件复制到既有集群即可$ wget pulsar:binary_release_url下载归档插件包后将其解压到 Presto 的插件目录plugin 目录通常为$PRESTO_HOME/plugin并把上文 2. 配置 Presto Pulsar Connector 中的pulsar.properties放入既有集群的 catalog 目录$PRESTO_HOME/etc/catalog/重启 Presto 节点后即可通过 SQL 查询 Pulsar 数据。该方式适合在已有查询基础设施上增量引入 Pulsar 数据源无需迁移既有的 Presto 集群。4. 部署一个全新的 Pulsar SQL 集群由于 Pulsar SQL 由 Trino 驱动其 worker 的部署配置方式与 Presto 完全一致可以直接复用 Presto launcher 的命令行参数。4.1 sql-worker 命令与参数sql-worker命令在 bin/pulsar 中实现./bin/pulsar sql-worker实际会调用${PRESTO_HOME}/bin/launcher并默认传入--etc-dir ${PULSAR_PRESTO_CONF}即conf/presto。因此它的所有命令行参数与 Presto launcher 完全相同可用--help查看$ ./bin/pulsar sql-worker --help Usage: launcher [options] command Commands: run, start, stop, restart, kill, status Options: -h, --help show this help message and exit -v, --verbose Run verbosely --etc-dirDIR Defaults to INSTALL_PATH/etc --launcher-configFILE Defaults to INSTALL_PATH/bin/launcher.properties --node-configFILE Defaults to ETC_DIR/node.properties --jvm-configFILE Defaults to ETC_DIR/jvm.config --configFILE Defaults to ETC_DIR/config.properties --log-levels-fileFILE Defaults to ETC_DIR/log.properties --data-dirDIR Defaults to INSTALL_PATH --pid-fileFILE Defaults to DATA_DIR/var/run/launcher.pid --launcher-log-fileFILE Defaults to DATA_DIR/var/log/launcher.log (only in daemon mode) --server-log-fileFILE Defaults to DATA_DIR/var/log/server.log (only in daemon mode) -D NAMEVALUE Set a Java system property参数要点command支持run前台运行、start/stop/restart/kill守护进程管理、status状态查看--etc-dir指定配置目录默认conf/presto--data-dir指定数据目录--jvm-config、--config、--log-levels-file分别覆盖 JVM、集群与日志配置文件-D NAMEVALUE用于设置 Java 系统属性。4.2 使用自定义配置目录运行集群默认配置位于${project.root}/conf/presto仓库路径 conf/presto你可以通过修改默认配置实现定制也可以让 worker 从不同的配置目录读取配置、向不同的目录写数据$ ./bin/pulsar sql-worker run --etc-dir /tmp/incubator-pulsar/conf/presto --data-dir /tmp/presto-14.3 以守护进程方式启动$ ./bin/pulsar sql-worker start启动后可用status查看状态用stop/restart/kill管理进程。4.4 单机独立环境快速验证如需在单机快速体验官方配套文档 Query data with Pulsar SQL 给出了完整流程先./bin/pulsar standalone启动独立集群再./bin/pulsar sql-worker run启动 SQL worker最后./bin/pulsar sql进入 SQL CLI该命令在 bin/pulsar 中默认连接localhost:8081即可执行show catalogs;、show schemas in pulsar;等语句验证。5. 在多节点上部署集群Pulsar SQL 集群或标准 Presto 集群可部署在多个节点上其中一个节点充当coordinator协调者负责解析、规划与调度查询其余节点作为worker工作节点执行分片任务并读写数据。以下以三节点集群为例展示完整部署流程。5.1 第 1 步将二进制发行版复制到三个节点将 Pulsar 二进制发行版复制到三个节点节点 1、节点 2、节点 3。5.2 第 2 步配置 coordinator 节点第一个节点作为 Presto coordinator其最小配置${project.root}/conf/presto/config.properties对应仓库 conf/presto/config.properties如下coordinatortrue node-scheduler.include-coordinatortrue http-server.http.port8080 query.max-memory50GB query.max-memory-per-node1GB discovery-server.enabledtrue discovery.uricoordinator-url参数含义coordinatortrue本节点承担协调者角色node-scheduler.include-coordinatortrue允许调度器把任务也分配给 coordinator 节点自身小规模集群可让协调者同时干活http-server.http.port8080HTTP 服务端口query.max-memory50GB单个查询在集群中可用的总内存上限query.max-memory-per-node1GB单个查询在单节点上可用的内存上限discovery-server.enabledtrue开启 discovery 服务让其他节点通过discovery.uri加入集群discovery.uricoordinator-urldiscovery 服务地址替换为 coordinator 节点的实际 URL。5.3 第 3 步配置 worker 节点其余两个节点作为 worker 节点配置如下coordinatorfalse http-server.http.port8080 query.max-memory50GB query.max-memory-per-node1GB discovery.uricoordinator-urlworker 节点不开 discovery 服务只通过discovery.uri指向 coordinator 加入集群。5.4 第 4 步修改 Connector 配置在第 2、3 步的基础上为三个节点分别修改conf/presto/catalog/pulsar.properties中的pulsar.web-service-url与pulsar.zookeeper-uri使其指向真实的 Pulsar broker 与 ZooKeeper 集群多主机写法见 2.3 连接多主机 Pulsar 集群。5.5 第 5 步启动 coordinator 节点$ ./bin/pulsar sql-worker run5.6 第 6 步启动 worker 节点在其余两个节点上执行相同命令$ ./bin/pulsar sql-worker run5.7 第 7 步启动 SQL CLI 并验证集群状态在任意节点上启动 SQL CLI--server参数指向 coordinator 地址$ ./bin/pulsar sql --server coordinate_url在 CLI 中执行以下查询查看节点状态presto SELECT * FROM system.runtime.nodes; node_id | http_uri | node_version | coordinator | state --------------------------------------------------------------------- 1 | http://192.168.2.1:8081 | testversion | true | active 3 | http://192.168.2.2:8081 | testversion | false | active 2 | http://192.168.2.3:8081 | testversion | false | active输出中的coordinator列标识各节点角色state为active表示节点健康并已加入集群。更多 Presto 部署细节可参考 Trino 官方部署文档。6. 重要注意事项LAC 与数据可见性:::notebroker 不会主动推进 LACLast Add Confirmed最后确认添加的条目因此当 Pulsar SQL 绕过 broker 直连 BookKeeper 查询数据时它只能读取到所有 bookie 已学习到的 LAC 之前的 entry。可以通过在broker.conf中设置bookkeeperExplicitLacIntervalInMills让 broker 周期性写 LAC。:::这一注意事项的底层逻辑可从 PulsarConnectorCache.java 中得到印证SQL worker 会创建直连 ZooKeeper/BookKeeper 的 BookKeeperClientConfiguration默认setUseV2WireProtocol(true)即使用 v2 协议通过 piggyback 方式携带 LAC读取数据不再经过 broker。由于 producer 的最新写入只有在 bookie 确认LAC 推进后才对直读客户端可见若 broker 不显式推进 LAC最新的少量数据可能暂时查询不到。在broker.conf中开启bookkeeperExplicitLacIntervalInMills毫秒为单位即可让 broker 周期性地显式写入 LAC缩短数据可见延迟。配置文件中对应的pulsar.bookkeeper-explicit-interval参数默认 0则是 SQL worker 侧 BookKeeper 客户端的显式 LAC 间隔设置。7. 部署自检清单完成部署后建议按以下顺序自检Connector 配置确认conf/presto/catalog/pulsar.properties中的pulsar.web-service-url、pulsar.zookeeper-uri与集群实际地址一致多主机地址用逗号分隔集群角色确认config.properties中 coordinator 节点coordinatortrue且开启discovery-server.enabledworker 节点coordinatorfalse且discovery.uri指向 coordinator端口与内存确认http-server.http.port未被占用query.max-memory/query.max-memory-per-node与节点内存匹配进程状态使用./bin/pulsar sql-worker status确认各节点进程存活数据可见性若查询不到最新数据检查 broker 是否已配置bookkeeperExplicitLacIntervalInMills。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar SQLPresto Pulsar Connector部署与配置完整指南Apache Pulsar SQLPresto Pulsar Connector部署与配置完整指南 Apache Pulsar SQL 是构建在 Prest消息队列后端流处理Apache Pulsar SQL 配置与集群部署完整指南Presto Pulsar Connector 参数详解与多节点实战Apache Pulsar SQL 配置与集群部署完整指南Presto Pulsar Connector 参数详解与多节点实战 Apache Pulsar S消息队列后端流处理Apache Pulsar SQL 部署与配置实战Presto Pulsar Connector 参数详解与多节点集群搭建Apache Pulsar SQL 部署与配置实战Presto Pulsar Connector 参数详解与多节点集群搭建 Pulsar SQL 基于 Tri消息队列后端流处理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考