Airflow connect to sql server select results to a data frame

user_4706430 0
Viewed 2692

Airflow-pandas-read-sql-query to dataframe

i am trying to connect to SQL server local to get data from a table and process the data using pandas operations but i m failing to figure out how to pass the select query results to a data frame the below works to clear data in the table

``` sql_command = """ DELETE FROM [TestDB].[dbo].[PythonTestData] """

t3 = MsSqlOperator( task_id = 'run_test_proc',
                mssql_conn_id = 'mssql_local',
                sql = sql_command,
                dag = dag,
                database = 'TestDB',
                autocommit = True) ```

the intended pandas is


query = 'SELECT * FROM [ClientData] '#where  product_name='''+i+''''''

df = pd.read_sql(query, conn)
pn_list = df['ClientID'].tolist()
#print("The original pn_list is : " + str(pn_list))
for i in pn_list:
    varw= str(i)
    queryw = 'SELECT * FROM [ClientData] where  [ClientID]='''+varw+''
    dfw = pd.read_sql(queryw, conn)
    dfw = dfw.applymap(str)
    cols=['product_id','product_name','brand_id']
    x=dfw.values.tolist()
    x=x[0]
    ClientID=x[0]
    Name=x[1]
    Org=x[2]
    Email=x[3]
    #print('Name :'+Name+'   ,'+'Org :'+Org+'   ,'+'Email :'+Email+'    ,'+'ClientID :'+ClientID)
    salesData_qry= 'SELECT * FROM [TestDB].[dbo].[SalesData] where  [ClientID]='''+ClientID+''
    salesData_df= pd.read_sql(salesData_qry, conn)
    salesData_df['year1'] = salesData_df['Order Date'].dt.strftime('%Y')
    salesData_df['OrderMonth'] =  salesData_df['Order Date'].dt.strftime('%b')
    filename ='Daily_Campaign_Report_'+Name+'_'+Org+'_'+datetime.now().strftime("%Y%m%d_%H%M%S")
    p = Path('C:/Users/user/Documents/WorkingData/')
    salesData_df.to_csv(Path(p,  filename + '.csv'))```

Please point me to correct approach as i m new to airflow
airflow
2 Answers

I'm not so clear on how you generate the query code but in order to get dataframe from MsSQL you need to use MsSqlHook:

from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook

def mssql_func(**kwargs):
    hook = MsSqlHook(conn_id='mssql_local')
    df = hook.get_pandas_df(sql="YOUR_QUERY")
    #do whatever you need on the df


run_this = PythonOperator(
    task_id='mssql_task',
    python_callable=mssql_func,
    dag=dag
)
user_146244090

this is the code i am using for the dag

def mssql_func(**kwargs):
    conn = MsSqlHook.get_connection(conn_id="mssql_local")
    hook = conn.get_hook()
    df = hook.get_pandas_df(sql="SELECT * FROM [TestDB].[dbo].[ClientData]")
    #do whatever you need on the df
    print(df)

run_this = PythonOperator(
    task_id='mssql_task',
    python_callable=mssql_func,
    dag=dag
)

Error Log

[2021-01-12 16:07:15,114] {providers_manager.py:159} WARNING - The provider for package 'apache-airflow-providers-imap' could not be registered from because providers for that package name have already been registered
[2021-01-12 16:07:15,618] {base.py:65} INFO - Using connection to: id: mssql_local. Host: localhost, Port: 1433, Schema: dbo, Login: sa, Password: XXXXXXXX, extra: None
[2021-01-12 16:07:15,626] {taskinstance.py:1396} ERROR - (18456, b"Login failed for user 'sa'.DB-Lib error message 20018, severity 14:\nGeneral SQL Server error: Check messages from the SQL Server\nDB-Lib error message 20002, severity 9:\nAdaptive Server connection failed (localhost)\nDB-Lib error message 20002, severity 9:\nAdaptive Server connection failed (localhost)\n")
Traceback (most recent call last):
  File "src/pymssql.pyx", line 636, in pymssql.connect
  File "src/_mssql.pyx", line 1964, in _mssql.connect
  File "src/_mssql.pyx", line 682, in _mssql.MSSQLConnection.__init__
  File "src/_mssql.pyx", line 1690, in _mssql.maybe_raise_MSSQLDatabaseException
_mssql.MSSQLDatabaseException: (18456, b"Login failed for user 'sa'.DB-Lib error message 20018, severity 14:\nGeneral SQL Server error: Check messages from the SQL Server\nDB-Lib error message 20002, severity 9:\nAdaptive Server connection failed (localhost)\nDB-Lib error message 20002, severity 9:\nAdaptive Server connection failed (localhost)\n")
user_47064300
Related