Airflow 2.0 - running locally keeps running the function

user_14444816 0
Viewed 128

I have the below task keeps running I know this because it runs a query in Snowflake and I keep getting the DUO push notification. every. 5. seconds! What can I do to stop this and only have it run when the DAG runs

This is the task:

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
)

This is the method being called in the sql part:

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-2.x airflow
1 Answers

The sql=SnowHook().run_fk_alter_statements(schema,query) call in your instantiation of the SnowflakeQueryOperator is actually top-level code so it will execute every time the DAG is parsed by the Scheduler. You need to find a way to have that function called within an operator's execute() method.

You could add a TaskFlow function/PythonOperator task to push the output from run_fk_alter_statements() to XCom and then the SnowflakeQueryOperator uses this XCom to execute the SQL(s) that's generated.

user_157723770
Related