气流连接到 sql server 选择结果到数据框

如何解决气流连接到 sql server 选择结果到数据框

Airflow-pandas-read-sql-query 到数据框

我正在尝试连接到本地 sql 服务器以从表中获取数据并使用 Pandas 操作处理数据，但我无法弄清楚如何将选择查询结果传递给数据框 以下用于清除表格中的数据

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

预期的熊猫是

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 

解决方法

这是我用于 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
)

错误日志

[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,severity 9:\nAdaptive Server connection failed (localhost)\n")