airflow - Airflow - 尝试执行一组 Python 函数
问题描述
我正在尝试执行一个由几个 Python 函数组成的 Airflow 脚本。这些功能基本上是查询数据库并执行一些任务。我正在尝试执行这是 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
解决方案
这是来自 Python 的基本错误消息。
NameError: name 'task_instance' is not defined
告诉task_instance
你想使用它时无处可寻。
任务实例在已经传递给函数的上下文中提供。
气流发送带有设置的上下文
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;""")
推荐阅读
- scheme - 球拍:为什么不适用于“位图”功能?
- node.js - nginx 服务器将自定义端口上的请求转发到节点应用程序
- angular - jwt token 过期时间(asp.net core)
- node.js - Semver 库给了我一个无效的 0.0.001 版本
- django - 在 ListView 中拖放元素 - Django
- html - Mac 与 Windows 上的 Chrome 的 rem 计算在字体大小和行高之间的差异?
- c# - 从datagridview c#加载图像
- go - golang中值类型/指针类型的区别
- php - 我无法用 php 更新我的数据库
- javascript - 使用新的浏览器窗口来控制/显示父窗口上的元素