Version Conditions préalables Permissions Création d'un cluster Amazon ECS Exemple de code

Connexion à Amazon ECS à l'aide de l'opérateur `ECS`

Cette rubrique décrit comment vous pouvez utiliser le ECSOperator pour vous connecter à un conteneur Amazon Elastic Container Service (Amazon ECS) depuis Amazon MWAA. Au cours des étapes suivantes, vous allez ajouter les autorisations requises au rôle d'exécution de votre environnement, utiliser un CloudFormation modèle pour créer un cluster Amazon ECS Fargate, et enfin créer et charger un DAG qui se connecte à votre nouveau cluster.

Version

Vous pouvez utiliser l'exemple de code de cette page avec Apache Airflow v2 en Python 3.10 et Apache Airflow v3 en Python 3.11.

Conditions préalables

Pour utiliser l'exemple de code de cette page, vous aurez besoin des éléments suivants :

Un environnement Amazon MWAA.

Permissions

Le rôle d'exécution de votre environnement nécessite une autorisation pour exécuter des tâches dans Amazon ECS. Vous pouvez soit associer la politique FullAccess AWS gérée par AmazonECS_ à votre rôle d'exécution, soit créer et associer la politique suivante à votre rôle d'exécution.

Outre l'ajout des autorisations requises pour exécuter des tâches dans Amazon ECS, vous devez également modifier la déclaration de politique relative aux CloudWatch journaux dans votre rôle d'exécution Amazon MWAA afin de permettre l'accès au groupe de journaux de tâches Amazon ECS, comme indiqué ci-dessous. Le groupe de journaux Amazon ECS est créé par le CloudFormation modèle dansCréation d'un cluster Amazon ECS.
```
{
  "Effect": "Allow",
  "Action": [
    "logs:CreateLogStream",
    "logs:CreateLogGroup",
    "logs:PutLogEvents",
    "logs:GetLogEvents",
    "logs:GetLogRecord",
    "logs:GetLogGroupFields",
    "logs:GetQueryResults"
  ],
  "Resource": [
    "arn:aws:logs:us-east-1:123456789012:log-group:airflow-environment-name-*",
    "arn:aws:logs:*:*:log-group:ecs-mwaa-group:*"
  ]
}
```

Pour plus d'informations sur le rôle d'exécution Amazon MWAA et sur la manière d'associer une politique, consultez. Rôle d'exécution

Création d'un cluster Amazon ECS

À l'aide du CloudFormation modèle suivant, vous allez créer un cluster Amazon ECS Fargate à utiliser avec votre flux de travail Amazon MWAA. Pour plus d'informations, reportez-vous à la section Création d'une définition de tâche dans le guide du développeur Amazon Elastic Container Service.

Créez un fichier JSON avec le code suivant et enregistrez-le sousecs-mwaa-cfn.json.


{
    "AWSTemplateFormatVersion": "2010-09-09",
    "Description": "This template deploys an ECS Fargate cluster with an Amazon Linux image as a test for MWAA.",
    "Parameters": {
        "VpcId": {
            "Type": "AWS::EC2::VPC::Id",
            "Description": "Select a VPC that allows instances access to ECR, as used with MWAA."
        },
        "SubnetIds": {
            "Type": "List<AWS::EC2::Subnet::Id>",
            "Description": "Select at two private subnets in your selected VPC, as used with MWAA."
        },
        "SecurityGroups": {
            "Type": "List<AWS::EC2::SecurityGroup::Id>",
            "Description": "Select at least one security group in your selected VPC, as used with MWAA."
        }
    },
    "Resources": {
        "Cluster": {
            "Type": "AWS::ECS::Cluster",
            "Properties": {
                "ClusterName": {
                    "Fn::Sub": "${AWS::StackName}-cluster"
                }
            }
        },
        "LogGroup": {
            "Type": "AWS::Logs::LogGroup",
            "Properties": {
                "LogGroupName": {
                    "Ref": "AWS::StackName"
                },
                "RetentionInDays": 30
            }
        },
        "ExecutionRole": {
            "Type": "AWS::IAM::Role",
            "Properties": {
                "AssumeRolePolicyDocument": {
                    "Statement": [
                        {
                            "Effect": "Allow",
                            "Principal": {
                                "Service": "ecs-tasks.amazonaws.com"
                            },
                            "Action": "sts:AssumeRole"
                        }
                    ]
                },
                "ManagedPolicyArns": [
                    "arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy"
                ]
            }
        },
        "TaskDefinition": {
            "Type": "AWS::ECS::TaskDefinition",
            "Properties": {
                "Family": {
                    "Fn::Sub": "${AWS::StackName}-task"
                },
                "Cpu": 2048,
                "Memory": 4096,
                "NetworkMode": "awsvpc",
                "ExecutionRoleArn": {
                    "Ref": "ExecutionRole"
                },
                "ContainerDefinitions": [
                    {
                        "Name": {
                            "Fn::Sub": "${AWS::StackName}-container"
                        },
                        "Image": "137112412989.dkr.ecr.us-east-1.amazonaws.com/amazonlinux:latest",
                        "PortMappings": [
                            {
                                "Protocol": "tcp",
                                "ContainerPort": 8080,
                                "HostPort": 8080
                            }
                        ],
                        "LogConfiguration": {
                            "LogDriver": "awslogs",
                            "Options": {
                                "awslogs-region": {
                                    "Ref": "AWS::Region"
                                },
                                "awslogs-group": {
                                    "Ref": "LogGroup"
                                },
                                "awslogs-stream-prefix": "ecs"
                            }
                        }
                    }
                ],
                "RequiresCompatibilities": [
                    "FARGATE"
                ]
            }
        },
        "Service": {
            "Type": "AWS::ECS::Service",
            "Properties": {
                "ServiceName": {
                    "Fn::Sub": "${AWS::StackName}-service"
                },
                "Cluster": {
                    "Ref": "Cluster"
                },
                "TaskDefinition": {
                    "Ref": "TaskDefinition"
                },
                "DesiredCount": 1,
                "LaunchType": "FARGATE",
                "PlatformVersion": "1.3.0",
                "NetworkConfiguration": {
                    "AwsvpcConfiguration": {
                        "AssignPublicIp": "ENABLED",
                        "Subnets": {
                            "Ref": "SubnetIds"
                        },
                        "SecurityGroups": {
                            "Ref": "SecurityGroups"
                        }
                    }
                }
            }
        }
    }
}

Dans votre invite de commande, utilisez la AWS CLI commande suivante pour créer une nouvelle pile. Vous devez remplacer les valeurs par des valeurs pour SubnetIds les groupes de sécurité SecurityGroups et les sous-réseaux de votre environnement Amazon MWAA.


aws cloudformation create-stack \
--stack-name my-ecs-stack --template-body file://ecs-mwaa-cfn.json \
--parameters ParameterKey=SecurityGroups,ParameterValue=your-mwaa-security-group \
ParameterKey=SubnetIds,ParameterValue=your-mwaa-subnet-1\\,your-mwaa-subnet-1 \
--capabilities CAPABILITY_IAM

Vous pouvez également utiliser le script shell suivant. Le script extrait les valeurs requises pour les groupes de sécurité et les sous-réseaux de votre environnement à l'aide de la get-environment AWS CLI commande, puis crée la pile en conséquence. Pour exécuter le script, procédez comme suit.

Copiez et enregistrez le script ecs-stack-helper.sh dans le même répertoire que votre CloudFormation modèle.


#!/bin/bash

joinByString() {
  local separator="$1"
  shift
  local first="$1"
  shift
  printf "%s" "$first" "${@/#/$separator}"
}

response=$(aws mwaa get-environment --name $1)

securityGroupId=$(echo "$response" | jq -r '.Environment.NetworkConfiguration.SecurityGroupIds[]')
subnetIds=$(joinByString '\,' $(echo "$response" | jq -r '.Environment.NetworkConfiguration.SubnetIds[]'))

aws cloudformation create-stack --stack-name $2 --template-body file://ecs-cfn.json \
--parameters ParameterKey=SecurityGroups,ParameterValue=$securityGroupId \
ParameterKey=SubnetIds,ParameterValue=$subnetIds \
--capabilities CAPABILITY_IAM

Exécutez le script à l'aide des commandes suivantes. Remplacez environment-name et stack-name par vos informations.
```
chmod +x ecs-stack-helper.sh
./ecs-stack-helper.bash environment-name stack-name
```

En cas de succès, vous vous référez à la sortie suivante indiquant votre nouvel identifiant de CloudFormation pile.

{
  "StackId": "arn:aws:cloudformation:us-west-2:123456789012:stack/my-ecs-stack/123456e7-8ab9-01cd-b2fb-36cce63786c9"
}

Une fois que votre CloudFormation stack est AWS terminé et que vos ressources Amazon ECS ont été provisionnées, vous êtes prêt à créer et à charger votre DAG.

Exemple de code

Ouvrez une invite de commande et accédez au répertoire dans lequel votre code DAG est stocké. Par exemple :
```
cd dags
```

Copiez le contenu de l'exemple de code suivant et enregistrez-le localement sousmwaa-ecs-operator.py, puis chargez votre nouveau DAG sur Amazon S3.

Apache Airflow v3


import boto3
import pendulum

from airflow.providers.amazon.aws.operators.ecs import EcsRunTaskOperator
from airflow.sdk import dag, task

CLUSTER_NAME="mwaa-ecs-test-cluster" #Replace value for CLUSTER_NAME with your information.
CONTAINER_NAME="mwaa-ecs-test-container" #Replace value for CONTAINER_NAME with your information.
LAUNCH_TYPE = "FARGATE"


@dag(
    schedule=None,
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["ecs", "fargate"],
)
def ecs_fargate_dag():
    @task()
    def get_ecs_service_config() -> dict:
        """Fetch task definition and network config from the ECS service."""
        client = boto3.client("ecs")
        services = client.list_services(cluster=CLUSTER_NAME, launchType=LAUNCH_TYPE)
        service = client.describe_services(cluster=CLUSTER_NAME, services=services["serviceArns"])
        return {
            "task_definition": service["services"][0]["taskDefinition"],
            "network_configuration": service["services"][0]["networkConfiguration"],
        }

    service_config = get_ecs_service_config()

    EcsRunTaskOperator(
        task_id="ecs_operator_task",
        cluster=CLUSTER_NAME,
        task_definition=service_config["task_definition"],
        launch_type=LAUNCH_TYPE,
        overrides={
            "containerOverrides": [
              {
                  "name": CONTAINER_NAME,
                  "command": ["ls", "-l", "/"],
              },
            ],
        },
        network_configuration=service_config["network_configuration"],
        awslogs_group="mwaa-ecs-zero",
        awslogs_stream_prefix=f"ecs/{CONTAINER_NAME}",
    )


ecs_fargate_dag()

Apache Airflow v2


from http import client
from airflow import DAG
from airflow.providers.amazon.aws.operators.ecs import EcsRunTaskOperator
from airflow.utils.dates import days_ago
import boto3

CLUSTER_NAME="mwaa-ecs-test-cluster" #Replace value for CLUSTER_NAME with your information.
CONTAINER_NAME="mwaa-ecs-test-container" #Replace value for CONTAINER_NAME with your information.
LAUNCH_TYPE="FARGATE"

with DAG(
    dag_id = "ecs_fargate_dag",
    schedule_interval=None,
    catchup=False,
    start_date=days_ago(1)
) as dag:
  client=boto3.client('ecs')
  services=client.list_services(cluster=CLUSTER_NAME,launchType=LAUNCH_TYPE)
  service=client.describe_services(cluster=CLUSTER_NAME,services=services['serviceArns'])

  ecs_operator_task = EcsRunTaskOperator(
    task_id = "ecs_operator_task",
    dag=dag,
    cluster=CLUSTER_NAME,
    task_definition=service['services'][0]['taskDefinition'],
    launch_type=LAUNCH_TYPE,
    overrides={
      "containerOverrides":[
        {
          "name":CONTAINER_NAME,
          "command":["ls", "-l", "/"],
        },
      ],
    },
    network_configuration=service['services'][0]['networkConfiguration'],
    awslogs_group="mwaa-ecs-zero",
    awslogs_stream_prefix=f"ecs/{CONTAINER_NAME}",
  )

Note

Dans l'exemple DAG, pourawslogs_group, vous devrez peut-être modifier le groupe de journaux en lui attribuant le nom de votre groupe de journaux de tâches Amazon ECS. L'exemple suppose un groupe de journaux nommémwaa-ecs-zero. Pourawslogs_stream_prefix, utilisez le préfixe de flux du journal des tâches Amazon ECS. L'exemple suppose un préfixe de flux de journal,ecs.

Exécutez la AWS CLI commande suivante pour copier le DAG dans le bucket de votre environnement, puis déclenchez le DAG à l'aide de l'interface utilisateur d'Apache Airflow.
```
aws s3 cp your-dag.py s3://your-environment-bucket/dags/
```

En cas de succès, vous obtiendrez une sortie similaire à la suivante dans les journaux des ecs_operator_task tâches du ecs_fargate_dag DAG :

[2022-01-01, 12:00:00 UTC] {{ecs.py:300}} INFO - Running ECS Task -
Task definition: arn:aws:ecs:us-west-2:123456789012:task-definition/mwaa-ecs-test-task:1 - on cluster mwaa-ecs-test-cluster
[2022-01-01, 12:00:00 UTC] {{ecs-operator-test.py:302}} INFO - ECSOperator overrides:
{'containerOverrides': [{'name': 'mwaa-ecs-test-container', 'command': ['ls', '-l', '/']}]}
.
.
.
[2022-01-01, 12:00:00 UTC] {{ecs.py:379}} INFO - ECS task ID is: e012340b5e1b43c6a757cf012c635935
[2022-01-01, 12:00:00 UTC] {{ecs.py:313}} INFO - Starting ECS Task Log Fetcher
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] total 52
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx   1 root root    7 Jun 13 18:51 bin -> usr/bin
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-xr-x   2 root root 4096 Apr  9  2019 boot
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   5 root root  340 Jul 19 17:54 dev
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   1 root root 4096 Jul 19 17:54 etc
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   2 root root 4096 Apr  9  2019 home
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx   1 root root    7 Jun 13 18:51 lib -> usr/lib
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx   1 root root    9 Jun 13 18:51 lib64 -> usr/lib64
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   2 root root 4096 Jun 13 18:51 local
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   2 root root 4096 Apr  9  2019 media
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   2 root root 4096 Apr  9  2019 mnt
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   2 root root 4096 Apr  9  2019 opt
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-xr-x 103 root root    0 Jul 19 17:54 proc
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-x-\-\-   2 root root 4096 Apr  9  2019 root
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   2 root root 4096 Jun 13 18:52 run
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx   1 root root    8 Jun 13 18:51 sbin -> usr/sbin
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x   2 root root 4096 Apr  9  2019 srv
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-xr-x  13 root root    0 Jul 19 17:54 sys
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxrwxrwt   2 root root 4096 Jun 13 18:51 tmp
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x  13 root root 4096 Jun 13 18:51 usr
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x  18 root root 4096 Jun 13 18:52 var
.
.
.
[2022-01-01, 12:00:00 UTC] {{ecs.py:328}} INFO - ECS Task has been successfully executed

Avertissement JavaScript est désactivé ou n'est pas disponible dans votre navigateur.

Pour que vous puissiez utiliser la documentation AWS, Javascript doit être activé. Vous trouverez des instructions sur les pages d'aide de votre navigateur.

Conventions de rédaction

Amazon EKS (extrait)

Utiliser dbt avec Amazon MWAA

Connexion à Amazon ECS à l'aide de l'opérateur ECS