Airflow 2.0 - 在本地运行会一直运行该功能答案

【问题标题】：Airflow 2.0 - running locally keeps running the functionAirflow 2.0 - 在本地运行会一直运行该功能
【发布时间】：2021-12-28 00:22:24
【问题描述】：

我有以下任务一直在运行我知道这一点，因为它在 Snowflake 中运行查询并且我不断收到 DUO 推送通知。每一个。 5秒！我能做些什么来阻止它并且只在 DAG 运行时运行它

这是任务：

create_foreign_keys = SnowflakeQueryOperator(
    dag=dag,
    task_id='check_and_run_foreign_key_query',
    sql=SnowHook().run_fk_alter_statements(schema,query),
    trigger_rule=TriggerRule.ALL_DONE
)

这是在sql部分调用的方法：

def run_fk_alter_statements(self, schema, additional_fk):

    fk_query_path = "/fkeys.sql"

    fd = open(f'{fk_query_path}', 'r')
    query = fd.read()
    fd.close()

    additions = []


    for fk in additional_fk:
            additions.append(f""" or (t2.table_name = '{fk['table_name']}' and t2.column_name = '{fk['column_name']}'
                            and t1.table_name = '{fk['ref_table_name']}' and t1.column_name = '{fk['ref_column_name']}')\n""".upper())


    raw_out = self.execute_query(query.format(schema=schema, fks=''.join(additions)), fetch_all=True)

    query_jobs = []
    for raw_query in raw_out:
        query_jobs.append(raw_query[0])


    return query_jobs

【问题讨论】：

标签： airflow airflow-2.x

【解决方案1】：

SnowflakeQueryOperator 实例化中的 sql=SnowHook().run_fk_alter_statements(schema,query) 调用实际上是顶级代码，因此每次调度程序解析 DAG 时都会执行它。您需要找到一种方法在操作员的 execute() 方法中调用该函数。

您可以添加一个 TaskFlow 函数/PythonOperator 任务将输出从 run_fk_alter_statements() 推送到 XCom，然后 SnowflakeQueryOperator 使用此 XCom 执行生成的 SQL。

【讨论】：