使用代码,如何更新气流变量? [英] With code, how do you update an airflow variable?
问题描述
我需要以编程方式更新在Airflow中创建的变量,但找不到如何使用代码来解决问题的答案。
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')
在函数末尾,我想将column_number加1
At the end of the function, I would like to increment the column_number by one
我已经尝试过:
Variable.set_val( column_number,int(column_number)+ 1)
它不起作用。
以下是完整的参考代码:
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)
推荐答案
使用 Variable.set
代替 Variable.set_val
。 set_val()
是 val
属性的设置器,不适用于外部使用。这应该做您想要的:
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.
参考: https://github.com/apache/incubator-airflow/blob/1.10.1/airflow/models.py#L4558-L4569
这篇关于使用代码,如何更新气流变量?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!