如何在Airflow的DockerOperator中使用AWS ECR私有镜像并解决配置问题

• 1. DAG代码存在HTML转义字符错误
     你贴的代码里的"、>>都是HTML转义后的内容，实际部署到dags目录的代码如果是这个写法，会直接触发语法错误，导致t1任务运行失败，进而t2报upstream failed错误。你需要把所有"替换为双引号"，把末尾的>>替换为>>。
• 2. Airflow容器未挂载宿主机docker.sock
     你的docker-compose.yml里webserver服务没有挂载宿主机的Docker套接字文件，即使DockerOperator里指定了docker_url="unix://var/run/docker.sock"，Airflow容器内部也无法访问到宿主机的Docker daemon，导致容器启动失败。
     修正方式是给webserver的volumes配置增加一行：

- /var/run/docker.sock:/var/run/docker.sock

同时要保证Airflow容器内的用户有权限访问该套接字，测试阶段可以临时给webserver增加user: root配置，生产环境建议将容器内的airflow用户加入docker用户组。

• 3. DockerOperator参数配置不合理
     你指定了固定的container_name='task__export_data'，如果上一次任务异常退出没有清理容器，下一次启动会因为容器重名报错。建议增加auto_remove=True参数，任务执行完成后自动销毁容器，也可以删除固定容器名配置，让Docker自动生成容器名。
     另外拉取私有ECR镜像建议增加force_pull=True参数，保证每次运行都拉取最新的镜像版本。
• 4. aws_con连接配置错误
     Airflow的Docker类型连接拉取ECR镜像有固定的配置规则，如果UI创建时参数填错会导致镜像拉取失败：
  • 连接类型选择Docker
  • Host填写你的ECR仓库地址：XXXXXXXX.dkr.ecr.eu-central-1.amazonaws.com
  • Login固定填写AWS
  • Password填写aws ecr get-login-password --region eu-central-1返回的临时密钥，注意该密钥有效期只有12小时，建议给EC2实例绑定有权限拉取ECR镜像的IAM角色，无需手动配置密钥。

1. CLI方式创建 #

首先进入Airflow webserver容器，执行以下命令：

airflow connections add 'aws_con' \
--conn-type 'docker' \
--conn-host 'XXXXXXXX.dkr.ecr.eu-central-1.amazonaws.com' \
--conn-login 'AWS' \
--conn-password "$(aws ecr get-login-password --region eu-central-1)"

注意需要先在Airflow容器内安装awscli工具，并且配置好AWS访问权限。

2. 代码方式创建 #

可以编写单独的初始化脚本执行创建：

from airflow.models import Connection
from airflow.utils.session import create_session
import boto3

# 配置你的ECR区域
REGION = "eu-central-1"
ECR_HOST = "XXXXXXXX.dkr.ecr.eu-central-1.amazonaws.com"

# 获取ECR临时登录密码
ecr = boto3.client("ecr", region_name=REGION)
token = ecr.get_authorization_token()["authorizationData"][0]
password = token["authorizationToken"].decode().split(":")[1]

with create_session() as session:
    # 先删除已存在的同名连接避免重复
    session.query(Connection).filter(Connection.conn_id == "aws_con").delete()
    # 创建新连接
    docker_conn = Connection(
        conn_id="aws_con",
        conn_type="docker",
        host=ECR_HOST,
        login="AWS",
        password=password
    )
    session.add(docker_conn)
    session.commit()

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