缓存KV存储消息队列流处理后端【免费下载链接】hazelcastHazelcast is a unified real-time data platform combining stream processing with a fast data store, allowing customers to act instantly on>项目地址https://gitcode.com/gh_mirrors/ha/hazelcast点击查看免费下载本文基于 docs/design/jet/025-jet-job-partition-pruning.md 展开。它系统讲解了 Hazelcast Platform 5.4 引入的 Jet 任务分区剪枝Partition Pruning机制如何让一个 Job 只部署到真正持有所需数据分区的成员上成员剪枝如何在 IMap 扫描时只读取必要分区扫描分区剪枝以及为什么放弃了处理器剪枝Processor Pruning。读完本文你将理解三类剪枝的边界与取舍、ExecutionPlanBuilder中分区分配的核心算法以及 SQL 侧SpecificPartitionsImapReaderPms等元供应商的实现原理并能在自己的 SQL/Jet 作业中预判剪枝生效的条件与场景。背景剪枝要解决的性能问题在 Hazelcast Platform 5.4 之前Jet 任务总是被部署到集群中的全部数据成员所有非 lite member上。即使某个 DAG 顶点并不会用到所有成员Jet 依然会在未被实际使用的成员上创建no-op 处理器为这些顶点创建进/出队列并启动处理器尽管它们立即完成队列随即以DONE_ITEM关闭对于完全未被使用的成员协调者coordinator依然要向它发送InitExecutionOperation并等待其完成。即便处理器是 no-op 或根本没有数据可处理这也是一笔不必要的开销在很小的批处理作业batch job或很大的集群上表现得尤为明显。分区剪枝的目标正是消除这类浪费。术语澄清原文档对三类剪枝给出了精确定义三者是相互独立、可叠加的优化手段术语含义Member pruning成员剪枝阻止不持有请求数据的集群成员参与 Job 执行Processor pruning处理器剪枝消除冗余的阶段处理器stage processor创建Scan partition pruning扫描分区剪枝提取分区条件只选择扫描处理器元供应商SpecificPartitionsImapReaderPms它负责产生ReadMapOrCacheP所需读取的分区目标与非目标该项目的目标与上表术语一一对应只把 Job 部署到包含所需分区的成员上限制 IMap 扫描所涉及的分区数量。明确划出的非目标避免范围失控不支持对流式streaming作业做任何剪枝——未来可能考虑但当前不在范围内不对成员剪枝和处理器剪枝做迁移容错migration-tolerance——如果 Job 启动期间发生分区迁移Job 会以次优方式运行可能从其他成员拉取数据这与当前行为一致不支持分区索引扫描partitioned index scan——纯 Jet 中不支持索引扫描只有 SQL 有专门的处理器。总体设计三个阶段互补设计上把成员剪枝、处理器剪枝和扫描分区剪枝拆分成相互独立的阶段。它们可以彼此补强但每一种剪枝单独使用本身就是一个有效的优化。成员剪枝Member pruning成员剪枝本身可以有相当简单的实现方案在ExecutionPlanBuilder执行计划创建阶段中完成即可。此前ProcessorMetaSupplier#get的契约给执行计划创建算法带来了严格假设但该契约已经改变为剪枝提供了空间。成员剪枝的典型用例注默认情况下示意图的基本单元是 DAG 顶点但有时会以处理器为基本单元文中会明确说明。用例 1简单 Map 扫描——对扫描类 Job 剪掉除 Member 1 以外的所有成员。之所以强调这个简单场景是因为提交的查询中很大一部分形如SELECT * FROM map WHERE ...。如果能把 partitionKey 提示给优化器提交的 Job 甚至可能完全本地化- SCAN[map2, partitionKey1]用例 2大集群中的 Scan - Hash Join——剪掉没有任何连接的成员集群中只保留 Member 1 和 Member K- HASH JOIN - SCAN[map1, partitionKey1] - SCAN[map2, partitionKeyK]用例 3Scan - Transform - Aggregate——该图中的基本单元是处理器因为实际图形态为Scan - Transform - Aggregate。这里可以整体消除 Member 2根据分区键条件只有 Member 1 包含所需数据而汇聚点是最终目的地。这也是后续处理器剪枝优化的一个好例子- AGGREGATION - TRANSFORM - SCAN[map, partitionKeyK]三种候选方案设计阶段曾评估过三种方案它们的差异在于在哪个阶段意识到某个成员可以被从 Job 执行中剪掉Generic 方法通用方案核心思想是让ExecutionPlanBuilder对收到的 DAG 做详细分析并在创建执行计划前尝试优化它。剪枝成员的唯一要求是 source 定义了分区属性策略partitioning attribute strategy。它能为 SQL 和 Pipeline API 的所有 connector 统一实现成员剪枝但代价是问题分析复杂度高、开发周期长、边角情况corner cases多。SQL-oriented 方法SQL 导向方案与通用方案不同成员剪枝主要受益于小型作业而这类作业绝大多数由 SQL 生成团队的目标是替代 Predicate API其中PartitionPredicate可用于分区数据查询。具体做法是利用 SQL 优化阶段为所有关系确定分区然后把所需分区集合提取到JobConfig中由__sql.requiredPartitions参数承载从而指示ExecutionPlanBuilder只挑选所需成员。Mixed 方法混合方案在 SQL 导向方案基础上在ExecutionPlanBuilder计算所需成员时附加 DAG 分析以快速确定运行 Job 所需的成员。该方案关注两个场景支持allToOne边所有边都指向一个成员并且按单键分区。这些分区和成员若为多个也应被纳入 DAG 结果。主要动机是大多数查询很可能通过allToOne把扫描数据转发给协调者需要在恰当位置把协调者和相应分区纳入 Job。支持一般的partitioned边解锁聚合类操作。DAG 分析结果会包含一个布尔标志决定是否需要保留集群中的所有分区。基于以上分析DAG 分析结果可能影响分区分配如果 DAG 存在distributed-partitioned边未被使用的分区将被分配给所有运行 Job 所需的成员。读者可能担心这会降低性能但混合成员剪枝与扫描处理器分区剪枝是耦合的后者恰好弥补了partitioned边场景下这个潜在的性能缺口。总体算法5 步结合 ExecutionPlanBuilder.java 中getPartitionAssignment的实现与注释算法可归纳为若未收到所需分区required partitions就把全部分区分配给集群中所有成员走默认的普通分配路径若收到了所需分区则执行 DAG 分析找出所有持有 Job 所需数据分区dataPartitions的成员添加具有显式路由Edge.distributeTo的成员若尚未添加。此步骤之后得到的成员即为执行 Job 所需的全部成员required members把不存储数据但出于主要是路由原因需要的额外分区分配给这些 required members。源码印证在ExecutionPlanBuilder.getPartitionAssignment中ExecutionPlanBuilder.java当dataPartitions null时直接遍历全部分区0..partitionCount-1按分区所有者partitionService.getPartitionOwnerOrWait分配到成员所有者不在成员列表时以 round-robin 方式兜底分配当dataPartitions ! null时只遍历dataPartitions集合随后把extraRequiredMemberAddresses来自Edge.distributeTo强制加入成员列表最后在allPartitionsRequired || !routingPartitions.isEmpty()时把剩余分区allPartitionsRequired时取全部分区减去dataPartitions否则取routingPartitions减去dataPartitions以 round-robin 方式补到 required members 上——这些分区仅用于内部路由。此外ExecutionPlanBuilder.java 中jobConfig.getArgument(KEY_REQUIRED_PARTITIONS)正是读取 SQL 侧写入JobConfig的__sql.requiredPartitions验证了 SQL-oriented/Mixed 方案与执行计划的衔接点与之配套的 PartitionArrangement.java 则承载剪枝场景下的分区安排信息。扫描分区剪枝Scan partition pruning扫描分区剪枝的关键原则是生成扫描处理器的专用处理器元供应商PMS应当能够自足地计算出要扫描的精确分区集合。为此SQL 侧设计了一套便捷 API以便把最少必要的信息传给 PMS 构造函数。IMap 专用可剪枝扫描 PMS 的设计与实现SpecificPartitionsImapReaderPms是执行 IMap 扫描的新元供应商位于 SpecificPartitionsImapReaderPms.java。它的设计要点自足计算能够自行计算要扫描的精确分区集合优雅降级如果给定 IMap 不具备分区策略partitioning strategy可以完全不做分区计算回退到普通全量扫描。其工作流程对应源码实现init阶段计算可剪枝分区集SpecificPartitionsImapReaderPms.java当requiredPartitionsExprs非空时遍历每个分区键表达式通过ExpressionEvalContext求值得到分区键分量partitionKeyComponents再调用PartitioningStrategyUtil.getPartitionIdFromKeyComponents计算出分区 ID若分区键无效如为 null则回退到非剪枝逻辑。由于requiredPartitionsExprs可能产出重复分区最终结果会做sorted().distinct()去重排序保证partitionsToScan唯一且有序并保存partitionAssignment供后续按成员筛选。get阶段按成员产出精确分区SpecificPartitionsImapReaderPms.java若partitionsToScan null返回普通LocalProcessorSupplier全量扫描否则对每个地址用二分查找partitionAssignment已排序过滤出该成员本地要扫描的分区构造携带memberPartitionsToScan的LocalProcessorSupplier进而在ProcessorSupplier生成处理器时只让ReadMapOrCacheP扫描这些必需分区。可复用性/协作性控制isReusable()在requiredPartitionsExprs null时返回 trueinitIsCooperative()依据表达式是否全部协作cooperative返回对应结果避免阻塞 Jet 线程。测试印证E2E 测试 SqlPartitionPruningE2ETest.java 覆盖了简单分区键when_scanWithSimplePruningKey_then_prunable、Portable/Compact 类型分区键、ANALYZE扫描、复合分区键when_scanWithCompoundPruningKey_then_prunable等场景单元测试 SpecificPartitionsImapReaderPmsTest.java 则直接验证 PMS 的分区计算逻辑。适用场景扫描分区剪枝在成员剪枝不可用时依然有效。例如两个表做 UNION其中只有一个扫描可剪枝此时无法做成员剪枝因为另一个扫描需要全部成员但可以对这个可剪枝的扫描局部应用扫描分区剪枝减少其读取的分区数。分区感知的lazyForceTotalParallelismOne为了与现有的forceTotalParallelismOne互补设计引入了一个新的分区感知PMS 构建器lazyForceTotalParallelismOne它利用提供的分区键表达式动态计算成员把总并行度total parallelism降到 1它不缓存计算出的成员地址从而防止错误复用被缓存的执行计划在分区迁移后使用陈旧计划。源码印证SQL 侧配套的 LazyDefiningSpecificMemberPms.java 实现了“只在持有给定分区键的节点上运行 ProcessorSupplier”的元供应商init阶段通过ExpressionEvalContext求值分区键既支持参数索引partitionArgIndex也支持表达式供应商partitionKeyExprSupplier调用getOwnerAddress(context, partitionKey)计算所属成员get阶段对非所属地址返回ExpectNothingProcessorSupplier注释明确指出“ExpectNothingProcessorSupplier可能被分区剪枝消除当它被 SQL 使用时”并要求本地并行度必须为 1否则抛IllegalArgumentExceptionpreferredLocalParallelism()返回 1。被拒绝的机会处理器剪枝Processor pruning处理器剪枝被分成了两类intra-member成员内与inter-member成员间。处理器剪枝的用例用例Scan - Transform - Aggregate单级聚合在成员剪枝示例的优化结果上可以更进一步——利用分区参与知识调优本地并行度参数直接不创建不参与数据处理的处理器这是intra-member处理器剪枝的好目标。双级聚合- COMBINE - ACCUMULATE - TRANSFORM - SCAN[map, partitionKeyK]两阶段聚合可能是inter-member处理器剪枝的目标。但处理器逻辑对 Jet 是不透明的opaque改动它需要大量变更可能危及执行计划的正确性。潜在替代做法是在 SQL 优化阶段把两阶段聚合翻译为单阶段再通过成员剪枝消除成员即 scan 和 flatmap 节点。决策不支持最终决定不支持这种处理器剪枝理由有三相对低效大多数带广播边broadcast edge的 DAG 本就是以算法正确性为前提构建的对大多数用例难以实现不符合整体目标。另外值得一提DagNodeUtil实际上实现了 inter-member 处理器剪枝但它并不依赖分区剪枝的元信息meta-information。最终范围与落地最初计划只实现 SQL 导向方案但经过几轮迭代后最终范围确定为成员剪枝采用 Mixed 方法SQL 优化阶段提取所需分区写入JobConfig的__sql.requiredPartitionsExecutionPlanBuilder内的 DAG 分析扫描处理器分区剪枝SpecificPartitionsImapReaderPms自足计算分区处理器剪枝被拒绝因为非通用、实现复杂。从当前仓库源码看这一范围已完整落地ExecutionPlanBuilder.getPartitionAssignment承载了 5 步成员/分区分配算法SpecificPartitionsImapReaderPms与LazyDefiningSpecificMemberPms分别实现了扫描分区剪枝与分区感知的 total parallelism 1 约束SqlPartitionPruningE2ETest则给出了可运行的行为验证。对使用者而言需要记住的关键约束是剪枝目前仅适用于批处理batch作业、依赖 IMap 的分区策略与分区键条件、不承诺分区迁移期间的最优运行且索引扫描不在支持范围内。赞分享缓存KV存储消息队列流处理后端【免费下载链接】hazelcastHazelcast is a unified real-time data platform combining stream processing with a fast data store, allowing customers to act instantly on>项目地址https://gitcode.com/gh_mirrors/ha/hazelcast点击查看免费下载相关推荐Daft 扫描层深度解析ScanOperator、ScanTask 与分区剪枝下推实现Daft 扫描层深度解析ScanOperator、ScanTask 与分区剪枝下推实现 本文基于 Daft 数据引擎Rust 核心中的 daft scan大数据数据分析数据工程AI 应用Unity Movement AI 2D与3D运动系统对比选择最适合你的游戏类型Unity Movement AI 2D与3D运动系统对比选择最适合你的游戏类型 Unity Movement AI是一款强大的Unity运动AI库提供了完Videogrep 终极指南高效处理视频字幕文件的完整解决方案Videogrep 终极指南高效处理视频字幕文件的完整解决方案 你是否曾需要从长视频中快速提取特定对话片段或者想要根据关键词自动创建精彩集锦面对海量视频内音视频上一篇OpenVR SDK终极指南10个技巧让你快速掌握跨平台VR开发下一篇ComfyUI增强插件完全指南解锁AI工作流新体验的10个实用技巧创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考