如何在气流中呈现带有.sql参数的MySqlOperator文件？

Question

在将参数(xcom从以前的任务推送)传递到.sql文件中的SQL查询时，我需要帮助。但是，我无法使用“参数”选项这样做，尽管该选项能够呈现以前任务中的xcom值。让我知道我做错了什么。

谢谢:)

start = EmptyOperator(
            task_id="start",
    )

fetch_cust_id = PythonOperator(
    task_id = "fetch",
    python_callable = lambda: 'C001',
)

update_orders = MySqlOperator(
    task_id="update",
    mysql_conn_id="mysql_default",
    database="my_db",
    sql="/update.sql",
    parameters={
        "custid": "{{ ti.xcom_pull(task_ids='fetch') }}"
    }
)

start >> fetch_cust_id >> update_orders

SQL文件(update.sql)：

UPDATE orders
SET placed = 'yes'
WHERE
custid = {{ custid }}
;

:(

​

​

Answer 1

parameters用于将“变量”传递给SqlAlchemy引擎。在这种情况下，渲染不是在气流引擎中进行的。如果您想要使用这个，您需要使用SqlAlchemy语法。示例：

sql="SELECT * FROM pet WHERE birth_date BETWEEN SYMMETRIC %(begin_date)s AND %(end_date)s",
parameters={"begin_date": "2020-01-01", "end_date": "2020-12-31"},

但是在您的例子中，您想要模板xcom，所以根本没有理由使用parameters。你希望渲染是通过气流完成的。

您可以直接在sql中设置它，因为sql是一个模板字段：

UPDATE orders
SET placed = 'yes'
WHERE custid = "{{ ti.xcom_pull(task_ids='fetch') }}";