Airflow 在排程任务时，会根据 DAG 和任务的状态生成 TaskInstance 实例，并使用数据库维护任务的状态。如果在执行任务时出现任何异常情况，任务实例就会被标记为“FAILED”。但是，如果在标记为“FAILED”之前数据库中出现了一些问题，可能会在 DAG 上创建重复的任务实例。

为了解决这个问题，可以使用DAG中的provide_context参数，提供 context 里面的dag_run_id，重新获取并重写 dag_run_id，确保推入任务实例的时候会插入唯一的 dag_run_id。

代码示例：

from airflow import DAG
from airflow.models import DagRun


default_args = {
    'start_date': datetime(2022, 1, 1),
}

dag = DAG(
    dag_id='my_dag_id',
    default_args=default_args,
    catchup=False,
)
def replace_dag_run_id(**kwargs):
    dag_run = DagRun.find(dag_id='my_dag_id')[0]
    dag_run.run_id = 'Replaced-dag-run-id'
    dag_run.verify_integrity()
    dag_run.update_state()

    if dag_run.external_trigger:
        dag_run.state = State.RUNNING
    session.commit()


my_task = PythonOperator(
    task_id='my_task_id',
    python_callable=my_function,
    provide_context=True,
    op_kwargs={'param1': 'value1', 'param2': 'value2'},
    dag=dag,
)

t.set_upstream(my_task)