根据气流中 sql 查询的结果创建动态任务

Question

我正在尝试使用 TaskGroup 创建动态任务，并将结果保存在变量中。该变量每 N 分钟修改一次，具体取决于数据库查询，但是当第二次修改该变量时，调度程序出现故障

基本上我需要根据查询中收到的唯一行数来创建任务。

以 TaskGroup(f"task") 作为任务：

    data_variable = Variable.get("df")
    data = data_variable

    try :
        if data != False and data !='none':
            df = pd.read_json(data)

            for field_id in df.field.unique():
             

                task1 = PythonOperator(
                   
                )
                task2 = PythonOperator(
                   
                )

               
                task1 >> task2

    except:
        pass

有没有办法通过任务组来做到这一点？

Answer 1

这是 Airflow 的反模式。

虽然您可以在顶级代码中使用Variable.get("df")，但您不应该这样做。使用任何数据库创建查询的变量/连接/任何其他代码只能在操作员范围内或使用 Jinja 模板完成。原因是 Airflow 会定期解析 DAG 文件（如果您没有更改 min_file_process_interval 的默认值，则每 30 秒解析一次），因此每 30 秒与数据库交互的代码将导致该数据库的负载过重。 对于其中一些情况，未来的气流版本会发出警告（请参阅PR）

气流任务应尽可能静态（或缓慢变化）。