我對 Airflow 和 Python 很陌生。我想要做的是從 Bash 命令中獲取結果并用該回傳組合連接字串：
import pyodbc as odbc
import pandas as pd
import datetime as dt
from airflow.operators.bash import BashOperator
from airflow.decorators import dag, task
@dag(schedule_interval='0 15 10 * *', start_date=dt.datetime(2021, 10, 1), catchup=False)
def my_dag2():
wsl_ip = BashOperator(
task_id="wsl_ip",
bash_command="grep -m 1 nameserver /etc/resolv.conf | awk '{print $2}'",
do_xcom_push=True
)
@task()
def run():
def busca_informes_cvm(ano,mes):
url = 'http://dados.cvm.gov.br/dados/FI/DOC/INF_DIARIO/DADOS/inf_diario_fi_{:4d}{:02d}.csv'.format(ano,mes)
return pd.read_csv(url, sep=';')
today = dt.date.today()
ano = (today.year)
mes= (today.month)-1
file_name_Comp = '{:4d}-{:02d}'.format(ano,mes)
file_name = '{:4d}{:02d}.csv'.format(ano,mes)
path_name = r'C:\Airflow\{:4d}{:02d}.csv'.format(ano,mes)
conn = odbc.connect('Driver={ODBC Driver 17 for SQL Server};Server= ' wsl_ip ';Database=CVM;uid=Airflow;pwd=ubuntu')
df = pd.read_sql_query('select max(DT_COMPTC) from Historico;', conn)
left = df[''].str[:7]
if file_name_Comp <= left[0]:
print('Sair')
else:
informes_diarios = busca_informes_cvm(ano,mes)
informes_diarios.to_csv(file_name, sep=';', index=False)
db_view_nm = '[dbo].[Bulk]'
qry = "BULK INSERT " db_view_nm " FROM '" path_name "' WITH (FIELDTERMINATOR = ';', ROWTERMINATOR = '0x0a', FIRSTROW = 2,ROWS_PER_BATCH = 100000 )"
cursor = conn.cursor()
success = cursor.execute(qry)
conn.commit()
cursor.close
print('Concluído')
execute = run()
etl_dag = my_dag2()
我需要找到一種將 wsl_ip ia 轉換為字串的方法。任何幫助，將不勝感激。
uj5u.com熱心網友回復：
要獲取“wsl_ip”任務的輸出，您可以使用.output為 Airflow 中的每個操作員公開的屬性。此屬性是對xcom_pull()稱為 an的經典方法的抽象XComArg（請參閱此處的檔案）。
你可以嘗試這樣的事情：
wsl_ip = BashOperator(
task_id="wsl_ip",
bash_command="grep -m 1 nameserver /etc/resolv.conf | awk '{print $2}'",
do_xcom_push=True
)
@task()
def run(ip):
...
conn = odbc.connect('Driver={ODBC Driver 17 for SQL Server};Server= ' ip ';Database=CVM;uid=Airflow;pwd=ubuntu')
...
execute = run(ip=wsl_ip.output)
該run任務流函式現在需要的輸入XCom您的推BashOperator任務應轉換為字串。使用.output這種方式還自動創建了“wsl_ip”和“運行”任務太之間的任務相關。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/353029.html
標籤：Python sql-server 空气流动
上一篇：我想更新表中的字串