我需要以编程方式更新在Airflow中创建的变量,但是找不到如何使用代码来解决的答案。
我已经使用以下代码检索了变量:
column_number = Variable.get('column_number')
在该函数的结尾,我想将column_number加1
我已经试过了:
Variable.set_val("column_number", int(column_number) + 1)
而且它不起作用。
这是完整的代码供参考:
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 …Run Code Online (Sandbox Code Playgroud) airflow ×1