Airflow如何将BashOperator返回值转为字符串用于ODBC连接

你当前代码的核心问题是直接把BashOperator任务对象当成字符串拼接，正确做法是拉取该任务推送到XCom的执行结果，具体修改如下：

1. 核心修改点 #

• 把wsl_ip任务的输出作为参数传入run任务，Airflow会自动解析XCom中的返回值
• 对拿到的IP值做去空白处理，避免换行符/空格影响连接串格式
• 修正任务依赖关系，保证先执行wsl_ip再执行run

2. 修正后完整代码 #

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(wsl_ip_str):
        # 先去除IP字符串里的换行、空格等空白字符
        wsl_ip_str = wsl_ip_str.strip()
        
        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)

        # 这里用处理后的IP字符串拼接
        conn = odbc.connect('Driver={ODBC Driver 17 for SQL Server};Server= '+ wsl_ip_str +';Database=CVM;uid=Airflow;pwd=ubuntu')
        df = pd.read_sql_query('select max(DT_COMPTC) from Historico;', conn)
        # 按索引取第一列避免无列名报错
        left = df.iloc[:,0].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')

    # 把wsl_ip任务的输出作为参数传入run任务，自动建立依赖关系
    execute = run(wsl_ip.output)

etl_dag = my_dag2()

3. 额外注意事项 #

• 你原代码中查询max(DT_COMPTC)没有别名，直接取df['']会报错，建议修改SQL为select max(DT_COMPTC) as max_dt from Historico，再用df['max_dt']取值更稳妥
• 原代码中调用cursor.close时缺少括号，已修正为cursor.close()

内容的提问来源于stack exchange，提问作者Fcoatis