2

我的想法是foo创建一个生成输入列表（用户、报告、日志文件等）的任务，并为输入列表中的每个元素启动一个任务。目标是利用 Airflow 的重试和其他逻辑，而不是重新实现它。

所以，理想情况下，我的 DAG 应该是这样的： 在此处输入图像描述

这里唯一的变量是生成的任务数。在所有这些都完成后，我想再做一些任务，因此为每项任务设置一个新的 DAG 似乎并不合适。

这是我的代码：

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2015, 6, 1)
}

dag = DAG('dynamic_dag_generator', schedule_interval=None, default_args=default_args)

foo_operator = BashOperator(
    task_id='foo',
    bash_command="echo '%s'" % json.dumps(range(0, random.randint(40,60))),
    xcom_push=True,
    dag=dag)

def gen_nodes(**kwargs):
    ti = kwargs['ti']
    workers = json.loads(ti.xcom_pull(task_ids='foo'))

    for wid in workers:
        print("Iterating worker %s" % wid)
        op = PythonOperator(
            task_id='test_op_%s' % wid,
            python_callable=lambda: print("Dynamic task!"),
            dag=dag
        )

        op.set_downstream(bar_operator)
        op.set_upstream(dummy_op)

gen_subdag_node_op = PythonOperator(
    task_id='gen_subdag_nodes',
    python_callable=gen_nodes,
    provide_context=True,
    dag=dag
)

gen_subdag_node_op.set_upstream(foo_operator)

dummy_op = DummyOperator(
    task_id='dummy',
    dag=dag
)

dummy_op.set_upstream(gen_subdag_node_op)

bar_operator = DummyOperator(
    task_id='bar',
    dag=dag)

bar_operator.set_upstream(dummy_op)

在日志中，我可以看到gen_nodes正确执行（即Iterating worker 5等）。但是，新任务没有被安排，也没有证据表明它们已被执行。

我在网上找到了相关的代码示例，例如 this，但无法使其工作。我错过了什么吗？

或者，是否有更合适的方法来解决这个问题（隔离工作单元）？

4

1 回答 1

5

目前，airflow 不支持在 dag 运行时添加/删除任务。

工作流顺序将是在 dag 运行开始时评估的任何内容。

请参阅此处的第二段。

这意味着您不能根据运行中发生的事情添加/删除任务。您可以基于与运行无关的内容在 for 循环中添加 X 任务，但在运行开始后，不会更改工作流形状/顺序。

很多时候，您可以在 dag 运行期间使用 aBranchPythonOperator来做出决定（这些决定可以基于您的xcom值），但它们必须是沿着工作流中已经存在的分支的决定。

Dag 运行，并且 Dag 定义以不完全直观的方式在气流中分离，但或多或​​少在 dag 运行中创建/生成的任何内容（xcom、dag_run.conf等）都不能用于定义 dag 本身。

于 2017-06-19T17:44:40.420 回答