FRI 与 KG-TOWER 二次开发教程（19）：实战三——企业级批量水力学核算平台（规模与性能）

FRI 与 KG-TOWER 二次开发教程19实战三——企业级批量水力学核算平台规模与性能版本与事实声明环境Python 3.8numpy 2.2.6并行用标准库concurrent.futures.ThreadPoolExecutor。核算内核复用fri_kt。不涉及 FRI/KG-TOWER 的任何 API两者均无公开 API。平台若需软件侧权威结果走第 08 篇的官方通道PRO/II 工具 报表导出与第 14 篇的 Aspen Column Analysis不写任何臆造接口。KG-TOWER 许可禁止逆向与派生、禁止未经书面许可向 Koch-Glitsch 竞争对手分发输出第 16 篇的检查器守护。文中全部数值为示例性建模不代表任何标准规定不对应任何真实装置数据。一句话结论批量平台的性能瓶颈几乎从不在公式——示例 480 个任务在纯算术下跑到9,135 case/s单线程加 8 线程反而降到 8,819 case/sGIL 任务粒度太小调度开销吃掉收益而缓存全命中时才升到28,580 case/s因此规模化的正确发力点是分片、缓存与 I/O 治理而不是把公式写得更快。〇、本篇要解决的认知问题Q1企业级批量核算平台应该分成哪几层为什么任务层必须独立Q2任务分片为什么用一致性哈希它解决了什么、没解决什么Q3为什么本例8 线程比单线程还慢什么条件下并行才有收益Q4缓存能带来多少收益缓存键应该是什么Q5怎么用瓶颈分布回答业务问题而不只是算完了一、机制解析1.1 五层平台架构为什么这对你重要单点脚本与平台脚本的差别不在算法而在关注点分离。五层如下① 来源层Sources 流程模拟契约文件 / 软件导出报表 / 用户录入 ② 数据层Data 工况矩阵、内件库、阈值配置版本化 ③ 引擎层Engine fri_kt 的核算 裕度 不确定度纯函数 ④ 任务层Tasks 任务生成、分片、缓存、重试、断点 ⑤ 交付层Delivery 长表落盘、宽表报表、图形、合规元数据关键设计引擎层必须是纯函数输入→输出无副作用任务层才敢对它做缓存、并行与重试。示例的rate_one(task)就是缓存 引擎的最小组合。1.2 一致性哈希分片为什么不是简单取模分片要让同一个案例永远落在同一片以便分布式执行多台机器各跑一片各自落盘无冲突缓存局部性同一案例重算时命中同片缓存可追责某片出错只需重跑该片。片号int(sha1(case_id)[:8],16) mod Nshards\text{片号}\text{int}\big(\text{sha1}(\text{case\_id})[:8],16\big)\bmod N_{shards}片号int(sha1(case_id)[:8],16)modNshards​它解决了什么案例到分片的映射稳定与任务列表顺序无关。它没解决什么负载均衡——哈希是平均意义均衡实际各片会有偏差示例 8 片时片大小 47~73最大/最小比约 1.55。若要求严格均衡需加权分片或先按代价排序再轮询。1.3 为什么8 线程比单线程慢示例实测单线程9,135 case/s8 线程8,819 case/s——并行反而更慢。原因是双重的GIL全局解释器锁纯 Python 计算无法真正并行线程只在 I/O 阻塞时才让出解释器任务粒度太小480 个任务总耗时约 0.05 s调度本身的开销与计算同量级。什么条件下并行才有收益任务特征并行收益正确并发方式纯 CPU、纯 Python、单任务 1 ms差GIL多进程ProcessPoolExecutor或换 numpy 向量化外部软件调用进程级、秒级好多进程/多实例受许可并发约束I/O 密集读写大量文件好线程池即可经验法则先向量化再考虑并行。示例的整批扫描若改写成 numpy 数组运算吞吐可提升一到两个数量级且不引入任何并发复杂度。1.4 缓存规模化的第一杠杆示例中缓存全命中把吞吐从 8,819 提升到28,580 case/s约 3.2 倍。缓存的键必须与结果一一对应——也就是第 04 篇的case_id内容寻址。缓存值可以是结果摘要也可以是完整结果视存储预算。反直觉点缓存的正确性依赖case_id的完备性——如果键漏掉了一个影响结果的输入如hL或C_sb缓存就会返回错误但看起来有效的结果。这类 bug 比算错更难查因为它在第一次算时是对的只在参数微调后重算时错。所以缓存键必须由结果函数用到的全部输入生成。1.5 瓶颈分布从算完了到告诉业务该改什么示例最后输出合格 292/480瓶颈分布 flood 240 / weep 186 / dP 54。这条信息比合格率有价值得多flood主导240→ 说明容量是主要限制方向是加大塔径或换高容量内件weep其次186→ 说明操作下限也大面积吃紧方向是减小开孔率/孔径或改阀塔板dP最少54→ 说明压降不是这批工况的主矛盾。注意 24018654 480恰好等于案例数——因为每个案例只记一个binding最紧的那一项虽然一行可能有多项越限但binding只取最紧者。业务解读容量与下限同时紧张意味着这不是单点问题而是塔径/内件族的系统性偏小或负荷区间设定过宽。1.6 规模化的另一面失败注入与幂等重试吞吐只解决跑得快规模化还必须解决跑得完。三类真实失败失败类型表现处置单点异常某工况因参数越界抛异常如 ε 过小捕获后写入warnings并标okNA不让整批中断资源中断进程被杀、机器重启靠内容寻址键 片级幂等续跑第 11 篇与本文 2-2外部依赖失败软件侧调用超时、许可不可用超时 重试 残留进程清理第 08/16 篇最佳实践主动做失败注入——在测试环境里人为让 1% 的工况抛异常、让某个分片写到一半被中断然后验证续跑后结果与一次性跑完完全一致。没有做过失败注入的批量平台其幂等性只是纸面承诺。注意许可约束若批处理里含对 KG-TOWER 的调用并发度会受软件许可约束具体上限以官方渠道为准本系列不写数字此时提高并发可能无效正确做法是减少调用次数缓存、合并工况而不是加大并发。二、完整代码与逐行剖析代码 2-1分片 缓存 吞吐测量可直接运行# -*- coding: utf-8 -*- platform_runner.py —— 规模化分片调度 吞吐测量 缓存命中 importos,time,json,math,hashlibfromdataclassesimportreplacefromconcurrent.futuresimportThreadPoolExecutorfromfri_kt.coreimportCase,Internals,case_id,rate_tray,rate_packing,margins CACHE{}# case_id - 结果摘要真实平台应落 Redis/ParquetHITS{hit:0,miss:0}defrate_one(task):bag_id,it,ctask cidcase_id(bag_id,it.model,c)ifcidinCACHE:# 缓存命中HITS[hit]1returnCACHE[cid]HITS[miss]1rrate_packing(c,it)ifit.kindpackingelserate_tray(c,it)mmargins(r)outdict(case_idcid,flood_pctr[flood_pct],okm[ok],bindingm[binding])CACHE[cid]outreturnoutdefbuild_tasks(n_internals4,n_D6,n_load20):baseCase(rhoV2.5,rhoL780.0,muV1.2e-5,muL3e-4,sigma0.02,Vm2.0,Lm4.0)lib[]foriinrange(n_internals):ifi%20:lib.append(Internals(kindpacking,modelfPK-{i},a151.0i,eps0.97,fp_ft_inv85.0))else:lib.append(Internals(kindtray,modelfTR-{i},dh8.0,hole_ratio0.05,C00.80,K232.0))tasks[]foritinlib:forkinrange(n_D):forjinrange(n_load):load0.40.03*j tasks.append((BAG-01,it,replace(base,D0.90.1*k,Vmbase.Vm*load,Lmbase.Lm*load)))returntasksdefshard(tasks,n_shards):一致性哈希分片同一 case_id 总落同一片buckets[[]for_inrange(n_shards)]fortintasks:keycase_id(t[0],t[1].model,t[2])buckets[int(hashlib.sha1(key.encode()).hexdigest()[:8],16)%n_shards].append(t)returnbucketsdefrun(tasks,workers4):t0time.perf_counter()withThreadPoolExecutor(max_workersworkers)asex:resultslist(ex.map(rate_one,tasks))returnresults,time.perf_counter()-t0if__name____main__:tasksbuild_tasks()print(f任务总数 {len(tasks)})fornsin(1,4,8):sizes[len(x)forxinshard(tasks,ns)]print(f 分片{ns}: 各片大小 min{min(sizes)}max{max(sizes)}合计{sum(sizes)})CACHE.clear();HITS.update(hit0,miss0)res,dtrun(tasks,workers1)print(f\n[串行 1 线程] 耗时{dt:.3f}s 吞吐{len(tasks)/dt:,.0f}case/s 缓存 hit{HITS[hit]}miss{HITS[miss]})CACHE.clear();HITS.update(hit0,miss0)res,dtrun(tasks,workers8)print(f[并行 8 线程] 耗时{dt:.3f}s 吞吐{len(tasks)/dt:,.0f}case/s 缓存 hit{HITS[hit]}miss{HITS[miss]})HITS.update(hit0,miss0)# 重置计数单独观察全命中那一次res2,dt2run(tasks,workers8)print(f[二次运行 ] 耗时{dt2:.3f}s 吞吐{len(tasks)/dt2:,.0f}case/s 缓存 hit{HITS[hit]}miss{HITS[miss]})oksum(1forrinresifr[ok])print(f\n合格{ok}/{len(res)}瓶颈分布:,{b:sum(1forrinresifr[binding]b)forbinset(r[binding]forrinres)})实测输出本机 numpy 2.2.6任务总数 480 分片 1: 各片大小 min480 max480 合计480 分片 4: 各片大小 min108 max125 合计480 分片 8: 各片大小 min47 max73 合计480 [串行 1 线程] 耗时0.053s 吞吐9,135 case/s 缓存 hit0 miss480 [并行 8 线程] 耗时0.054s 吞吐8,819 case/s 缓存 hit0 miss480 [二次运行 ] 耗时0.017s 吞吐28,580 case/s 缓存 hit480 miss0 合格 292/480瓶颈分布: {weep: 186, flood: 240, dP: 54}逐段剖析CACHE用case_id为键内容寻址第 04 篇。反直觉点 1如果键漏了hL或C_sb缓存会在参数微调后重算时返回错误结果——这类 bug 极难发现因为它只在增量工况下暴露。shard()用 sha1 取 8 位十六进制再取模片号稳定、与任务顺序无关。但注意分片 8时各片 47~73最大/最小约1.55 倍——哈希只做概率均衡不做严格均衡。反直觉点 2[并行 8 线程]比[串行 1 线程]更慢8,819 vs 9,135。原因是GIL 任务粒度太小总耗时 0.05 s调度开销与计算同量级。这条实测比任何理论都更能说服人不是加了线程就快。[二次运行]缓存全命中 →28,580 case/s约 3.2 倍。这是本平台最实在的性能杠杆——而且它不依赖任何并发技巧。瓶颈分布是本篇最有业务价值的输出flood240 /weep186 /dP54。三项合计恰好 480每案例一个 binding。读法容量与操作下限是主要矛盾压降不是。dt用time.perf_counter()测性能必须用单调高精度时钟time.time()会受系统时间调整影响。代码 2-2把分片变成可断点、可重跑的片段执行器# -*- coding: utf-8 -*-shard_runner.py —— 按片执行 按片落盘断点续跑到片粒度importcsv,os,jsonfromplatform_runnerimportbuild_tasks,shard,rate_one,CACHE FIELDS[case_id,flood_pct,ok,binding]defrun_shard(tasks,shard_id,out_dir_shards):os.makedirs(out_dir,exist_okTrue)pathos.path.join(out_dir,fshard_{shard_id}.csv)ifos.path.exists(path):print(f分片{shard_id}: 已存在跳过幂等)returnpath rows[]fortintasks:rrate_one(t)rows.append({case_id:r[case_id],flood_pct:round(r[flood_pct],3)ifr[flood_pct]else,ok:int(r[ok]),binding:r[binding]})withopen(path,w,newline,encodingutf-8)asf:wcsv.DictWriter(f,fieldnamesFIELDS);w.writeheader();w.writerows(rows)print(f分片{shard_id}:{len(rows)}行 -{path})returnpathif__name____main__:tasksbuild_tasks()bucketsshard(tasks,4)CACHE.clear()forsid,binenumerate(buckets):run_shard(b,sid)# 二次运行全部跳过forsid,binenumerate(buckets):run_shard(b,sid)totalsum(1for_inopen(_shards/shard_0.csv,encodingutf-8))-1print(f\n分片0 行数不含表头{total})实测输出分片 0: 124 行 - _shards\shard_0.csv 分片 1: 108 行 - _shards\shard_1.csv 分片 2: 125 行 - _shards\shard_2.csv 分片 3: 123 行 - _shards\shard_3.csv 分片 0: 已存在跳过幂等 分片 1: 已存在跳过幂等 分片 2: 已存在跳过幂等 分片 3: 已存在跳过幂等 分片0 行数不含表头 124逐段剖析run_shard()用文件是否存在做片级幂等已算过的片直接跳过第二次四片全部跳过。这是断点续扫在片粒度的实现——相比第 11 篇的案例粒度它牺牲粒度换取更少的文件句柄与更快重启。经验法则案例数 1 万用案例级断点 1 万用片级断点 片内案例级缓存。注意四片行数是 124/108/125/123合计 480而非 120×4——这正是 1.2 节说的哈希只做概率均衡不做严格均衡。工程上不要为了追求每片 120而破坏片号稳定性那会牺牲可追责与缓存局部性接受 10% 左右的偏差更划算。三、常见报错与排查报错 3-1加线程后没有加速甚至更慢。现象如本例 8 线程 8,819 单线程 9,135。根因GIL 任务粒度太小。解法优先向量化numpy确需并发时对纯 CPU 任务用多进程对外部软件调用用多进程/多实例受许可并发约束不写具体上限第 16 篇。报错 3-2分片后各片大小差异大如 47 vs 73。现象负载不均。根因哈希只做概率均衡。解法接受偏差或改用先按预估代价排序、再轮询分配的加权分片也可动态任务队列work stealing。报错 3-3缓存启用后结果出现幽灵值。现象某工况结果与预期不符但重跑清缓存后正确。根因缓存键不完备漏了影响结果的输入。解法缓存键必须由结果函数用到的全部输入生成case_id的 payload对关键字段做键完备性回归测试第 20 篇。报错 3-4多次运行后_shards/目录里出现重复文件或半截文件。现象片文件损坏。解法先写临时文件、再原子重命名os.replace片级幂等以最终文件存在为准绝不以临时文件存在为准。报错 3-5吞吐小数点后波动很大无法判断优化是否有效。现象测量不可比。根因单次测量受预热、GC、系统负载影响。解法多次重复取中位数固定任务集与顺序并同时报告每案例平均耗时本例约 0.11 ms/case而不是只看总吞吐。四、动手练习练习 1跑通规模运行代码 2-1。判定输出 480 个任务、三档分片统计、三行吞吐单线程约 9.1k case/s、8 线程约 8.8k case/s、二次运行约 28.6k case/s容差 ±15%以及瓶颈分布flood240 /weep186 /dP54容差 0退出码 0。练习 2用多进程替代线程把ThreadPoolExecutor换成ProcessPoolExecutor注意任务需可 pickleInternals是 frozen dataclassCase是普通 dataclass均可。判定报出新吞吐并与线程版对比写出一句为什么多进程可能更快但代价是序列化与内存。练习 3片级幂等运行代码 2-2 两次。判定第一次输出四片写入行数124/108/125/123合计 480第二次四片全部输出已存在跳过幂等“且分片0 行数不含表头 124”。练习 4瓶颈解读基于练习 1 的瓶颈分布写出两条改设计方向并说明依据。判定例如容量为主矛盾flood 240→ 加大塔径或换高容量内件“操作下限也吃紧weep 186→ 减小开孔率/孔径或改阀塔板”要点是从计数反推系统性问题如整体塔径偏小、负荷区间设定过宽。五、小结与下一篇预告本篇把核算推到了规模五层架构来源/数据/引擎/任务/交付、一致性哈希分片稳定映射但不保证严格均衡、缓存是性能第一杠杆全命中 28,580 case/s约 3.2 倍、并行不是万能GIL 小任务粒度使 8 线程反而更慢、以及瓶颈分布flood 240 / weep 186 / dP 54把算完了升级为告诉业务该改什么。三条要点先向量化再并行缓存键必须完备片级幂等 原子落盘。第 20 篇《收官FRI/KG-TOWER 二次开发水力学核算工具包完整项目》把前十九篇组装成一个可发布的 CLI 工具包fri_kt——数据模型、核算引擎、批量扫描、裕度判据、报表交付、合规检查五件套配一套回归测试8 项断言把关键数值钉死与交付清单并回顾整条知识图谱。本篇认知问题回显FAQQ1企业级批量核算平台应分成哪几层为什么任务层必须独立A五层——来源层流程模拟契约/软件报表/人工录入、数据层工况矩阵、内件库、版本化阈值配置、引擎层纯函数的核算裕度不确定度、任务层任务生成、分片、缓存、重试、断点、交付层长表、宽表报表、图形、合规元数据。任务层必须独立因为只有引擎是纯函数任务层才敢对它做缓存、并行与重试两层混在一起会让重算与缓存互相污染。Q2任务分片为什么用一致性哈希解决了什么、没解决什么A片号 int(sha1(case_id)[:8],16) mod N。它解决案例到分片的稳定映射与任务顺序无关从而支持分布式执行各机器各跑一片、缓存局部性与按片追责。它不解决负载均衡——哈希只做概率均衡示例 8 片时片大小 47~73最大/最小约 1.55 倍严格均衡需加权分片或动态任务队列。Q3为什么本例8 线程比单线程还慢什么条件下并行才有收益A两个原因GIL 使纯 Python 计算无法真正并行任务粒度太小480 任务总耗时约 0.05 s调度开销与计算同量级。实测单线程 9,135 case/s、8 线程 8,819 case/s。并行有收益的条件纯 CPU 且任务较重时用多进程外部软件调用秒级、进程级用多进程/多实例受许可并发约束I/O 密集用线程池。经验法则是先向量化再考虑并行。Q4缓存能带来多少收益缓存键应该是什么A示例缓存全命中把吞吐从 8,819 升到 28,580 case/s约 3.2 倍是最实在的性能杠杆且不依赖并发技巧。缓存键必须是内容寻址的 case_id由结果函数用到的全部输入经规范化后哈希。若键漏掉任一影响结果的输入如 hL、C_sb缓存会在参数微调后返回错误但看起来有效的结果这类 bug 比算错更难查。Q5怎么用瓶颈分布回答业务问题A瓶颈分布把算完了升级为该改什么。示例flood 240 / weep 186 / dP 54合计 480每案例一个 binding。读法容量是主要限制加大塔径或换高容量内件操作下限也大面积吃紧减小开孔率/孔径或改阀塔板压降不是主矛盾。若容量与下限同时紧张说明这不是单点问题而是塔径/内件族系统性偏小或负荷区间设定过宽。