Airflow

Question

我正在尝试执行由几个 Python 函数组成的 Airflow 脚本。这些函数基本上查询数据库并执行一些任务。我正在尝试执行 this is 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

Answer 1

这是来自 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;""")

Answer 2

既然问题变大了，我觉得补充第二个答案比较合适。

即使根据评论 "I removed the indentation portion of the code" 进行了编辑，我仍然不确定这段代码：

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' object has no attribute 'execute') 中使用 PostgresHook 的想法对我来说似乎很有趣。你可能想在另一个问题中继续这个想法。

Airflow - 尝试执行一组 Python 函数

Airflow - Trying to execute a set of Python functions

airflow-scheduler