PostgresHook 中的 Airflow conf 参数抛出错误

如何解决PostgresHook 中的 Airflow conf 参数抛出错误

脚本从气流 conf {"flag":"NA","metric_name":"RED"} 中获取输入参数。我正在尝试在 sql 的 where 子句中使用一个参数值。有人可以检查我通过 where caluse 的方式是否有任何问题吗？

错误：-“{”处或附近的语法错误

def get_metrics(**kwargs):
    varteam_flag=(kwargs['dag_run'].conf['flag'])
    print("flag :",varteam_flag)
    params={"param1": varteam_flag}
    print(" params :",params['param1'])
    conn_id = kwargs.get('conn_id')
    pg_hook = PostgresHook(conn_id)
    sql="select count(1) as cnt FROM odw.metrics where  flag = {{ params.param1 }} "
    print(sql)
    records = pg_hook.get_records(sql)
    print("Records count is  :",str(records))
    return records

getData_metrics=Pythonoperator(task_id='getData_metrics',python_callable=get_metrics,op_kwargs {'conn_id':'veas'},provide_context=True,dag=dag)

解决方法

您不能像您那样在可调用的 python 中使用 Jinja。 调用运算符时正在评估 Jinja，因此在您的情况下，jinja 引擎不适用于 sql 语句。

你需要做的只是使用一个简单的赋值：

sql=f"select count(1) as cnt FROM odw.metrics where  flag = {params.param1} "

我不太清楚你为什么要做这么多作业 我相信

sql=f"select count(1) as cnt FROM odw.metrics where flag = {varteam_flag} "

应该提供相同的sql语句。

,

如何在 UI 中传递命令行参数/conf 参数并在 PostgresHook/PostgresOperator 中使用它[简单的 python 变量在这里可以使用 f 或 .format()]。

1. If you are using PostgresHook and calling PythonOperator for the module
def get_metrics(**kwargs):
    varteam_flag=(kwargs['dag_run'].conf['flag'])
    conn_id = kwargs.get('conn_id')
    pg_hook = PostgresHook(conn_id)
    connection = pg_hook.get_conn()
    sql=f"select count(1) as cnt FROM odw.metrics where  flag = {varteam_flag} "
    print(sql)
    cursor = connection.cursor()
    cursor.execute(sql)
    rows = cursor.fetchall()
    cols = list(map(lambda x: x[0],cursor.description))

2. if you are using PostgresOperator [you can use jinja template]

gerics=PostgresOperator(task_id='gerics',postgres_conn_id='veritas',params={'param1': 10 },sql=['truncate table test.dept1','insert into test.dept1 (select * from test.dept where rating={{ params.param1 }} )'],

如果有其他写作方式，请告诉我。