Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

With code, how do you update an airflow variable?

Tags:

I need to update a variable I have made in Airflow programmatically but I can not find the answer on how to do that with code.

I have retrieved my variable with this code:

column_number = Variable.get('column_number')

At the end of the function, I would like to increment the column_number by one

I have tried this: Variable.set_val("column_number", int(column_number) + 1)

And it does not work.

Here is the full code for reference:

import airflow from datetime import datetime, timedelta from random import randint from airflow import DAG from airflow.hooks.postgres_hook import PostgresHook from airflow.models import Variable from airflow.operators.python_operator import PythonOperator  args = {     'owner': 'besteman',     'start_date': datetime.utcnow(),     'retries': 1,     'retry_delay': timedelta(minutes=30) }  dag = DAG(dag_id='test-postgres', default_args=args, schedule_interval='@hourly')  def add_columns_and_values():      column_number = Variable.get('column_number')      pg_hook = PostgresHook(postgres_conn_id='airflow-test')      add_columns = f'ALTER TABLE students ADD COLUMN test{column_number} smallint;'      pg_hook.run(add_columns)       for i in range(8):         add_values = f"UPDATE students SET test{column_number} = '{randint(50, 100)}' WHERE id = {i+1};"         pg_hook.run(add_values)      Variable.set_val("column_number", int(column_number) + 1)   t1 = PythonOperator(task_id='add_columns_values',     python_callable=add_columns_and_values,     dag=dag) 
like image 608
Justin Besteman Avatar asked Jan 04 '19 19:01

Justin Besteman


People also ask

How do you add a variable in Airflow?

From Airflow version 1.10. 10 you can add Airflow variables from the Terminal. To be completely clear, these are just environment variables with a specific naming convention. All Airflow variables must be set with the syntax AIRFLOW_VAR_{VARIABLE_NAME} , all uppercase.

How do you pass a variable in Airflow?

You can pass parameters from the CLI using --conf '{"key":"value"}' and then use it in the DAG file as "{{ dag_run. conf["key"] }}" in templated field.

How do you use variables in Airflow Dag?

Variables in Airflow are a generic way to store and retrieve arbitrary content or settings as a simple key-value store within Airflow. Variables can be listed, created, updated, and deleted from the UI (Admin -> Variables), code, or CLI. In addition, JSON settings files can be bulk uploaded through the UI.

How do I set an environment variable in Airflow?

As per this answer, the variables should be put in /etc/default/airflow (on Debian/Ubuntu) or /etc/sysconfig/airflow (on Centos/Redhat). Show activity on this post. If you are just running a local instance you should be able to use environment variables like you expect.


1 Answers

Use Variable.set instead of Variable.set_val. set_val() is a setter for the val attribute and not intended for outside use. This should do what you want:

Variable.set("column_number", int(column_number) + 1) 

It will make the actual update to the database, along with handling session and serialization for you if needed.

Reference: https://github.com/apache/incubator-airflow/blob/1.10.1/airflow/models.py#L4558-L4569

like image 169
Daniel Huang Avatar answered Sep 23 '22 17:09

Daniel Huang