我正在尝试运行一个登录 Amazon Redshift DB 的 python 脚本,然后执行 SQL 命令。我使用名为 Airflow 的工具进行工作流程管理。运行以下代码时,我可以正常登录数据库,但在尝试执行 SQL 命令时出现以下错误。
**AttributeError: 'NoneType' object has no attribute 'execute'**
代码:
## 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)
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 ;""")
任何人都可以帮我解决这个问题吗?谢谢..
完整代码:
## 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': ['[email protected]']
}
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_callable 中调用
data_warehouse_login
函数,但该函数未定义。假设这意味着 db_log
并且第一个任务成功,那么您实际上并没有对第二个任务执行任何操作(因为您的 xcom_push
仅在错误时触发)。
一般情况下不建议 xcom-ing 连接对象。或者,您可能需要考虑使用随附的 PostgresHook,它应该涵盖您的使用案例,并且与 Amazon Redshift 配合使用效果同样良好。
https://github.com/apache/incubator-airflow/blob/master/airflow/hooks/postgres_hook.py