sql-server - 如何在 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 参数指的是 connection您的 Airflow 数据库中的条目，而不是实际的连接 URI。

您有几个用于添加连接的选项:

• 用户界面:在管理 -> 连接下
• 命令行:使用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)