python - 如何将来自 AWS ECR 的私有(private)镜像与 Airflow 的 DockerOperator 一起使用？

Question

我使用 CLI(使用 Click 构建)创建了一个自定义包。这个包可以做两件事:运行预处理和运行机器学习模型。我为这个客户包创建了一个 Docker 镜像，并将其推送到 AWS (ECR) 上的私有(private)注册表。

现在我想用 Airflow 运行这个容器，我想在 EC2 实例上运行它。我正在使用 docker-compose 运行它。

对于这个例子，我将只关注一个任务:运行容器进行预处理。

但是，现在我得到 t2 的“上游失败”。

from datetime import timedelta
import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.docker_operator import DockerOperator


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(1),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'example_pipeline',
    default_args=default_args,
    description='example data pipeline.',
    schedule_interval=timedelta(minutes=3)
)


t1 = BashOperator(
    task_id='starting_airflow',
    bash_command='echo "Starting Airflow DAG..."',
    dag=dag,
)


t2 = DockerOperator(
    task_id='data_pipeline',
    image='XXXXXXXX.dkr.ecr.eu-central-1.amazonaws.com/rwg:latest',
    container_name='task__export_data',
    command="run-preprocessing",
    network_mode="bridge",
    api_version="auto",
    docker_url="unix://var/run/docker.sock",
    docker_conn_id='aws_con',
    dag=dag
)

t1 >> t2

我通过 UI 创建了“aws_con”。但这似乎不起作用。

此外，这是我的 docker-compose.yml 文件。

version: '3'
services:
  postgres:
    image: postgres:9.6
    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow
    ports:
      - "5432:5432"
    volumes:
      - ./pgdata:/var/lib/postgresql/data

  webserver:
    build: .
    restart: always
    depends_on:
      - postgres
    environment:
      - LOAD_EX=n
      - EXECUTOR=Local
      - FERNET_KEY=jsDPRErfv8Z_eVTnGfF8ywd19j4pyqE3NpdUBA_oRTo=
    volumes:
      - ./dags:/usr/local/airflow/dags
    ports:
      - "8080:8080"
    command: webserver
    healthcheck:
      test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-webserver.pid ]"]
      interval: 30s
      timeout: 30s
      retries: 3

关于 Docker Operator，我做错了什么？

第二个问题:如何通过代码或 cli 创建这个“aws_con”？

Answer 1

您应该指定 docker 连接。

这是在 Airflow 中传递凭据的标准方式，Docker 有一个专用的 Docker 连接，您可以在 Airlfow DB 或 Secrets 中定义该连接，并将连接的 ID 作为 docker_conn_id 传递给 DockerOperator -参数(您也可以在那里指定 url，这样您就不需要在运算符(operator)中传递 docker_url)。

在这里查看 Python API:

https://airflow.apache.org/docs/apache-airflow-providers-docker/stable/_api/airflow/providers/docker/operators/docker/index.html

这里是关于 Docker 连接的单独页面(docker_conn_id 描述链接到此页面):

https://airflow.apache.org/docs/apache-airflow-providers-docker/stable/connections/docker.html#howto-connection-docker