CODEWITHSUNDEEP
pythonvariablesoperatorsairflowdirected-acyclic-graphs
Rohan0980
Rohan0980

Reputation: 47

Airflow variables getting updated even if the DAG is not running

I am reading an integer variable from airflow variables and then incrementing the value by one each time the DAG runs and set it to the variable again.

But after the below code the variable at UI changes each time page is refreshed or so. Idk what is causing such behavior

counter = Variable.get('counter')
s = BashOperator(
    task_id='echo_start_variable',
    bash_command='echo ' + counter,
    dag=dag,
)
Variable.set("counter", int(counter) + 1)

sql_query = "SELECT * FROM UNNEST(SEQUENCE({start}, {end}))"
sql_query = sql_query.replace('{start}', start).replace('{end}', end)
submit_query = PythonOperator(
    task_id='submit_athena_query',
    python_callable=run_athena_query,
    op_kwargs={'query': sql_query, 'db': 'db',
               's3_output': 's3://s3-path/rohan/date=' + current_date + '/'},
    dag=dag)

e = BashOperator(
    task_id='echo_end_variable',
    bash_command='echo ' + counter,
    dag=dag,
)

s >> submit_query >> e

Upvotes: 2

Views: 2567

Answers (1)

Elad Kalif
Elad Kalif

Reputation: 15931

Airflow process that DAG file every 30 seconds (default of min_file_process_interval setting) this means that any top level code you have is running every 30 seconds so Variable.set("counter", int(counter) + 1) will cause the Variable counter to be increased by 1 every 30 seconds.

It's a bad practice to interact with Variables in top level code (regardless of the increasing value issue). It opens a connection to the metastore database every 30 seconds which may cause serious problems and overwhelm the database.

To get the value of Variable you can use Jinja:

e = BashOperator(
    task_id='echo_end_variable',
    bash_command='echo {{ var.value.counter }}',
    dag=dag,
)

This is a safe way to use variables as the value is being retrieved only when the operator is executed.

If you want to increase the value of the variable by 1 then do it with PythonOpeartor:

def increase():
    counter = Variable.get('counter')
    Variable.set("counter", int(counter) + 1)

increase_op = PythonOperator(
    task_id='increase_task',
    python_callable=increase,
    dag=dag)

The python callable will be executed only when the operator runs.

Upvotes: 6

Related Questions