气流 - 在回填任务之前清理数据库

Question

在对Aiflow DAG连接的任务进行回填之前，我需要清理数据库（Postgres/Mongo）。让我们假设这个简单的dag：

     ->task2
task1       ->task4
     ->task3

每个任务都会按照正常执行的数据库每天都可以将其结果

每天都可以。

但是，当涉及到此DAG的回填时，过去几天已经处理过，但是我们需要回填一次一次计算结果（即更改了任务2的代码，我们需要刷新）已经有数据与过去几天相应的数据库，在回填过程中将提出IE重复。因此，现在我需要删除与需要在执行回填之前需要回填的任务相对应的手动数据。

我要：对于每个任务，将制作脚本，该脚本将进行当前执行日期的数据库中的相应数据，并在运行实际任务之前运行此脚本，该任务将填充数据库中的数据再次。但是出于绩效原因，我希望此删除脚本仅在处理回填并且不在每日执行dag时运行它才能运行。

我缺少的是如何判断任务是由回填还是正常的每日执行（某些标志等）来运行，或者也许可以选择运行 pre_task_execute_execute_when_backfill_backfill_callable_callable 函数>在气流任务中分配某个地方？

Answer 1

目前，我以下面的方式结束了两种方法：

def is_task_backfill(day: datetime.date) -> bool:
    """check if task is running on new day data or 
    is it running in backfill mode"""
    return day < datetime.date.today() - datetime.timedelta(1)

def delete_day_data_when_backfill(day, session, func, **kwargs):
    """clean up data from postgres/mongo table that corresponds with day parameter.
    Main purpose is to run this method before backfill
    
    :param day: date for deletion of corresponding data for current task
    :type day: datetime.date 
    """
    if is_task_backfill(day):
        func(day, session, **kwargs)

我正在呼叫 delete_day_data_data_when_backfill（）在我想执行的每个任务的开始时，都使用相应的功能将数据库清除为参数。因此DAG任务模块可以看起来像：

def delete_day_data(Table, day, session):
        session.query(
            Table,
        ).filter(
            Table.date == day,
        ).delete()

def fill_database_with_day_data(day):
   with Session(engine) as session:
      delete_day_data_when_backfill(day, session, delete_day_data)
      function_that_fills_with_day_data()

总结：
我缺少判断气流是否正在执行或气流正在执行正常的日常任务。因此，回答问题是此方法：

def is_task_backfill(day: datetime.date) -> bool:
    """check if task is running on new day data or 
    is it running in backfill mode"""
    return day < datetime.date.today() - datetime.timedelta(1)

如果执行 day 小于 .today（）-TimeDelta（1），它将是回填执行在我的情况下（每天执行任务时）。非常简单的条件，我在气流文档中没有找到任何标志，但是我没有搜索很长时间。

希望这对某人有帮助:)