当下游任务定义取决于上游结果时，如何设置DAG

Question

我的问题是关于一个DAG，它基于计算MySQL表中被上游任务删除和重建的行数，动态定义一个group of parallel tasks。我遇到的困难是，在我的上游任务中，我TRUNCATE在重新构建它之前先清除它。这是sherlock_join_and_export_task。当我这样做时，行数降到零，并且我的动态生成的任务将停止定义。当表被恢复时，图的结构也一样，但任务不再执行。相反，它们在树视图中显示为黑框：

下面是sherlock_join_and_export_task删除count = worker.count_online_table()行中引用的表之后的DAG：

在sherlock_join_and_export_task完成后，DAG如下所示：

不过，这些任务都没有排队和执行。达格一直在跑，什么也没发生。在

在这种情况下，我会使用子DAG吗？关于如何设置这个，或者重写现有的DAG有什么见解吗？我用一个LocalExecutor在AWS服务器上运行这个。以下代码供参考：

from datetime import datetime
import os
import sys

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator

BATCH_SIZE = 75000

from preprocessing.marketing.minimalist.table_builder import OnlineOfflinePreprocess

worker = OnlineOfflinePreprocess()

def partial_process_flow(batch_size, offset):
    worker = OnlineOfflinePreprocess()
    worker.import_offline_data()
    worker.import_online_data(batch_size, offset)
    worker.merge_aurum_to_sherlock()
    worker.upload_table('aurum_to_sherlock')

def batch_worker(batch_size, offset, DAG):
    return PythonOperator(
        task_id="{0}_{1}".format(offset, batch_size),
        python_callable=partial_process_flow,
        op_args=[batch_size, offset],
        dag=DAG)

DAG = DAG(
  dag_id='minimalist_data_preproc',
  start_date=datetime(2018, 1, 7, 2, 0, 0, 0), #..EC2 time. Equal to 11pm hora México
  max_active_runs=1,
  concurrency=4,
  schedule_interval='0 9 * * *', #..4am hora mexico
  catchup=False
)

clear_table_task = PythonOperator(
    task_id='clear_table_task',
    python_callable=worker.clear_marketing_table,
    op_args=['aurum_to_sherlock'],
    dag=DAG
)

sherlock_join_and_export_task = PythonOperator(
    task_id='sherlock_join_and_export_task',
    python_callable=worker.join_online_and_send_to_galileo,
    dag=DAG
)

sherlock_join_and_export_task >> clear_table_task

count = worker.count_online_table()
if count == 0:
    sherlock_join_and_export_task >> batch_worker(-99, -99, DAG) #..dummy task for when left join deleted
else:
    format_table_task = PythonOperator(
        task_id='format_table_task',
        python_callable=worker.format_final_table,
        dag=DAG
    )

    build_attributions_task = PythonOperator(
        task_id='build_attributions_task',
        python_callable=worker.build_attribution_weightings,
        dag=DAG
    )

    update_attributions_task = PythonOperator(
        task_id='update_attributions_task',
        python_callable=worker.update_attributions,
        dag=DAG
    )

    first_task = batch_worker(BATCH_SIZE, 0, DAG)
    clear_table_task >> first_task
    for offset in range(BATCH_SIZE, count, BATCH_SIZE):
        first_task >> batch_worker(BATCH_SIZE, offset, DAG) >> format_table_task

    format_table_task >> build_attributions_task >> update_attributions_task

下面是一个简化的DAG工作原理：

^{pr2}$

Answer 1

看看你的dag，我想你已经实现了一个非幂等流程，而气流并不是真正配置的。与其截短/更新您正在构建的表，您可能应该保留对任务的配置并只更新开始日期/结束日期，以便在任务级别启用和禁用这些任务以进行调度，或者甚至在每次迭代时运行所有这些任务，并在脚本中检查该表，以便在作业被禁用时仅运行hello world。在

Answer 2

我和这个用例抗争了很长时间。简言之，基于不断变化的资源状态构建的dag，尤其是db表，在气流中飞行得并不好。在

我的解决方案是编写一个小的自定义操作符，它是truggerdageoperator的子类，它执行查询，然后为每个子进程触发dagruns。在

它使流程“join”下游更有趣，但在我的用例中，我可以使用另一个dag来解决它，如果给定一天的所有子流程都已完成，则该dag会进行轮询和短路。在其他情况下，分区传感器可以做到这一点。在

我有几个这样的用例（基于动态源的迭代dag触发器），在与使动态子dag工作（很多）进行了大量的斗争之后，我切换到了这个“触发器子流程”策略，并且从那以后一直做得很好。在

注意-这可能会为一个目标（目标）生成大量的dagruns。这使得UI在某些地方具有挑战性，但它是可行的（我已经开始直接查询数据库，因为我还没有准备好编写一个插件来填充UI）