如何在气流中从文件执行SQL查询？(PostgresSQL算子)

Question

我使用的是PostgresSQL操作符。任务如下：

emailage_transformations = PostgresOperator(
    task_id = 'emailage_transformations',
    sql = '/home/ubuntu/airflow_ci/current/scripts/antifraud/emailage_transformations.sql',
    postgres_conn_id = 'redshift',
    autocommit = True,
    dag = dag)

首先，文件的内容是下一个：

select cd_pedido_nr,fraud_score,risk_band,ip_risk_level
into antifraud.stg_emailage_id_pedido
from antifraud.stg_emailage_id_email e
left join antifraud.info_emails i on id_email = cd_email_nr
;

我所犯的错误是

jinja2.exceptions.TemplateNotFound: /home/ubuntu/airflow_ci/current/scripts/antifraud/emailage_transformations.sql

因此，我在查询中添加了几个方括号，以满足jinja2模板的要求，现在文件代码是：

{select cd_pedido_nr,fraud_score,risk_band,ip_risk_level
into antifraud.stg_emailage_id_pedido
from antifraud.stg_emailage_id_email e
left join antifraud.info_emails i on id_email = cd_email_nr
;}

然而，我仍然有同样的错误。我怎么能解决呢？

Answer 1

我在下面的链接中指出，您应该向DAG提供一个template_searchpath，以便它能够选择您的外部文件(SQL或其他文件)。

• External files in Airflow DAG
• How to pass SQL as file with parameters to Airflow Operator
• searchpath gives TemplateNotFound error in Airflow and cannot find the SQL script

或者，使外部文件可被发现，例如AIRFLOW_HOME或通过other tricks也可以工作。