С кодом, как вы обновите переменную воздушного потока? - PullRequest
Новая
       2

С кодом, как вы обновите переменную воздушного потока?

0 голосов
/ 04 января 2019

Мне нужно программно обновить переменную, которую я создал в Airflow, но я не могу найти ответ, как это сделать с помощью кода.

Я получил свою переменную с этим кодом:

column_number = Variable.get('column_number')

В конце функции я хотел бы увеличить значение column_number на единицу

Я пробовал это: 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 = 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)

1 Ответ

0 голосов
/ 05 января 2019

Используйте Variable.set вместо Variable.set_val. set_val() является установщиком атрибута val и не предназначен для внешнего использования. Это должно делать то, что вы хотите: 

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

Будет выполнено актуальное обновление базы данных, а также, при необходимости, будет обрабатываться сеанс и сериализация для вас.

Ссылка: https://github.com/apache/incubator-airflow/blob/1.10.1/airflow/models.py#L4558-L4569

...