Version Prérequis Création d'une clé publique pour Amazon EC2 Création du cluster Créez unmwaaespace de noms Créez un rôle pourmwaaespace de noms Création et attachement d'un rôle IAM pour le cluster Amazon EKS Créez le fichier requirements.txt Création d'un mappage d'identité pour Amazon EKS Créer le kubeconfig Création d'un DAG Ajoutez le DAG etkube_config.yamlvers le compartiment Amazon S3 Activer et déclencher l'exemple

Utilisation d'Amazon MAAA avec Amazon EKS

L'exemple suivant montre comment utiliser Amazon Managed Workflows pour Apache Airflow avec Amazon EKS.

Rubriques

Version
Prérequis
Création d'une clé publique pour Amazon EC2
Création du cluster
Créez unmwaaespace de noms
Créez un rôle pourmwaaespace de noms
Création et attachement d'un rôle IAM pour le cluster Amazon EKS
Créez le fichier requirements.txt
Création d'un mappage d'identité pour Amazon EKS
Créer le kubeconfig
Création d'un DAG
Ajoutez le DAG etkube_config.yamlvers le compartiment Amazon S3
Activer et déclencher l'exemple

Version

L'exemple de code de cette page peut être utilisé avecApache Airflow v1dansPython 3.7.

Vous pouvez utiliser l'exemple de code de cette page avecApache Airflow v2 et versions ultérieuresdansPython 3.10.

Prérequis

Pour utiliser l'exemple présenté dans cette rubrique, vous aurez besoin des éléments suivants :

UnEnvironnement Amazon MWA.
eksctl. Pour en savoir plus, voirInstaller eksctl.
kubectl. Pour en savoir plus, voirInstallation et configuration de kubectl. Dans certains cas, il est installé avec eksctl.
Une paire de clés EC2 dans la région dans laquelle vous créez votre environnement Amazon MWAA. Pour en savoir plus, voirCréation ou importation d'une paire de clés.

Note

Lorsque vous utilisez uneksctlcommande, vous pouvez inclure une--profilepour spécifier un profil autre que le profil par défaut.

Création d'une clé publique pour Amazon EC2

Utilisez la commande suivante pour créer une clé publique à partir de votre paire de clés privées.


ssh-keygen -y -f myprivatekey.pem > mypublickey.pub

Pour en savoir plus, voirRécupération de la clé publique pour votre paire de clés.

Création du cluster

Utilisez la commande suivante pour créer le cluster. Si vous souhaitez attribuer un nom personnalisé au cluster ou le créer dans une autre région, remplacez les valeurs du nom et de la région. Vous devez créer le cluster dans la même région que celle dans laquelle vous avez créé l'environnement Amazon MWAA. Remplacez les valeurs des sous-réseaux pour qu'elles correspondent aux sous-réseaux de votre réseau Amazon VPC que vous utilisez pour Amazon MWAA. Remplacez la valeur dussh-public-keycorrespondant à la clé que vous utilisez. Vous pouvez utiliser une clé existante d'Amazon EC2 qui se trouve dans la même région, ou créer une nouvelle clé dans la même région que celle dans laquelle vous créez votre environnement Amazon MWAA.


eksctl create cluster \
--name mwaa-eks \
--region us-west-2 \
--version 1.18 \
--nodegroup-name linux-nodes \
--nodes 3 \
--nodes-min 1 \
--nodes-max 4 \
--with-oidc \
--ssh-access \
--ssh-public-key MyPublicKey \
--managed \
--vpc-public-subnets "subnet-11111111111111111, subnet-2222222222222222222" \
--vpc-private-subnets "subnet-33333333333333333, subnet-44444444444444444"

La création du cluster prend un certain temps. Une fois que vous avez terminé, vous pouvez vérifier que le cluster a été créé avec succès et que le fournisseur IAM OIDC est configuré à l'aide de la commande suivante :


eksctl utils associate-iam-oidc-provider \
--region us-west-2 \
--cluster mwaa-eks \
--approve

Créez un`mwaa`espace de noms

Après avoir confirmé que le cluster a été créé avec succès, utilisez la commande suivante pour créer un espace de noms pour les pods.


kubectl create namespace mwaa

Créez un rôle pour`mwaa`espace de noms

Après avoir créé l'espace de nommage, créez un rôle et une liaison de rôles pour un utilisateur Amazon MWAA sur EKS qui pourra exécuter des pods dans un espace de nommage MWAA. Si vous avez utilisé un nom différent pour l'espace de noms, remplacez mwaa dans-n mwaaavec le nom que vous avez utilisé.


cat << EOF | kubectl apply -f - -n mwaa
kind: Role
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: mwaa-role
rules:
  - apiGroups:
      - ""
      - "apps"
      - "batch"
      - "extensions"
    resources:      
      - "jobs"
      - "pods"
      - "pods/attach"
      - "pods/exec"
      - "pods/log"
      - "pods/portforward"
      - "secrets"
      - "services"
    verbs:
      - "create"
      - "delete"
      - "describe"
      - "get"
      - "list"
      - "patch"
      - "update"
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: mwaa-role-binding
subjects:
- kind: User
  name: mwaa-service
roleRef:
  kind: Role
  name: mwaa-role
  apiGroup: rbac.authorization.k8s.io
EOF

Vérifiez que le nouveau rôle peut accéder au cluster Amazon EKS en exécutant la commande suivante. Assurez-vous d'utiliser le nom correct si vous n'avez pas utilisémwaa:


kubectl get pods -n mwaa --as mwaa-service

Vous devriez voir s'afficher un message indiquant :


No resources found in mwaa namespace.

Création et attachement d'un rôle IAM pour le cluster Amazon EKS

Vous devez créer un rôle IAM, puis le lier au cluster Amazon EKS (k8s) afin qu'il puisse être utilisé pour l'authentification via IAM. Le rôle est utilisé uniquement pour se connecter au cluster et ne dispose d'aucune autorisation pour les appels de console ou d'API.

Créez un nouveau rôle pour l'environnement Amazon MWAA en suivant les étapes décrites dansRôle d'exécution Amazon MWAA. Toutefois, au lieu de créer et d'associer les politiques décrites dans cette rubrique, associez la politique suivante :


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "airflow:PublishMetrics",
            "Resource": "arn:aws:airflow:${MWAA_REGION}:${ACCOUNT_NUMBER}:environment/${MWAA_ENV_NAME}"
        },
        {
            "Effect": "Deny",
            "Action": "s3:ListAllMyBuckets",
            "Resource": [
                "arn:aws:s3:::{MWAA_S3_BUCKET}",
                "arn:aws:s3:::{MWAA_S3_BUCKET}/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject*",
                "s3:GetBucket*",
                "s3:List*"
            ],
            "Resource": [
                "arn:aws:s3:::{MWAA_S3_BUCKET}",
                "arn:aws:s3:::{MWAA_S3_BUCKET}/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:CreateLogGroup",
                "logs:PutLogEvents",
                "logs:GetLogEvents",
                "logs:GetLogRecord",
                "logs:GetLogGroupFields",
                "logs:GetQueryResults",
                "logs:DescribeLogGroups"
            ],
            "Resource": [
                "arn:aws:logs:${MWAA_REGION}:${ACCOUNT_NUMBER}:log-group:airflow-${MWAA_ENV_NAME}-*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": "cloudwatch:PutMetricData",
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "sqs:ChangeMessageVisibility",
                "sqs:DeleteMessage",
                "sqs:GetQueueAttributes",
                "sqs:GetQueueUrl",
                "sqs:ReceiveMessage",
                "sqs:SendMessage"
            ],
            "Resource": "arn:aws:sqs:${MWAA_REGION}:*:airflow-celery-*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kms:Decrypt",
                "kms:DescribeKey",
                "kms:GenerateDataKey*",
                "kms:Encrypt"
            ],
            "NotResource": "arn:aws:kms:*:${ACCOUNT_NUMBER}:key/*",
            "Condition": {
                "StringLike": {
                    "kms:ViaService": [
                        "sqs.${MWAA_REGION}.amazonaws.com"
                    ]
                }
            }
        },
        {
            "Effect": "Allow",
            "Action": [
                "eks:DescribeCluster"
            ],
            "Resource": "arn:aws:eks:${MWAA_REGION}:${ACCOUNT_NUMBER}:cluster/${EKS_CLUSTER_NAME}"
        }     
    ]
}

Après avoir créé le rôle, modifiez votre environnement Amazon MWAA pour utiliser le rôle que vous avez créé comme rôle d'exécution pour l'environnement. Pour modifier le rôle, modifiez l'environnement à utiliser. Vous sélectionnez le rôle d'exécution sousAutorisations.

Problèmes connus :

Il existe un problème connu lié aux rôles ARN dont les sous-chemins ne peuvent pas s'authentifier auprès d'Amazon EKS. La solution consiste à créer le rôle de service manuellement plutôt que d'utiliser celui créé par Amazon MWAA lui-même. Pour en savoir plus, voirLes rôles avec des chemins ne fonctionnent pas lorsque le chemin est inclus dans leur ARN dans le configmap aws-auth
Si la liste des services Amazon MWAA n'est pas disponible dans IAM, vous devez choisir une autre politique de service, telle qu'Amazon EC2, puis mettre à jour la politique de confiance du rôle pour qu'elle corresponde à ce qui suit :
```
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": [
          "airflow-env.amazonaws.com",
          "airflow.amazonaws.com"
        ]
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
```
Pour en savoir plus, voirComment utiliser les politiques de confiance avec les rôles IAM.

Créez le fichier requirements.txt

Pour utiliser l'exemple de code de cette section, assurez-vous d'avoir ajouté l'une des options de base de données suivantes à votrerequirements.txt. Pour en savoir plus, consultez Installation des dépendances Python.

Création d'un mappage d'identité pour Amazon EKS

Utilisez l'ARN correspondant au rôle que vous avez créé dans la commande suivante afin de créer un mappage d'identité pour Amazon EKS. Changer de régionvotre régionvers la région dans laquelle vous avez créé l'environnement. Remplacez l'ARN pour le rôle, et enfin, remplacezmwaa-execution-roleavec le rôle d'exécution de votre environnement.


eksctl create iamidentitymapping \
--region your-region \
--cluster mwaa-eks \
--arn arn:aws:iam::111222333444:role/mwaa-execution-role \
--username mwaa-service

Créer le `kubeconfig`

Utilisez la commande suivante pour créer lekubeconfig:


aws eks update-kubeconfig \
--region us-west-2 \
--kubeconfig ./kube_config.yaml \
--name mwaa-eks \
--alias aws

Si vous avez utilisé un profil spécifique lorsque vous avez couruupdate-kubeconfigvous devez supprimer leenv:section ajoutée au fichier kube_config.yaml afin qu'il fonctionne correctement avec Amazon MWAA. Pour ce faire, supprimez les éléments suivants du fichier, puis enregistrez-le :


env:
- name: AWS_PROFILE
  value: profile_name

Création d'un DAG

Utilisez l'exemple de code suivant pour créer un fichier Python, tel quemwaa_pod_example.pypour le DAG.

Apache Airflow v2


"""
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""
from airflow import DAG
from datetime import datetime
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

default_args = {
   'owner': 'aws',
   'depends_on_past': False,
   'start_date': datetime(2019, 2, 20),
   'provide_context': True
}

dag = DAG(
   'kubernetes_pod_example', default_args=default_args, schedule_interval=None)

#use a kube_config stored in s3 dags folder for now
kube_config_path = '/usr/local/airflow/dags/kube_config.yaml'

podRun = KubernetesPodOperator(
                       namespace="mwaa",
                       image="ubuntu:18.04",
                       cmds=["bash"],
                       arguments=["-c", "ls"],
                       labels={"foo": "bar"},
                       name="mwaa-pod-test",
                       task_id="pod-task",
                       get_logs=True,
                       dag=dag,
                       is_delete_operator_pod=False,
                       config_file=kube_config_path,
                       in_cluster=False,
                       cluster_context='aws'
                       )

Apache Airflow v1


"""
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""
from airflow import DAG
from datetime import datetime
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator

default_args = {
   'owner': 'aws',
   'depends_on_past': False,
   'start_date': datetime(2019, 2, 20),
   'provide_context': True
}

dag = DAG(
   'kubernetes_pod_example', default_args=default_args, schedule_interval=None)

#use a kube_config stored in s3 dags folder for now
kube_config_path = '/usr/local/airflow/dags/kube_config.yaml'

podRun = KubernetesPodOperator(
                       namespace="mwaa",
                       image="ubuntu:18.04",
                       cmds=["bash"],
                       arguments=["-c", "ls"],
                       labels={"foo": "bar"},
                       name="mwaa-pod-test",
                       task_id="pod-task",
                       get_logs=True,
                       dag=dag,
                       is_delete_operator_pod=False,
                       config_file=kube_config_path,
                       in_cluster=False,
                       cluster_context='aws'
                       )

Ajoutez le DAG et`kube_config.yaml`vers le compartiment Amazon S3

Mettez le DAG que vous avez créé et lekube_config.yamlfichier dans le compartiment Amazon S3 pour l'environnement Amazon MWAA. Vous pouvez placer des fichiers dans votre compartiment à l'aide de la console Amazon S3 ou duAWS Command Line Interface.

Activer et déclencher l'exemple

Dans Apache Airflow, activez l'exemple puis déclenchez-le.

Une fois qu'il s'est exécuté et s'est terminé correctement, utilisez la commande suivante pour vérifier le pod :


kubectl get pods -n mwaa

Vous devez voir des résultats similaires à ce qui suit :


NAME READY STATUS RESTARTS AGE
mwaa-pod-test-aa11bb22cc3344445555666677778888 0/1 Completed 0 2m23s

Vous pouvez ensuite vérifier la sortie du pod à l'aide de la commande suivante. Remplacez la valeur du nom par la valeur renvoyée par la commande précédente :


kubectl logs -n mwaa mwaa-pod-test-aa11bb22cc3344445555666677778888

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

Intégration avec Amazon EMR

Utilisation du ECSOperator