How to use mssql_conn_id in Airflow MsSqlOperator?

s_t_20911 0
Viewed 12777

I'm trying to use the MsSqlOperator in my Airflow workflow, but I can't work out how to set the connection string.

I've tried setting mssql_conn_id to be the connection string itself

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)

I get the error

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

so I suppose mssql_conn_id needs to be defined. Somewhere. Any ideas?

I'm able to connect to the MS SQL database using sqlalchemy like this:

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()

so I know the server is up and running.

airflow sql-server
2 Answers

The mssql_conn_id parameter refers to a connection entry in your airflow database, not the actual connection URI.

You have a few options for adding a connection:

  • UI: under Admin -> Connections
  • Command line: use airflow connections --add --conn-id my_mssql --conn_uri mssql+pyodbc://sa:password@172.17.0.2
  • Environment variable: set AIRFLOW_CONN_MY_MSSQL=mssql+pyodbc://sa:password@172.17.0.2

Then just reference the conn_id in the operator:

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

Code Complete

Go in admin > Connection and edit mssql_default

from airflow import DAG
from airflow.operators.mssql_operator import MsSqlOperator    

default_arg = {'owner': 'airflow'
              ,'start_date': '2022-01-01'
}

dag = DAG(
    'name_task',
    default_args=default_arg,
    description='description here',
    schedule_interval=None,
    catchup=False
)

task = MsSqlOperator(
         task_id='task_test',
         mssql_conn_id='mssql_default',
         sql=f"create table abc (a int)",            
         autocommit=True,
         database='DatabaseGilmar',
         dag=dag
)
user_119904700
Related