在Airflow v1中，如果一个任务被跳过了，那么它的子任务也会被跳过，并且被标记为“skipped”。然而，在Airflow v2中，如果一个任务被跳过了，它的子任务将继续运行，并且不会被标记为“skipped”。

这些行为的更改可能会影响一些Airflow v1升级到v2的用户。如果任何任务的依赖被跳过，则需要添加一个判断来检查此任务是否应该被执行。可以使用以下代码示例进行操作：

from airflow.operators.python_operator import PythonOperator

def my_func(**kwargs): print("Running task...")

def check_skip(**kwargs): ti = kwargs['ti'] if ti.is_skip_trigger(): print("Skipping task...") ti.xcom_push(key='skip', value=True) else: print("Running task...")

Airflow v1

task = PythonOperator( task_id='my_task', python_callable=my_func, dag=dag )

Airflow v2

task = PythonOperator( task_id='my_task', python_callable=check_skip, dag=dag )

Add downstream task

downstream_task = PythonOperator( task_id='my_downstream_task', python_callable=my_func, dag=dag )

Airflow v1

task.set_downstream(downstream_task)

Airflow v2

skip_downstream = BranchPythonOperator( task_id='skip_downstream', python_callable=lambda: "{{ ti.xcom_pull(key='skip', default=False) }}", dag=dag ) task >> [skip_downstream, downstream_task]