向MySqlOperator传递和使用参数的方法如下:
from airflow.providers.mysql.operators.mysql import MySqlOperator
# 创建MySqlOperator对象并传递参数
task = MySqlOperator(
task_id='mysql_task',
sql='SELECT * FROM table WHERE column = %(param)s',
parameters={'param': 'value'},
mysql_conn_id='mysql_conn'
)
在上述示例中,通过parameters
参数将param
参数传递给了MySqlOperator对象。在SQL语句中可以使用%(param)s
的形式引用该参数。
xcom_push
方法将参数传递给XCom,然后在后续任务中使用xcom_pull
方法获取参数。例如:from airflow.providers.mysql.operators.mysql import MySqlOperator
# 前一个任务中传递参数
task1 = PythonOperator(
task_id='task1',
python_callable=push_param,
op_kwargs={'param': 'value'},
provide_context=True
)
# 后续任务中获取参数
task2 = MySqlOperator(
task_id='mysql_task',
sql='SELECT * FROM table WHERE column = %(param)s',
mysql_conn_id='mysql_conn',
params={'param': '{{ task_instance.xcom_pull(task_ids="task1", key="param") }}'}
)
在上述示例中,通过op_kwargs
参数将param
参数传递给了前一个任务。在后续任务中,通过params
参数使用{{ task_instance.xcom_pull(task_ids="task1", key="param") }}
的形式获取该参数。
这些方法可以根据具体的需求选择使用。在使用参数时,可以根据实际情况构造SQL语句,传递给MySqlOperator执行相应的操作。
领取专属 10元无门槛券
手把手带您无忧上云