根据气流中的上游任务杀死下游任务

Question

我有一个 DAG，其中有 2 个任务。

t1=  PythonOperator(
    task_id='Check_Files_in_S3',
    provide_context=False,
    python_callable=checkFilesInS3,
    xcom_push=True,
    dag=dag)

t2 =  PythonOperator(
    task_id='snowflakeLoad',
    provide_context=True,
    python_callable=snowflakeLoad,
    xcom_push=True,
    dag=dag)

t1>>t2

第一个任务在 S3 中查找文件，如果文件可用，则任务应该是成功的，并且应该使用下游任务将数据加载到雪花中。

我的要求是，如果文件不可用，那么 task1 需要发送邮件，那么它应该停止下游任务（task2）。有没有办法实现这一点，比如返回 false 会使下游任务失败？如果早期的 DAG 运行失败，depends_on_past 也不会运行 DAG？或者如果其中一项任务在当前运行中失败，它不会运行下游任务？

Answer 1

你有很多问题，所以让我分解一下。

有没有办法实现这一点，比如返回 false 会使下游任务失败？

等待某事发生的通常方法是使用传感器（继承自 BaseSensorOperator）类。它们专门用于低资源消耗和易于配置以感知外部对象/资源。 但是，在您的情况下，这是一个检查并忘记（如果它不存在，请不要重试）。然后你可以坚持使用 PythonOperator。您只需引发一个异常，它就会被视为任务失败，从而阻止下游任务（在您的情况下为 t2）运行。

如果之前的 DAG 运行失败，depends_on_past 也不会运行 DAG？或者如果其中一项任务在当前运行中失败，它不会运行下游任务？

如果我们有两个任务，假设我们有两个 DAGRun。一个在时间点x，一个在时间点x + 1。如果您有depends_on_past=True，则x + 1 的任务t1 的运行将在x 时查看运行t1，并且只有在运行成功时才会开始。 这同样适用于t2 的x + 1，它将检查x + 1 的任务t1 是否已完成，然后检查t2 的时间x 是否成功。

Answer 2

当没有文件时，您可以在checkFilesInS3 中提出AirflowSkipException。这将导致下游任务在引发异常之前也被跳过，您也可以发送电子邮件。

from airflow import DAG
from datetime import datetime
from airflow.exceptions import AirflowSkipException
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.utils.email import send_email


def checkFilesInS3(**kwargs):
    your_code
    if no_files: #Raise exception to make downstream tasks skip
        send_email(to="someone@somewhere.com, subject="task failed!"")
        raise AirflowSkipException("No files found!")


def snowflakeLoad(**kwargs):
    pass


default_args = {
    'owner': 'airflow',
    'start_date': datetime(2021, 6, 23),

}
with DAG(dag_id='question',
         default_args=default_args,
         schedule_interval=None,
         catchup=False
         ) as dag:

    start = DummyOperator(task_id='start_task')

    t1 = PythonOperator(
        task_id='Check_Files_in_S3',
        python_callable=checkFilesInS3,
    )

    t2 = PythonOperator(
        task_id='snowflakeLoad',
        python_callable=snowflakeLoad,
    )

    start >> t1 >> t2