有没有人设法使用 Docker apache/airflow:2.0.1 访问 SQL Server答案

【问题标题】：Has anyone managed to access SQL Server with Docker apache/airflow:2.0.1有没有人设法使用 Docker apache/airflow:2.0.1 访问 SQL Server
【发布时间】：2021-07-07 09:27:03
【问题描述】：

我正在使用基本容器容器apache/airflow:2.0.1 使用 docker-compose 尝试 apache 气流。

我正在关注这个教程https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html#

如何运行一个简单的查询来从 SQL Server 数据库中获取数据？

现阶段，我只是想看看是否可行。

我已经厌倦了扩展图像

FROM apache/airflow:2.0.1
RUN pip install apache-airflow-providers-microsoft-azure==1.2.0rc1
RUN pip install --no-cache-dir --user apache-airflow-providers-microsoft-mssql
# this fails
# RUN pip install --no-cache-dir --user apache-airflow-providers-odbc

用它来获取数据

def mssql_func(**kwargs):
    conn = MsSqlHook.get_connection(conn_id="mssql_default")
    hook = conn.get_hook()
    df = hook.get_pandas_df(sql="SELECT top 5 * FROM sometable")
    #do whatever you need on the df
    print(df)

有什么想法吗？

【问题讨论】：

那么你有什么错误吗？
那么您有 Microsoft SQL Server 吗？链接的教程使用的是 Postgres。
请发布完整的 dag 代码。
@derpirscher 我什至没有收到错误，这是我的问题。原来 Celery Scheduler 刚刚停止，没有一个 DAG 正在工作。我把它吹走了，然后错误出现了，让我找到了答案。无论如何，谢谢。
@AlwaysLearning 我正在从我的公司 SQL Server 构建一个数据湖，我需要访问 SQL Server 来获取数据以便构建我的模型 - 没有其他数据库连接可以做

标签： python sql-server docker airflow

【解决方案1】：

我在这里回答我自己的问题，主要是因为我最初的问题并没有给人们太多继续下去的机会。非常感谢所有回复的人。

我需要添加一些包以使 mssql 连接正常工作，因此我扩展了 docker 映像。这是一个错误，所以我也需要添加apache-airflow-providers-microsoft-azure。

Dockerfile

FROM apache/airflow:2.0.1
RUN pip install apache-airflow-providers-microsoft-azure==1.2.0rc1
RUN pip install apache-airflow-providers-microsoft-mssql

构建新映像并设置 env var 以在教程中指定的 docker-compose 中使用它

docker build -t search/apache:2.0.1 -f ./Dockerfile .
echo -e "AIRFLOW_IMAGE_NAME=custom/apache:2.0.1" >> .env

这是获取一些数据的代码，我通过 UI 在 Airflow 的连接屏幕上添加了连接。

DAG

from airflow import DAG
from airflow.operators.python import PythonOperator, get_current_context, task
from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'airflow',
}

with DAG(
    'test_etl_mssqlhook',
    default_args=default_args,
    description='ETL DAG Test 3',
    schedule_interval=None,
    start_date=days_ago(2),
    tags=['test'],
) as dag:
    dag.doc_md = __doc__

    def start(**kwargs):
        print("MEH!")

    def extract(**kwargs):
        conn = MsSqlHook.get_connection(conn_id="mssql_default")
        hook = conn.get_hook()
        df = hook.get_pandas_df(sql="SELECT top 5 * FROM dbo.atable")
        #do whatever you need on the df
        print(df)

    start_task = PythonOperator(
        task_id='start',
        python_callable=start,
    )

    extract_task = PythonOperator(
        task_id='extract',
        python_callable=extract,
    )

    start_task >> extract_task

【讨论】：