我有以下任務(wù)要解決：文件不定期通過端點(diǎn)發(fā)送并存儲在本地。我需要為這些文件中的每一個(gè)觸發(fā) DAG 運(yùn)行。對于每個(gè)文件，將執(zhí)行相同的任務(wù)總體而言，流程如下所示：對于每個(gè)文件，運(yùn)行任務(wù) A->B->C->D正在批量處理文件。雖然這項(xiàng)任務(wù)對我來說似乎微不足道，但我已經(jīng)找到了幾種方法來做到這一點(diǎn)，我對哪一種是“正確的”（如果有的話）感到困惑。第一種模式：使用實(shí)驗(yàn)性 REST API 觸發(fā) dag。也就是說，公開一個(gè) Web 服務(wù)，它攝取請求和文件，將其存儲到文件夾中，并使用實(shí)驗(yàn)性 REST api通過將 file_id 作為 conf 來觸發(fā) DAG缺點(diǎn)：REST api 仍處于實(shí)驗(yàn)階段，不確定 Airflow 如何處理同時(shí)出現(xiàn)許多請求的負(fù)載測試（這不應(yīng)該發(fā)生，但是，如果發(fā)生了怎么辦？）第二種模式：2 dags。一個(gè)用 TriggerDagOperator 感知和觸發(fā)，一個(gè)進(jìn)程。始終使用與之前描述的相同的 ws，但這次它只是存儲文件。然后我們有：第一個(gè) dag：使用 FileSensor 和 TriggerDagOperator 來觸發(fā) N 個(gè)給定 N 個(gè)文件的 dag第二天：任務(wù)A->B->C缺點(diǎn)：需要避免將相同的文件發(fā)送到兩個(gè)不同的 DAG 運(yùn)行。例子：文件夾 x.json 中的文件 傳感器找到 x，觸發(fā) DAG (1)傳感器返回并再次安排。如果 DAG (1) 未處理/移動(dòng)文件，則傳感器 DAG 可能會重新安排使用相同文件運(yùn)行的新 DAG。這是不需要的。第三種模式：對于文件中的文件，任務(wù) A->B->C正如在這個(gè)問題中看到的那樣。缺點(diǎn)：這可能有效，但我不喜歡 UI 可能會變得混亂，因?yàn)槊看?DAG 運(yùn)行看起來都不一樣，但它會隨著正在處理的文件數(shù)量而改變。此外，如果要處理 1000 個(gè)文件，則運(yùn)行可能會非常難以閱讀第四種模式：使用 subdags我還不確定它們是如何完全工作的，因?yàn)槲铱吹?nbsp;它們不被鼓勵(lì)（最后），但是應(yīng)該可以為每個(gè)文件生成一個(gè) subdag 并讓它運(yùn)行。類似于這個(gè)問題。缺點(diǎn)：似乎 subdags 只能與順序執(zhí)行器一起使用。我是否遺漏了什么并且過度思考了一些應(yīng)該（在我看來）非常直截了當(dāng)?shù)氖虑椋恐x謝