Воздушный поток - запускать каждую функцию питона отдельно

Question

У меня есть скрипт airflow ниже, который запускает все скрипты Python как одну функцию.Мне бы хотелось, чтобы каждая функция python выполнялась индивидуально, чтобы я мог отслеживать каждую функцию и ее состояние.

## Third party Library Imports

import psycopg2
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
#from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from sqlalchemy import create_engine
import io


# Following are defaults which can be overridden later on
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2018, 1, 23, 12),
'email': ['airflow@airflow.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}

dag = DAG('sample_dag', default_args=default_args, catchup=False, schedule_interval="@once")


#######################
## Login to DB


def db_log():
    global db_con
    try:
    db_con = psycopg2.connect(
    " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = 'port' sslmode = 'require' ")
    except:
        print("Connection Failed.")
        print('Connected successfully')
    return (db_con)

def insert_data():
    cur = db_con.cursor()
    cur.execute("""insert into tbl_1 select id,bill_no,status from tbl_2 limit 2;""")


def job_run():
    db_log()
    insert_data()



##########################################

t1 = PythonOperator(
    task_id='DB_Connect',
    python_callable=job_run,
    # bash_command='python3 ~/airflow/dags/sample.py',
    dag=dag)

t1

Вышеописанный скрипт работает просто отлично, но хотел бы разбить его по функциям, чтобы сохранитьлучше отслеживать.Может ли кто-нибудь помочь в этом.Tnx ..

Обновленный код (версия 2):

## Third party Library Imports

import psycopg2
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
#from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from sqlalchemy import create_engine
import io


# Following are defaults which can be overridden later on
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2018, 1, 23, 12),
'email': ['airflow@airflow.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}

dag = DAG('sample_dag', default_args=default_args, catchup=False, schedule_interval="@once")


#######################
## Login to DB


def db_log(**kwargs):
    global db_con
    try:
    db_con = psycopg2.connect(
    " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = 'port' sslmode = 'require' ")
    except:
        print("Connection Failed.")
        print('Connected successfully')
        task_instance = kwargs['task_instance']
        task_instance.xcom_push(value="db_con", key="db_log")
        return (db_con)

def insert_data(**kwargs):
    v1 = task_instance.xcom_pull(key="db_con", task_ids='db_log')
    return (v1)
    cur = db_con.cursor()
    cur.execute("""insert into tbl_1 select id,bill_no,status from tbl_2 limit 2;""")

#def job_run():
#    db_log()
#    insert_data()


##########################################

t1 = PythonOperator(
    task_id='Connect',
    python_callable=db_log,provide_context=True,
    dag=dag)

t2 = PythonOperator(
    task_id='Query',
    python_callable=insert_data,provide_context=True,
    dag=dag)


t1 >> t2

Answer 1

Для этого есть два возможных решения:

A) Создать несколько задач для каждой функции

Задачи в Airflow вызываются в отдельных процессах.Переменные, которые определены как global, не будут работать, поскольку вторая задача обычно не может видеть переменные первой задачи.

Представляем: XCOM.Это особенность Airflow, и мы уже ответили на несколько вопросов по этому поводу, например, здесь (с примерами): Python Airflow - результат возврата из PythonOperator

EDIT

Вы должны предоставить контекст и передать контекст вдоль , как написано в примерах.Для вашего примера это будет означать:

• добавить provide_context=True, к вашему PythonOperator
• изменить подпись job_run на def job_run(**kwargs):
• passkwargs для data_warehouse_login с data_warehouse_login(kwargs) внутри функции

B) Создать одну полную функцию

В этом самом сценарии я все равно удалил бы глобальный (просто позвоните insert_data, позвоните data_warehouse_login изнутри и верните соединение) и используйте только одну задачу.

Если произошла ошибка, выдать исключение.Воздушный поток справится с этим просто отлично.Просто убедитесь, что в исключение помещены соответствующие сообщения и используется лучший тип исключения.