指南：3 步讓任務不再重復計算)
Prefect 緩存策略實戰(zhàn)指南3 步讓任務不再重復計算【免費下載鏈接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.項目地址: https://gitcode.com/GitHub_Trending/pr/prefectPrefect 是一個用 Python 構建數(shù)據(jù)管道的任務編排框架它的任務緩存Cache Policy能幫你把算過的任務結果存下來、下次直接取從而省掉大量重復計算。這篇文章圍繞三個問題展開你的任務該不該緩存、Prefect 緩存策略怎么配、配置之后怎么驗證命中、出了問題怎么排。全文以可操作步驟為主跟著做就能跑通。一、開場同一條數(shù)據(jù)一天被算了三遍想象這樣一個畫面你的夜間 ETL 流程每小時跑一次其中拉取昨日訂單匯總這個任務每次都要連數(shù)據(jù)庫、跑兩分鐘聚合。凌晨 2 點跑一次、4 點跑一次、6 點再跑一次——輸入?yún)?shù)一模一樣結果一模一樣機器空轉六分鐘。你會遇到的情況就是這類任務重復執(zhí)行但產(chǎn)出沒有任何變化。在 Prefect 里task默認每次都真跑。如果你給它配上緩存策略第二次執(zhí)行時框架會直接拿出上次的結果任務耗時從兩分鐘變成幾秒。判斷標準很簡單這個任務換個時間點再算一遍結果會不會變。如果答案是不會它就有緩存價值。那么哪些任務符合這個條件下一章給一張自檢表。二、該不該緩存先做判斷再動手緩存不是越用越好。給一個有副作用或結果隨時變化的任務貼緩存反而會讓你讀到過期數(shù)據(jù)。動手之前先按下面這張表自檢你的任務類型任務特征適合緩存典型例子冪等同樣的輸入永遠得到同樣的輸出適合讀數(shù)、聚合、格式轉換輸入穩(wěn)定參數(shù)在短期內(nèi)不會變適合固定表名、固定查詢條件計算成本高或調用外部 API 有配額適合大表 join、付費接口查詢結果有時效性分鐘級就變謹慎配短過期時間實時價格、庫存有副作用寫庫、發(fā)消息、扣款不適合任何執(zhí)行一次就該只執(zhí)行一次的操作隨機性內(nèi)部用了隨機數(shù)、當前時間戳不適合除非把時間固定進輸入抽樣、打時間戳的報表怎么用這張表挑出適合一欄里的任務去配緩存不適合的一欄保持默認每次執(zhí)行謹慎的一欄配緩存但把過期時間壓短。驗證這一步是否做對也很直接給一個冪等任務開緩存、給一個寫庫任務不開跑兩遍流程前者第二次秒回后者兩次都真實執(zhí)行——符合預期說明判斷對了。三、看懂機制一次 Prefect 緩存的一生Prefect 緩存的源碼實現(xiàn)分布在兩處任務側的task參數(shù)src/prefect/tasks.py和編排側的 核心策略源碼。在CoreTaskPolicy的規(guī)則優(yōu)先級里CacheRetrieval排在最前、CacheInsertion排在后段——翻譯過來就是先查緩存再寫緩存。你可以把緩存理解成給算過的結果貼標簽標簽緩存鍵一樣就直接取下上次貼的標簽里的東西。按時間線看一次任務的完整生命周期未命中任務啟動前CacheRetrieval規(guī)則拿緩存鍵去查庫。第一次跑庫里沒有對應記錄放行執(zhí)行。執(zhí)行任務真實運行跑完進入成功終態(tài)。寫入CacheInsertion規(guī)則把緩存鍵 → 這次的狀態(tài)/結果寫進數(shù)據(jù)庫。存儲表是task_run_state_cache定義在 ORM 模型 里每條記錄包含緩存鍵、關聯(lián)的狀態(tài) ID 和創(chuàng)建時間。命中下次同鍵任務啟動第 1 步查到記錄任務直接跳到成功態(tài)不再執(zhí)行函數(shù)體。過期/失效如果配置了過期時間寫入時超過時長的舊記錄會被忽略或者你改了輸入、改了版本號鍵變了舊緩存自然作廢。驗證機制生效的辦法跑同一流程兩遍看第二個任務運行是否沒有實際執(zhí)行函數(shù)日志里沒有函數(shù)內(nèi)的打印狀態(tài)卻直接是 Completed并且任務記錄上能看到cache_key。四、動手三步配出可用的任務緩存第一步給任務生成緩存鍵沒有緩存鍵就沒有緩存。最小可用配置是套用內(nèi)置的task_input_hash它會對任務名、函數(shù)代碼和全部入?yún)⒆龉！囊粋€參數(shù)鍵就變緩存自動失效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就是全部開關。跑兩遍同參數(shù)流程驗證第二遍該任務應秒級結束。第二步設置緩存多久過期結果有時效的任務加一個cache_expiration即可。經(jīng)驗值是不超過你業(yè)務上能容忍的數(shù)據(jù)延遲from datetime import timedelta task( cache_key_fntask_input_hash, cache_expirationtimedelta(hours6), ) def fetch_price(): ...6 小時內(nèi)的重復請求走緩存超過 6 小時自動重新計算。想驗證等過期后或臨時改成幾分鐘再跑確認任務重新真實執(zhí)行。第三步規(guī)劃好失效手段有三種讓舊緩存作廢的方式按常用程度排改輸入task_input_hash已覆蓋參數(shù)一變自動失效換版本改邏輯但輸入沒變時用task(version2.0)版本號參與哈希升級即全部失效清空重來極端情況直接刪庫里的task_run_state_cache記錄謹慎使用。三步走完你的任務就有了自動命中、按時間過期、按版本換代的完整閉環(huán)。五、排障命中率和鍵沖突的現(xiàn)場排查緩存配好之后最常見的兩類問題是該命中沒命中和命中了錯誤的數(shù)據(jù)。按下面的流程走不用靠猜癥狀 A命中率低任務總在重跑定位先看是不是參數(shù)看起來一樣、實際不一樣——task_input_hash會對入?yún)⒆鰢栏窆Ｗ值漤樞蛲獾牟町?、datetime和字符串的差異都會改變鍵打印兩次運行的cache_key對比一下。定位確認函數(shù)體有沒有被改過——task_input_hash會把函數(shù)字節(jié)碼算進鍵改了一行代碼等于全量失效這是特性不是故障。處理把每次都變的參數(shù)如當前時間從入?yún)⑴策M函數(shù)內(nèi)部或改用自定義cache_key_fn只哈希關鍵參數(shù)。癥狀 B命中了但數(shù)據(jù)是錯的鍵沖突定位不同任務或不同環(huán)境的同名函數(shù)撞了緩存鍵。task_input_hash含任務名通常不會跨任務沖突自己寫的簡單cache_key_fn最容易漏項。處理給鍵加前綴或命名空間例如把任務名、版本號、環(huán)境名都拼進去做到一個鍵只屬于一個任務的一個版本。驗證修復后清掉舊鍵對應的記錄重跑確認新舊環(huán)境各走各的緩存。癥狀 C緩存越攢越大定位查task_run_state_cache表記錄數(shù)是否持續(xù)上漲多半是沒設過期時間。處理給任務補上cache_expiration或定期清理無引用的舊記錄。排障的總原則先比鍵再比版本最后才懷疑框架。六、進階按環(huán)境和條件切換緩存行為你經(jīng)常需要開發(fā)環(huán)境每次都真跑、生產(chǎn)環(huán)境盡量走緩存這類行為差異。做法是自己寫一個cache_key_fn讓它根據(jù)條件返回鍵或返回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是入?yún)⒆值?。同樣的寫法也可以做成按租戶、按?shù)據(jù)分區(qū)、按開關動態(tài)切換。驗證方式在 dev 環(huán)境跑兩遍確認每次都真執(zhí)行切到生產(chǎn)環(huán)境再跑兩遍第二遍應命中。這樣同一份代碼行為隨環(huán)境自動切換不需要維護兩套任務定義。七、收尾上線前自查清單 延伸資源把緩存策略用到生產(chǎn)之前逐項勾一遍只給冪等、輸入穩(wěn)定的任務開了緩存副作用任務保持每跑必執(zhí)行緩存鍵用task_input_hash或包含任務名 版本 關鍵入?yún)⒌淖远x鍵無跨任務沖突有時效的結果都設了cache_expiration時長不超過業(yè)務可容忍延遲改邏輯時用version升級觸發(fā)全量失效而不是靠祈禱開發(fā)環(huán)境禁緩存、生產(chǎn)啟用緩存的行為已驗證跑了兩遍流程確認第二遍命中緩存且結果正確知道去哪里查task_run_state_cache表來排查鍵沖突延伸資源均為倉庫內(nèi)文件可相對路徑直接打開緩存鍵生成與任務參數(shù)定義src/prefect/tasks.py檢索/寫入規(guī)則的編排順序src/prefect/server/orchestration/core_policy.py存儲表結構src/prefect/server/database/orm_models.py官方文檔索引docs/緩存這件事Prefect 把何時命中、何時寫入、何時失效都收斂到了任務參數(shù)和編排規(guī)則里。你只需要回答三個問題鍵怎么生成、過期多久、什么時候換版本。答對了重復計算就消失了?！久赓M下載鏈接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.項目地址: https://gitcode.com/GitHub_Trending/pr/prefect創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 Prefect 緩存策略實戰(zhàn)指南：3 步讓任務不再重復計算
Prefect 緩存策略實戰(zhàn)指南3 步讓任務不再重復計算【免費下載鏈接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.項目地址: https://gitcode.com/GitHub_Trending/pr/prefectPrefect 是一個用 Python 構建數(shù)據(jù)管道的任務編排框架它的任務緩存Cache Policy能幫你把算過的任務結果存下來、下次直接取從而省掉大量重復計算。這篇文章圍繞三個問題展開你的任務該不該緩存、Prefect 緩存策略怎么配、配置之后怎么驗證命中、出了問題怎么排。全文以可操作步驟為主跟著做就能跑通。一、開場同一條數(shù)據(jù)一天被算了三遍想象這樣一個畫面你的夜間 ETL 流程每小時跑一次其中拉取昨日訂單匯總這個任務每次都要連數(shù)據(jù)庫、跑兩分鐘聚合。凌晨 2 點跑一次、4 點跑一次、6 點再跑一次——輸入?yún)?shù)一模一樣結果一模一樣機器空轉六分鐘。你會遇到的情況就是這類任務重復執(zhí)行但產(chǎn)出沒有任何變化。在 Prefect 里task默認每次都真跑。如果你給它配上緩存策略第二次執(zhí)行時框架會直接拿出上次的結果任務耗時從兩分鐘變成幾秒。判斷標準很簡單這個任務換個時間點再算一遍結果會不會變。如果答案是不會它就有緩存價值。那么哪些任務符合這個條件下一章給一張自檢表。二、該不該緩存先做判斷再動手緩存不是越用越好。給一個有副作用或結果隨時變化的任務貼緩存反而會讓你讀到過期數(shù)據(jù)。動手之前先按下面這張表自檢你的任務類型任務特征適合緩存典型例子冪等同樣的輸入永遠得到同樣的輸出適合讀數(shù)、聚合、格式轉換輸入穩(wěn)定參數(shù)在短期內(nèi)不會變適合固定表名、固定查詢條件計算成本高或調用外部 API 有配額適合大表 join、付費接口查詢結果有時效性分鐘級就變謹慎配短過期時間實時價格、庫存有副作用寫庫、發(fā)消息、扣款不適合任何執(zhí)行一次就該只執(zhí)行一次的操作隨機性內(nèi)部用了隨機數(shù)、當前時間戳不適合除非把時間固定進輸入抽樣、打時間戳的報表怎么用這張表挑出適合一欄里的任務去配緩存不適合的一欄保持默認每次執(zhí)行謹慎的一欄配緩存但把過期時間壓短。驗證這一步是否做對也很直接給一個冪等任務開緩存、給一個寫庫任務不開跑兩遍流程前者第二次秒回后者兩次都真實執(zhí)行——符合預期說明判斷對了。三、看懂機制一次 Prefect 緩存的一生Prefect 緩存的源碼實現(xiàn)分布在兩處任務側的task參數(shù)src/prefect/tasks.py和編排側的 核心策略源碼。在CoreTaskPolicy的規(guī)則優(yōu)先級里CacheRetrieval排在最前、CacheInsertion排在后段——翻譯過來就是先查緩存再寫緩存。你可以把緩存理解成給算過的結果貼標簽標簽緩存鍵一樣就直接取下上次貼的標簽里的東西。按時間線看一次任務的完整生命周期未命中任務啟動前CacheRetrieval規(guī)則拿緩存鍵去查庫。第一次跑庫里沒有對應記錄放行執(zhí)行。執(zhí)行任務真實運行跑完進入成功終態(tài)。寫入CacheInsertion規(guī)則把緩存鍵 → 這次的狀態(tài)/結果寫進數(shù)據(jù)庫。存儲表是task_run_state_cache定義在 ORM 模型 里每條記錄包含緩存鍵、關聯(lián)的狀態(tài) ID 和創(chuàng)建時間。命中下次同鍵任務啟動第 1 步查到記錄任務直接跳到成功態(tài)不再執(zhí)行函數(shù)體。過期/失效如果配置了過期時間寫入時超過時長的舊記錄會被忽略或者你改了輸入、改了版本號鍵變了舊緩存自然作廢。驗證機制生效的辦法跑同一流程兩遍看第二個任務運行是否沒有實際執(zhí)行函數(shù)日志里沒有函數(shù)內(nèi)的打印狀態(tài)卻直接是 Completed并且任務記錄上能看到cache_key。四、動手三步配出可用的任務緩存第一步給任務生成緩存鍵沒有緩存鍵就沒有緩存。最小可用配置是套用內(nèi)置的task_input_hash它會對任務名、函數(shù)代碼和全部入?yún)⒆龉！囊粋€參數(shù)鍵就變緩存自動失效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就是全部開關。跑兩遍同參數(shù)流程驗證第二遍該任務應秒級結束。第二步設置緩存多久過期結果有時效的任務加一個cache_expiration即可。經(jīng)驗值是不超過你業(yè)務上能容忍的數(shù)據(jù)延遲from datetime import timedelta task( cache_key_fntask_input_hash, cache_expirationtimedelta(hours6), ) def fetch_price(): ...6 小時內(nèi)的重復請求走緩存超過 6 小時自動重新計算。想驗證等過期后或臨時改成幾分鐘再跑確認任務重新真實執(zhí)行。第三步規(guī)劃好失效手段有三種讓舊緩存作廢的方式按常用程度排改輸入task_input_hash已覆蓋參數(shù)一變自動失效換版本改邏輯但輸入沒變時用task(version2.0)版本號參與哈希升級即全部失效清空重來極端情況直接刪庫里的task_run_state_cache記錄謹慎使用。三步走完你的任務就有了自動命中、按時間過期、按版本換代的完整閉環(huán)。五、排障命中率和鍵沖突的現(xiàn)場排查緩存配好之后最常見的兩類問題是該命中沒命中和命中了錯誤的數(shù)據(jù)。按下面的流程走不用靠猜癥狀 A命中率低任務總在重跑定位先看是不是參數(shù)看起來一樣、實際不一樣——task_input_hash會對入?yún)⒆鰢栏窆Ｗ值漤樞蛲獾牟町?、datetime和字符串的差異都會改變鍵打印兩次運行的cache_key對比一下。定位確認函數(shù)體有沒有被改過——task_input_hash會把函數(shù)字節(jié)碼算進鍵改了一行代碼等于全量失效這是特性不是故障。處理把每次都變的參數(shù)如當前時間從入?yún)⑴策M函數(shù)內(nèi)部或改用自定義cache_key_fn只哈希關鍵參數(shù)。癥狀 B命中了但數(shù)據(jù)是錯的鍵沖突定位不同任務或不同環(huán)境的同名函數(shù)撞了緩存鍵。task_input_hash含任務名通常不會跨任務沖突自己寫的簡單cache_key_fn最容易漏項。處理給鍵加前綴或命名空間例如把任務名、版本號、環(huán)境名都拼進去做到一個鍵只屬于一個任務的一個版本。驗證修復后清掉舊鍵對應的記錄重跑確認新舊環(huán)境各走各的緩存。癥狀 C緩存越攢越大定位查task_run_state_cache表記錄數(shù)是否持續(xù)上漲多半是沒設過期時間。處理給任務補上cache_expiration或定期清理無引用的舊記錄。排障的總原則先比鍵再比版本最后才懷疑框架。六、進階按環(huán)境和條件切換緩存行為你經(jīng)常需要開發(fā)環(huán)境每次都真跑、生產(chǎn)環(huán)境盡量走緩存這類行為差異。做法是自己寫一個cache_key_fn讓它根據(jù)條件返回鍵或返回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是入?yún)⒆值?。同樣的寫法也可以做成按租戶、按?shù)據(jù)分區(qū)、按開關動態(tài)切換。驗證方式在 dev 環(huán)境跑兩遍確認每次都真執(zhí)行切到生產(chǎn)環(huán)境再跑兩遍第二遍應命中。這樣同一份代碼行為隨環(huán)境自動切換不需要維護兩套任務定義。七、收尾上線前自查清單 延伸資源把緩存策略用到生產(chǎn)之前逐項勾一遍只給冪等、輸入穩(wěn)定的任務開了緩存副作用任務保持每跑必執(zhí)行緩存鍵用task_input_hash或包含任務名 版本 關鍵入?yún)⒌淖远x鍵無跨任務沖突有時效的結果都設了cache_expiration時長不超過業(yè)務可容忍延遲改邏輯時用version升級觸發(fā)全量失效而不是靠祈禱開發(fā)環(huán)境禁緩存、生產(chǎn)啟用緩存的行為已驗證跑了兩遍流程確認第二遍命中緩存且結果正確知道去哪里查task_run_state_cache表來排查鍵沖突延伸資源均為倉庫內(nèi)文件可相對路徑直接打開緩存鍵生成與任務參數(shù)定義src/prefect/tasks.py檢索/寫入規(guī)則的編排順序src/prefect/server/orchestration/core_policy.py存儲表結構src/prefect/server/database/orm_models.py官方文檔索引docs/緩存這件事Prefect 把何時命中、何時寫入、何時失效都收斂到了任務參數(shù)和編排規(guī)則里。你只需要回答三個問題鍵怎么生成、過期多久、什么時候換版本。答對了重復計算就消失了?！久赓M下載鏈接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.項目地址: https://gitcode.com/GitHub_Trending/pr/prefect創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考