每个数据的 Airflow Dagrun，而不是预定的

Question

我当前面临的问题是，我在 MongoDB 集合中有文档，每个文档都需要由需要在非循环依赖图中运行的任务进行处理和更新。如果上游任务无法处理文档，则任何依赖任务都无法处理该文档，因为该文档尚未使用先决条件信息进行更新。

如果我要使用 Airflow，这给我留下了两个解决方案:

1. 为每个文档触发一个DAG，并使用--conf传入文档ID。这样做的问题是，这不是 Airflow 的预期使用方式；我永远不会运行预定的流程，并且根据文档在集合中的显示方式，我每天将赚取 1440 Dagrun。

2. 每个周期运行 DAG 来处理该周期在集合中创建的所有文档。这遵循 Airflow 预期的工作方式，但问题是，如果任务无法处理单个文档，则任何依赖任务都无法处理任何其他文档。此外，如果某项任务处理某个文档的时间比其他文档长，那么其他文档就会等待该文档继续沿着 DAG 向下移动。

还有比 Airflow 更好的方法吗？或者在 Airflow 中是否有比我目前看到的两种方法更好的方法来处理这个问题？

Answer 1

根据我在尝试回答这个问题时获得的知识，我得出的结论是Airflow 并不是完成这项工作的工具。

Airflow 专为预定、幂等 DAG 设计。 DagRun 还必须具有唯一的 execution_date；这意味着在完全相同的开始时间运行相同的 DAG(在我们同时接收两个文档的情况下几乎是不可能的。当然，我们可以连续立即安排下一个 DagRun，但这种限制应该表明任何在某种程度上，尝试以这种方式使用 Airflow 始终是一种黑客行为。

我发现的最可行的解决方案是使用 Prefect ，其开发目的是为了克服 Airflow 的一些限制:

“Prefect 假设流程可以出于任何原因随时运行。”

Prefect 相当于 DAG 的是 Flow；我们可以利用的流程的一个关键优势是易于参数化。然后，通过一些线程，我们可以为流中的每个元素运行一个 Flow。以下是一个流式 ETL 管道示例:

import time
from prefect import task, Flow, Parameter
from threading import Thread
​
​
def stream():
    for x in range(10):
        yield x
        time.sleep(1)
​
​
@task
def extract(x):
    # If 'x' referenced a document, in this step we could load that document
    return x
​
​
@task
def transform(x):
    return x * 2
​
​
@task
def load(y):
    print("Received y: {}".format(y))
​
​
with Flow("ETL") as flow:
    x_param = Parameter('x')
    e = extract(x_param)
    t = transform(e)
    l = load(t)
​
for x in stream():
    thread = Thread(target=flow.run, kwargs={"x": x})
    thread.start()