Airflow - Попытка выполнить набор функций Python

Question

Я пытаюсь выполнить скрипт Airflow, который состоит из нескольких функций Python. Эти функции в основном запрашивают базу данных и выполняют несколько задач. Я пытаюсь выполнить это Airflow, чтобы я мог контролировать каждую из этих функций по отдельности. Ниже приведен код, который я пытаюсь выполнить, и получите приведенную ниже ошибку

Подзадача: NameError: имя 'task_instance' не определено

## Third party Library Imports

import psycopg2
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
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': ['[e-mail]'],
'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

Может ли кто-нибудь помочь в этом. Спасибо..

Обновление 1:

Обнаружена ошибка

AttributeError: 'NoneType' object has no attribute 'execute'

указывая на последнюю строку в приведенном выше фрагменте кода

cur.execute("""insert into tbl_1 select id,bill_no,status from tbl_2 limit 2;""")

Полный код

Полный код:

## Third party Library Imports
import pandas as pd
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, 5, 29, 12),
'email': ['[e-mail]']
}

dag = DAG('sample1', default_args=default_args)

## Login to DB

def db_log(**kwargs):
  global db_con
  try:
    db_con = psycopg2.connect(
       " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = '5439'")
  except:
    print("I am unable to connect")
    print('Connection Task Complete')
    task_instance = kwargs['task_instance']
    task_instance.xcom_push(key="dwh_connection" , value = "dwh_connection")
    return (dwh_connection)



t1 = PythonOperator(
  task_id='DWH_Connect',
  python_callable=data_warehouse_login,provide_context=True,
  dag=dag)

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

def insert_data(**kwargs):
  task_instance = kwargs['task_instance']
  db_con_xcom = task_instance.xcom_pull(key="dwh_connection", task_ids='DWH_Connect')
  cur = db_con_xcom
  cur.execute("""insert into tbl_1 select limit 2 """)


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

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

t1 >> t2

Answer 1

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

Даже после редактирования из комментария "Я удалил часть кода с отступом", я все еще не уверен в этом фрагменте кода:

def db_log(**kwargs):
  global db_con
  try:
    db_con = psycopg2.connect(
       " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = '5439'")
  except:
    print("I am unable to connect")
    print('Connection Task Complete')
    task_instance = kwargs['task_instance']
    task_instance.xcom_push(key="dwh_connection" , value = "dwh_connection")
    return (dwh_connection)

Это должно выглядеть так:

def db_log(**kwargs):
  global db_con
  try:
    db_con = psycopg2.connect(
       " dbname = 'name' user = 'user' password = 'pass' host = 'host' port = '5439'")
  except:
    print("I am unable to connect")

  print('Connection Task Complete')
  task_instance = kwargs['task_instance']
  task_instance.xcom_push(key="dwh_connection" , value = "dwh_connection")
  #return (dwh_connection)  # don't need a return here

Кроме того, идея вашего другого вопроса ( Python - AttributeError: объект 'NoneType' не имеет атрибута 'execute') использовать PostgresHook кажется интересным для меня. Возможно, вы захотите продолжить эту мысль в другом вопросе.

Answer 2

Это основное сообщение об ошибке от Python.

NameError: name 'task_instance' is not defined

говорит, что task_instance нигде не найти, когда вы хотите его использовать.

Экземпляр задачи предоставляется в контексте, который уже передается в функцию.

Airflow отправляет контекст с настройкой

provide_context=True,

в рамках задачи. Также определение принимает kwargs:

def insert_data(**kwargs):

что тоже правильно.

коррекция

Сначала вам нужно вывести экземпляр задачи из контекста следующим образом:

task_instance = kwargs['task_instance']

Затем вы можете использовать экземпляр задачи, чтобы использовать xcom_pull, Таким образом, это должно выглядеть так (добавьте также несколько комментариев):

def insert_data(**kwargs):
    task_instance = kwargs['task_instance']
    db_con_xcom = task_instance.xcom_pull(key="db_con", task_ids='db_log')
    #return (v1)  # wrong, why return here?
    #cur = db_con.cursor()  # wrong, db_con might not be available
    cur = db_con_xcom
    cur.execute("""insert into tbl_1 select id,bill_no,status from tbl_2 limit 2;""")