如何为依赖于另一个任务的任务设置相等的priority_weight答案

【问题标题】：How to set equal priority_weight to task that depends on another task如何为依赖于另一个任务的任务设置相等的priority_weight
【发布时间】：2019-05-21 07:15:12
【问题描述】：

我有 8 组任务。每组都是一系列任务：task1 >> task2 >> task3。 task3 依赖于 task2，因此 task2 依赖于 task1。

我的问题是 task2 在所有 task1 完成之前永远不会启动。所以为了让 set1.task2 启动它必须先运行 set8.task1。

我最初的研究是关于 priority_weight 的东西，它可以包含在 DAG 的 default_args 中。我了解到 task1 对其下游的priority_weight 会更高。

有没有一种方法可以使所有优先级权重都相同。这样 set1.task2 就已经可以启动而不管 set2、3 等，因为它只依赖于 set1.task1。

谢谢！

【问题讨论】：

标签： airflow airflow-scheduler

【解决方案1】：

将weight_rule 设置为“上游”或“绝对”应该会有所帮助。这是来自BaseOperator 文档字符串：

:param weight_rule: weighting method used for the effective total
    priority weight of the task. Options are:
    ``{ downstream | upstream | absolute }`` default is ``downstream``
    When set to ``downstream`` the effective weight of the task is the
    aggregate sum of all downstream descendants. As a result, upstream
    tasks will have higher weight and will be scheduled more aggressively
    when using positive weight values. This is useful when you have
    multiple dag run instances and desire to have all upstream tasks to
    complete for all runs before each dag can continue processing
    downstream tasks. When set to ``upstream`` the effective weight is the
    aggregate sum of all upstream ancestors. This is the opposite where
    downtream tasks have higher weight and will be scheduled more
    aggressively when using positive weight values. This is useful when you
    have multiple dag run instances and prefer to have each dag complete
    before starting upstream tasks of other dags.  When set to
    ``absolute``, the effective weight is the exact ``priority_weight``
    specified without additional weighting. You may want to do this when
    you know exactly what priority weight each task should have.
    Additionally, when set to ``absolute``, there is bonus effect of
    significantly speeding up the task creation process as for very large
    DAGS. Options can be set as string or using the constants defined in
    the static class ``airflow.utils.WeightRule``

链接：https://github.com/apache/airflow/blob/master/airflow/models/baseoperator.py#L129-L150

【讨论】：

你把这个论点放在哪里？在 DAG 的默认参数中？
谢谢！我将它设置在每个任务中。我不确定这是否可以成为 DAG 中的默认参数。
我相信可以，试试看。

【解决方案2】：

为 DAG 添加任何参数将应用于该 DAG 下定义的所有任务。您可以在实例化 DAG 时在 default_args 中传递 weight_rule。

例如：

with DAG(
    "dag_1",
    schedule_interval="@daily", 
    catchup=False,
    start_date=datetime(2021, 9, 10),
    default_args={
        "priority_weight": 5,
        "pool": "testing_pool",
        "weight_rule": "absolute",
    },
) as dag:

【讨论】：