Prefect 缓存策略实战指南3 步让任务不再重复计算【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefectPrefect 是一个用 Python 构建数据管道的任务编排框架它的任务缓存Cache Policy能帮你把算过的任务结果存下来、下次直接取从而省掉大量重复计算。这篇文章围绕三个问题展开你的任务该不该缓存、Prefect 缓存策略怎么配、配置之后怎么验证命中、出了问题怎么排。全文以可操作步骤为主跟着做就能跑通。一、开场同一条数据一天被算了三遍想象这样一个画面你的夜间 ETL 流程每小时跑一次其中拉取昨日订单汇总这个任务每次都要连数据库、跑两分钟聚合。凌晨 2 点跑一次、4 点跑一次、6 点再跑一次——输入参数一模一样结果一模一样机器空转六分钟。你会遇到的情况就是这类任务重复执行但产出没有任何变化。在 Prefect 里task默认每次都真跑。如果你给它配上缓存策略第二次执行时框架会直接拿出上次的结果任务耗时从两分钟变成几秒。判断标准很简单这个任务换个时间点再算一遍结果会不会变。如果答案是不会它就有缓存价值。那么哪些任务符合这个条件下一章给一张自检表。二、该不该缓存先做判断再动手缓存不是越用越好。给一个有副作用或结果随时变化的任务贴缓存反而会让你读到过期数据。动手之前先按下面这张表自检你的任务类型任务特征适合缓存典型例子幂等同样的输入永远得到同样的输出适合读数、聚合、格式转换输入稳定参数在短期内不会变适合固定表名、固定查询条件计算成本高或调用外部 API 有配额适合大表 join、付费接口查询结果有时效性分钟级就变谨慎配短过期时间实时价格、库存有副作用写库、发消息、扣款不适合任何执行一次就该只执行一次的操作随机性内部用了随机数、当前时间戳不适合除非把时间固定进输入抽样、打时间戳的报表怎么用这张表挑出适合一栏里的任务去配缓存不适合的一栏保持默认每次执行谨慎的一栏配缓存但把过期时间压短。验证这一步是否做对也很直接给一个幂等任务开缓存、给一个写库任务不开跑两遍流程前者第二次秒回后者两次都真实执行——符合预期说明判断对了。三、看懂机制一次 Prefect 缓存的一生Prefect 缓存的源码实现分布在两处任务侧的task参数src/prefect/tasks.py和编排侧的 核心策略源码。在CoreTaskPolicy的规则优先级里CacheRetrieval排在最前、CacheInsertion排在后段——翻译过来就是先查缓存再写缓存。你可以把缓存理解成给算过的结果贴标签标签缓存键一样就直接取下上次贴的标签里的东西。按时间线看一次任务的完整生命周期未命中任务启动前CacheRetrieval规则拿缓存键去查库。第一次跑库里没有对应记录放行执行。执行任务真实运行跑完进入成功终态。写入CacheInsertion规则把缓存键 → 这次的状态/结果写进数据库。存储表是task_run_state_cache定义在 ORM 模型 里每条记录包含缓存键、关联的状态 ID 和创建时间。命中下次同键任务启动第 1 步查到记录任务直接跳到成功态不再执行函数体。过期/失效如果配置了过期时间写入时超过时长的旧记录会被忽略或者你改了输入、改了版本号键变了旧缓存自然作废。验证机制生效的办法跑同一流程两遍看第二个任务运行是否没有实际执行函数日志里没有函数内的打印状态却直接是 Completed并且任务记录上能看到cache_key。四、动手三步配出可用的任务缓存第一步给任务生成缓存键没有缓存键就没有缓存。最小可用配置是套用内置的task_input_hash它会对任务名、函数代码和全部入参做哈希——改一个参数键就变缓存自动失效from prefect import task from prefect.tasks import task_input_hash task(cache_key_fntask_input_hash) def build_summary(date: str, region: str): # 你的查询逻辑 return run_query(...)这一行cache_key_fntask_input_hash就是全部开关。跑两遍同参数流程验证第二遍该任务应秒级结束。第二步设置缓存多久过期结果有时效的任务加一个cache_expiration即可。经验值是不超过你业务上能容忍的数据延迟from datetime import timedelta task( cache_key_fntask_input_hash, cache_expirationtimedelta(hours6), ) def fetch_price(): ...6 小时内的重复请求走缓存超过 6 小时自动重新计算。想验证等过期后或临时改成几分钟再跑确认任务重新真实执行。第三步规划好失效手段有三种让旧缓存作废的方式按常用程度排改输入task_input_hash已覆盖参数一变自动失效换版本改逻辑但输入没变时用task(version2.0)版本号参与哈希升级即全部失效清空重来极端情况直接删库里的task_run_state_cache记录谨慎使用。三步走完你的任务就有了自动命中、按时间过期、按版本换代的完整闭环。五、排障命中率和键冲突的现场排查缓存配好之后最常见的两类问题是该命中没命中和命中了错误的数据。按下面的流程走不用靠猜症状 A命中率低任务总在重跑定位先看是不是参数看起来一样、实际不一样——task_input_hash会对入参做严格哈希字典顺序外的差异、datetime和字符串的差异都会改变键打印两次运行的cache_key对比一下。定位确认函数体有没有被改过——task_input_hash会把函数字节码算进键改了一行代码等于全量失效这是特性不是故障。处理把每次都变的参数如当前时间从入参挪进函数内部或改用自定义cache_key_fn只哈希关键参数。症状 B命中了但数据是错的键冲突定位不同任务或不同环境的同名函数撞了缓存键。task_input_hash含任务名通常不会跨任务冲突自己写的简单cache_key_fn最容易漏项。处理给键加前缀或命名空间例如把任务名、版本号、环境名都拼进去做到一个键只属于一个任务的一个版本。验证修复后清掉旧键对应的记录重跑确认新旧环境各走各的缓存。症状 C缓存越攒越大定位查task_run_state_cache表记录数是否持续上涨多半是没设过期时间。处理给任务补上cache_expiration或定期清理无引用的旧记录。排障的总原则先比键再比版本最后才怀疑框架。六、进阶按环境和条件切换缓存行为你经常需要开发环境每次都真跑、生产环境尽量走缓存这类行为差异。做法是自己写一个cache_key_fn让它根据条件返回键或返回None返回None表示这次不缓存import os from prefect.tasks import task_input_hash def env_cache_key(context, arguments): if os.environ.get(PREFECT_ENV) dev: return None return task_input_hash(context, arguments) task(cache_key_fnenv_cache_key) def etl_step(...): ...注意签名是(context, arguments)context里能拿到任务运行上下文arguments是入参字典。同样的写法也可以做成按租户、按数据分区、按开关动态切换。验证方式在 dev 环境跑两遍确认每次都真执行切到生产环境再跑两遍第二遍应命中。这样同一份代码行为随环境自动切换不需要维护两套任务定义。七、收尾上线前自查清单 延伸资源把缓存策略用到生产之前逐项勾一遍只给幂等、输入稳定的任务开了缓存副作用任务保持每跑必执行缓存键用task_input_hash或包含任务名 版本 关键入参的自定义键无跨任务冲突有时效的结果都设了cache_expiration时长不超过业务可容忍延迟改逻辑时用version升级触发全量失效而不是靠祈祷开发环境禁缓存、生产启用缓存的行为已验证跑了两遍流程确认第二遍命中缓存且结果正确知道去哪里查task_run_state_cache表来排查键冲突延伸资源均为仓库内文件可相对路径直接打开缓存键生成与任务参数定义src/prefect/tasks.py检索/写入规则的编排顺序src/prefect/server/orchestration/core_policy.py存储表结构src/prefect/server/database/orm_models.py官方文档索引docs/缓存这件事Prefect 把何时命中、何时写入、何时失效都收敛到了任务参数和编排规则里。你只需要回答三个问题键怎么生成、过期多久、什么时候换版本。答对了重复计算就消失了。【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考