编写和运行任务 DAG 的最简洁方法是什么？

Question

我想编写运行一个有向无环图 (DAG)，其中包含多个任务运行串行或并行。理想情况下它看起来像：

def task1():
    # ...

def task2():
    # ...

graph = Sequence([
    task1,
    task2,
    Parallel([
        task3,
        task4
    ]),
    task5
]

graph.run()

它会运行 1 -> 2 ->（3 和 4 同时） -> 5。任务需要访问全局范围以存储结果、写入日志和访问命令行参数。

我的用例是编写部署脚本。 并行任务受 IO 限制： 通常在远程服务器上等待完成一个步骤。

我研究了线程、asyncio、Airflow，但没有找到任何简单的库可以在没有样板代码的情况下允许它遍历和控制图形的执行。有这样的东西吗？

Answer 1

这是一个快速的概念验证实现。它可以像这样使用：

graph = sequence(
            lambda: print(1),
            lambda: print(2),
            parallel(
                lambda: print(3),
                lambda: print(4),
                sequence(
                    lambda: print(5),
                    lambda: print(6))),
             lambda: print(7)

graph()

1
2
3
5
6
4
7

sequence 生成一个包装 for 循环的函数，parallel 生成一个包装使用线程池的函数：

from typing import Callable
from multiprocessing.pool import ThreadPool

Task = Callable[[], None]

_pool: ThreadPool = ThreadPool()

def sequence(*tasks: Task) -> Task:
    def run():
        for task in tasks:
            task()

    return run  # Returning "run" to be used as a task by other "sequence" and "parallel" calls

def parallel(*tasks: Task) -> Task:
    def run():
        _pool.map(lambda f: f(), tasks)  # Delegate to a pool used for IO tasks

    return run

每次调用 sequence 和 parallel returns 一个新的 "Task"（一个不带参数且不返回任何内容的函数）。然后可以通过对 sequence 和 parallel.

的其他外部调用来调用该任务

关于 ThreadPool 的注意事项：

虽然这确实为 parallel 使用了线程池，但由于 GIL，这仍然一次只能执行一件事。这意味着 parallel 对于 CPU 绑定的任务基本上是无用的。
我没有指定池应该以多少线程开始。我认为它默认为您可以使用的内核数。如果需要更多，可以使用第一个参数 ThreadPool 指定要开始的数量。
为简洁起见，我不会清理 ThreadPool。如果你使用这个，你绝对应该这样做。
尽管 ThreadPool 是 multiprocessing 的一部分，但令人困惑的是它使用线程而不是进程。

Answer 2

您提到您的任务是 IO 绑定的，这意味着 asycnio 将是一个很好的选择。您可以尝试 aiodag 库，它是 asycnio 之上的一个极其轻便的接口，可让您轻松定义异步 dags：

import asyncio
from aiodag import task

@task
async def task1(x):
    ...

@task
async def task2(x):
    ...

@task
async def task3(x):
    ...

@task
async def task4(x):
    ...

@task
async def task5(x, y):
    ...

# rest of task funcs

async def main():
    t1 = task1()
    t2 = task2(t1)
    t3 = task3(t2)  # t3/t4 take t2, when t2 finishes, will run concurrently
    t4 = task4(t2)
    t5 = task5(t3, t4) # will wait until t3/t4 finish to execute
    await t5

loop = asyncio.new_event_loop()
asyncio.run_until_complete(main())

查看 aiodag github 页面上的自述文件，了解有关如何 constructed/optimally 执行 dag 的一些详细信息。 https://github.com/aa1371/aiodag

如果你不想被异步函数所束缚，那么请查看 dask 的延迟接口。 dag 的定义与 aiodag 的定义方式相同，其中 dag 是通过函数调用构造的。 Dask 将以最佳并行方案无缝处理执行你的 dag，并且可以分布在任意大的集群上以执行并行执行。

https://docs.dask.org/en/latest/delayed.html

编写和运行任务 DAG 的最简洁方法是什么？

What is the cleanest way to write and run a DAG of tasks?

python

concurrency

directed-acyclic-graphs

编写和 运行 任务 DAG 的最简洁方法是什么？

What is the cleanest way to write and run a DAG of tasks?

python

concurrency

directed-acyclic-graphs

编写和运行任务 DAG 的最简洁方法是什么？