如何在Airflow MsSqlOperator中使用mssql_conn_id？

Question

我正在尝试在Airflow工作流程中使用MsSqlOperator，但无法解决如何设置连接字符串的问题。

我尝试将mssql_conn_id设置为连接字符串本身

t2 = MsSqlOperator(
  task_id='sql-op',
  mssql_conn_id='sa:password@172.17.0.2',
  sql='use results; insert into airflow value("airflow","out")',
  dag=dag)

我得到错误

airflow.exceptions.AirflowException: The conn_id `sa:password@172.17.0.2` isn't defined

所以我想需要定义mssql_conn_id。某处。有任何想法吗？

我可以使用sqlalchemy这样连接到MS SQL数据库：

params = urllib.quote_plus("DRIVER={ODBC Driver 13 for SQL Server};SERVER=172.17.0.2;UID=SA;PWD=password")
engine = create_engine("mssql+pyodbc:///?odbc_connect=%s" % params)

conn = engine.connect()

因此我知道服务器已启动并正在运行。

Answer 1

该mssql_conn_id参数引用您的气流数据库中的连接条目，而不是实际的连接URI。

您可以通过以下几种方式来添加连接：

• UI：在“管理员”->“连接”下
• 命令行：使用 airflow connections --add --conn-id my_mssql --conn_uri mssql+pyodbc://sa:password@172.17.0.2
• 环境变量：设置 AIRFLOW_CONN_MY_MSSQL=mssql+pyodbc://sa:password@172.17.0.2

然后只需conn_id在运算符中引用即可：

t2 = MsSqlOperator(
    task_id='sql-op',
    mssql_conn_id='my_mssql',
    sql='use results; insert into airflow value("airflow","out")',
    dag=dag)