Airflow连接SQL Server报错：数据源名称未找到且无默认驱动

错误信息 #

[Driver Manager]Data source name not found and no default driver specified (0) (SQLDriverConnect).

我的代码实现 #

import pendulum
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.hooks.base_hook import BaseHook
from datetime import datetime, timedelta
import pyodbc

local_tz = pendulum.timezone("Asia/Kuala_Lumpur")


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 3, 26),
    'email_on_failure': False,
    'email_on_retry': False
    #'retries': 1,
    #'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    '1.1.initial_onboard_customer',
    default_args=default_args,
    description='Transfer data from onboard.public.customer to report.public.onboard_customer',
    schedule_interval=None,  # You can specify your desired schedule interval here
    catchup=False,
    tags = ['Initial Data Load', 'UAT']
)



def transfer_data():
    try:
        # Get connection details from Airflow's connection pool
        source_conn = BaseHook.get_connection('UAT_DB')
        # dest_conn = BaseHook.get_connection('UAT_report_db_connection_string')

        # Connect to the source database
        source_conn_str = f"dbname={source_conn.schema} user={source_conn.login} password={source_conn.password} host={source_conn.host} port={source_conn.port}"
        print(source_conn_str)
        source_conn = pyodbc.connect(source_conn_str)
        print(source_conn)

        # Connect to the destination database
        # dest_conn_str = f"dbname={dest_conn.schema} user={dest_conn.login} password={dest_conn.password} host={dest_conn.host} port={dest_conn.port}"
        # dest_conn = psycopg2.connect(dest_conn_str)

        with source_conn.cursor() as source_cur:

            # Get customer details from the source database
            query = "Select TOP 10 * from [dbo].[Banks];"
            result = source_cur.execute(query)
            print(result)
        

    except Exception as e:
        print("Error occurred while transferring data:", e)
        raise


# Define the PythonOperator to execute the transfer_data function
transfer_data_task = PythonOperator(
    task_id='1.1.initial_onboard_customer',
    python_callable=transfer_data,
    dag=dag
)

解决方法 #

1. 安装SQL Server ODBC驱动 #

报错核心原因是pyodbc找不到SQL Server对应的ODBC驱动，需根据Airflow运行的系统安装对应版本：

• Windows：下载微软官方的ODBC Driver for SQL Server，推荐安装最新稳定版（如18）。
• Linux（Ubuntu/Debian）：执行命令安装：

  sudo apt-get update
  sudo apt-get install -y msodbcsql18 unixodbc-dev
  

• macOS：通过Homebrew安装：

  brew install microsoft/mssql-release/mssql-tools18
  brew install unixodbc
  

2. 修正pyodbc连接字符串格式 #

代码中使用了PostgreSQL风格的连接字符串，pyodbc不识别这种格式，需改为SQL Server专用的连接格式，示例如下：

def transfer_data():
    try:
        source_conn = BaseHook.get_connection('UAT_DB')
        # 正确的SQL Server连接字符串
        source_conn_str = f"""
        DRIVER={{ODBC Driver 18 for SQL Server}};
        SERVER={source_conn.host},{source_conn.port};
        DATABASE={source_conn.schema};
        UID={source_conn.login};
        PWD={source_conn.password};
        TrustServerCertificate=yes;  # 开发/UAT环境可启用，生产环境建议关闭
        """
        source_conn = pyodbc.connect(source_conn_str)
        print("连接成功")
        
        with source_conn.cursor() as source_cur:
            query = "Select TOP 10 * from [dbo].[Banks];"
            source_cur.execute(query)
            # 打印查询结果示例
            rows = source_cur.fetchall()
            for row in rows:
                print(row)
                
    except Exception as e:
        print("数据传输出错:", e)
        raise

注意：DRIVER的值要和你安装的驱动版本完全一致，比如安装的是ODBC Driver 17，就改成DRIVER={ODBC Driver 17 for SQL Server}。

3. 验证Airflow连接配置 #

确认Airflow中UAT_DB连接的以下参数正确：

• Host：SQL Server实例的IP或主机名
• Port：SQL Server监听端口（默认1433）
• Schema：要连接的数据库名称
• Login/Password：SQL Server的合法账号密码

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