Airflow运行DAG报psycopg2.ProgrammingError无法适配PythonOperator错误

错误原因 #

报错的核心原因是你在调用ti.xcom_pull方法时，task_ids参数传值不符合要求：

• task_ids参数需要接收的是对应任务的ID字符串，而你直接传入了Python函数对象transform_data
• Airflow执行XCom查询时需要把task_id作为条件传入元数据库，psycopg2驱动无法将PythonOperator类型的对象转换为SQL可识别的字符串，因此抛出适配错误
• 你可以从报错的参数明细直接验证：'task_id_1': <Task(PythonOperator): transform_data>，这里传入的确实是任务对象而非字符串

解决方案 #

修改create_text_file函数中调用xcom_pull的代码即可：

1. 首先确认你定义transform_data对应的PythonOperator任务时设置的task_id值，假设你定义任务的代码如下：

transform_data_task = PythonOperator(
    task_id="transform_data", # 这里的字符串就是你要填的task_id
    python_callable=transform_data,
    provide_context=True,
    dag=dag
)

2. 将原来的错误代码：

close_data = ti.xcom_pull(key=None, task_ids=[transform_data])

修改为传入task_id字符串即可：

close_data = ti.xcom_pull(key=None, task_ids=["transform_data"])

如果希望避免手动写字符串写错，也可以直接引用任务实例的task_id属性：

close_data = ti.xcom_pull(key=None, task_ids=[transform_data_task.task_id])

内容的提问来源于stack exchange，提问作者Volatil3