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

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

0 голосов
/ 30 мая 2018

Я пытаюсь выполнить скрипт 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': ['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

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

Обновление 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': ['airflow@airflow.com']
}

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

Ответы [ 2 ]

0 голосов
/ 30 мая 2018

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

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

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 кажется мне интересным.Возможно, вы захотите продолжить эту мысль в другом вопросе.

0 голосов
/ 30 мая 2018

Это основное сообщение об ошибке от 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;""")
...