Docker环境下Airflow与Airbyte连接异常问题求助

问题描述 #

我正在学习基于Docker搭建Apache Airflow与Airbyte的数据管道，目前遇到两者无法建立连接的问题。查阅相关文档及同类问题帖，尝试调整配置但问题仍未解决。想确认我的连接配置是否存在问题，求经过验证的可用配置（含版本信息）。

系统配置 #

• Windows 10 操作系统
• Docker Desktop v4.34.3 (170107)
• Apache Airflow v2.8.3（已测试v2.10.3及最新版本）
• Airbyte v0.63.13

额外设置说明 #

• Airflow webserver端口为8080
• Airbyte端口为8000
• Airflow的docker-compose.yaml中，webserver和scheduler的extra_hosts字段均配置了"host.docker.internal:host-gateway"

Airflow与Airbyte的连接设置 #

• 对比线上配置示例，Airbyte连接模板无端口字段，因此我使用HTTP连接方式

错误日志（运行Airflow DAG时） #

f213ecff2dd7
*** Found logs served from host http://f213ecff2dd7:8793/log/dag_id=elt_and_dbt/run_id=manual__2024-11-10T10:19:32.960623+00:00/task_id=airbyte_postgres_postgres/attempt=1.log
[2024-11-10, 10:19:43 UTC] {local_task_job_runner.py:123} ▶ Pre task execution logs
[2024-11-10, 10:19:43 UTC] {base.py:84} INFO - Retrieving connection 'airbyte'
[2024-11-10, 10:19:43 UTC] {taskinstance.py:3311} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/airbyte/hooks/airbyte.py", line 157, in submit_sync_connection
    res = self.airbyte_api.jobs.create_job(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airbyte_api/jobs.py", line 100, in create_job
    raise e
  File "/home/airflow/.local/lib/python3.12/site-packages/airbyte_api/jobs.py", line 94, in create_job
    req = client.prepare_request(requests_http.Request('POST', url, params=query_params, data=data, files=form, headers=headers))
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/requests/sessions.py", line 484, in prepare_request
    p.prepare(
  File "/home/airflow/.local/lib/python3.12/site-packages/requests/models.py", line 367, in prepare
    self.prepare_url(url, params)
  File "/home/airflow/.local/lib/python3.12/site-packages/requests/models.py", line 438, in prepare_url
    raise MissingSchema(
requests.exceptions.MissingSchema: Invalid URL 'host.docker.internal/jobs': No scheme supplied. Perhaps you meant https://host.docker.internal/jobs?
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 767, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 733, in _execute_callable
    return ExecutionCallableRunner(
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/operator_helpers.py", line 252, in run
    return self.func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 417, in wrapper
    return func(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/airbyte/operators/airbyte.py", line 83, in execute
    job_object = hook.submit_sync_connection(connection_id=self.connection_id)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/airbyte/hooks/airbyte.py", line 165, in submit_sync_connection
    raise AirflowException(e)
airflow.exceptions.AirflowException: Invalid URL 'host.docker.internal/jobs': No scheme supplied. Perhaps you meant https://host.docker.internal/jobs?
[2024-11-10, 10:19:43 UTC] {taskinstance.py:1225} INFO - Marking task as FAILED. dag_id=elt_and_dbt, task_id=airbyte_postgres_postgres, run_id=manual__2024-11-10T10:19:32.960623+00:00, execution_date=20241110T101932, start_date=20241110T101943, end_date=20241110T101943
[2024-11-10, 10:19:43 UTC] {taskinstance.py:340} ▶ Post task execution logs

DAG文件代码片段（仅展示Airbyte相关部分） #

from airflow import DAG
from docker.types import Mount
from airflow.utils.dates import days_ago
from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
from airflow.providers.docker.operators.docker import DockerOperator

# Acquiring connection ID from Airbyte UI
CONN_ID = '594c2ffe-fe0e-4ccb-9c29-c02f4a9fb2bd'

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False
}

dag = DAG(
    'elt_and_dbt',
    default_args=default_args,
    description='An ELT workflow with dbt',
    start_date=days_ago(1),
    catchup=False
)

t1 = AirbyteTriggerSyncOperator(
    task_id='airbyte_postgres_postgres',
    airbyte_conn_id='airbyte',
    connection_id=CONN_ID,
    asynchronous=False,
    timeout=3600,
    wait_seconds=3,
    dag=dag
)

已尝试的操作 #

• 删除Airflow相关容器、卷和镜像，更换v2.10.3及最新版本，问题依旧
• Airbyte与数据库的连接（同样使用host.docker.internal）正常，暂未重新安装Airbyte
• 尝试过将Airflow与Airbyte连接的host字段设置为host.docker.internal和localhost，均无效
• 使用HTTP模板时尝试https://host.docker.internal，出现如下错误：

requests.exceptions.ConnectionError: HTTPSConnectionPool(host='host.docker.internal', port=443): Max retries exceeded with url: /v1/applications/token (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7fa52e062990>: Failed to establish a new connection: [Errno 111] Connection refused'))
During handling of the above exception, another exception occurred:

问题根源及解决方案 #

核心问题 #

从错误日志可明确：

1. 请求URL缺少协议头（http/https），导致解析失败
2. 未指定Airbyte的正确端口8000，且错误使用HTTPS协议（Airbyte默认用HTTP）

经过验证的可用配置 #

1. Airflow中Airbyte连接的正确配置

不要使用HTTP连接模板，Airflow提供了专门的Airbyte连接类型：

• 连接类型：选择Airbyte（而非HTTP）
• Host：host.docker.internal（如果加入同一Docker网络，也可以用Airbyte容器名airbyte-server）
• Port：8000（Airbyte默认API端口）
• API密钥：如果Airbyte未设置API密钥，留空即可

2. Docker网络互通优化（可选但推荐）

如果Airflow和Airbyte分别通过各自的docker-compose启动，建议将它们加入同一个自定义网络，避免依赖host.docker.internal：

• 创建自定义网络：

  docker network create data-pipeline-net
  

• 修改Airflow的docker-compose.yaml，添加网络配置：

  networks:
    default:
      external:
        name: data-pipeline-net
  

• 修改Airbyte的docker-compose.yaml，同样加入该网络

3. 版本兼容性组合

以下版本组合经过测试可正常工作：

• Docker Desktop v4.32.0+
• Apache Airflow v2.8.3 / v2.10.3
• Airbyte v0.63.13

验证步骤 #

1. 按照上述配置重新创建Airflow的Airbyte连接
2. 重启Airflow的scheduler和webserver容器
3. 触发DAG测试连接

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