Comment puis-je résoudre les erreurs que je reçois lorsque je déclenche une étape Amazon EMR avec Amazon MWAA ?

Brève description

Lorsque vous déclenchez une étape Amazon EMR avec Amazon MWAA, les erreurs suivantes peuvent s'afficher :

« An error occurred (AccessDeniedException) when calling the DescribeCluster operation »

« An error occurred (AccessDeniedException) when calling the AddJobFlowSteps operation »

« EMR endpoint is not reachable: botocore.exceptions.ConnectTimeoutError: Connect timeout on endpoint URL »

« An error occurred (InvalidRequestException) when calling the DescribeCluster operation: Cluster id is not valid. »

« An error occurred (ValidationException) when calling the AddJobFlowSteps operation: Maximum number of active steps(State = 'Running', 'Pending' or 'Cancel_Pending') for cluster exceeded. »

« An error occurred (ValidationException) when calling the AddJobFlowSteps operation: jobFlowId is not valid. »

« An error occurred (ValidationException) when calling the AddJobFlowSteps operation: A job flow that is shutting down, terminated, or finished may not be modified. »

« No module named 'airflow.providers.amazon.aws.operators.emr_add_steps »

Résolution

Remarque : si des erreurs surviennent lorsque vous exécutez des commandes de l’interface de la ligne de commande AWS (AWS CLI), consultez la page Résoudre les erreurs liées à AWS CLI. Vérifiez également que vous utilisez la version la plus récente de l'AWS CLI.

Une erreur AccessDeniedException s'est produite lors d'un appel d'API Amazon EMR

Ces erreurs se produisent lorsque le rôle d'exécution Amazon MWAA rencontre des problèmes d'autorisation liés aux actions Amazon EMR. Pour résoudre ces erreurs, assurez-vous que le rôle d'exécution Amazon MWAA dispose des autorisations DescribeCluster et AddJobFlowSteps.

Exemple :

{
"Version": "2012-10-17",
"Statement": [
    {
        "Sid": "EMRStepPermissions",
        "Effect": "Allow",
        "Action": [
                "elasticmapreduce:DescribeCluster",
                "elasticmapreduce:AddJobFlowSteps"
                ],
        "Resource": "arn:aws:elasticmapreduce:example-region:example-account-id:cluster/example-cluster-id"
    }
    ]
}

Remarque : Remplacez example-region par votre région AWS, example-account-id par votre ID de compte et example-cluster-id par le nom de votre ID de cluster.

EMR endpoint is not reachable: botocore.exceptions.ConnectTimeoutError: Connect timeout on endpoint URL

Cette erreur se produit lorsque l'environnement Amazon MWAA ne parvient pas à atteindre le cluster Amazon EMR en raison d'une mauvaise configuration réseau. Pour résoudre cette erreur, assurez-vous que la configuration Amazon MWAA Amazon Virtual Private Cloud (Amazon VPC) autorise le trafic sortant vers le cluster.

Pour le routage public, assurez-vous que les groupes de sécurité et le sous-réseau disposent d'une règle et d'un itinéraire pour autoriser le trafic sortant vers le cluster. Pour le routage privé, assurez-vous qu'un point de terminaison Amazon EMR est associé aux sous-réseaux et au groupe de sécurité de l'environnement Amazon MWAA.

Si Amazon EMR se trouve dans un autre Amazon VPC, utilisez l’appairage Amazon VPC pour établir une connectivité réseau appropriée.

« An error occurred (InvalidRequestException) when calling the CreateAgent operation: Cluster id is not valid »

Pour résoudre cette erreur, vérifiez que l'ID de cluster est présent dans la même région et dans le même compte que ceux pour lesquels l'environnement Amazon MWAA est configuré. De plus, pour permettre aux utilisateurs d'AWS Identity and Access Management (IAM) de vérifier les détails du cluster, définissez VisibleToAllUsers sur Vrai dans le paramètre JOB_FLOW_OVERRIDES. Pour plus d'informations, consultez la section Paramètres de requête.

Pour vérifier les autorisations requises pour que le rôle d'exécution Amazon MWAA répertorie le cluster, exécutez la commande describe-cluster.

Pour en savoir plus sur votre cluster Amazon EMR, consultez la section Afficher le statut et les détails du cluster Amazon EMR.

Une erreur s'est produite (ValidationException) lors de l'appel de l'opération AddJobFlowSteps : jobFlowId <'some-string'> n'est pas valide

Cette erreur se produit lorsque job_flow_id est transmis dans EMRAddStepsOperator. Pour résoudre ce problème, assurez-vous d'utiliser l'ID de cluster Amazon EMR comme job_flow_id. Pour plus d'informations, consultez la section Paramètres dans la documentation Apache Airflow.

Une erreur s'est produite (ValidationException) lors de l'appel de l'opération AddJobflowSteps : Un flux de travail qui s'arrête, est résilié ou est terminé ne peut pas être modifié

Cette erreur se produit lorsque l'opération AddJobFlowSteps ne peut pas être effectuée car le flux de travail Amazon EMR est arrêté, résilié ou terminé. Pour résoudre cette erreur, utilisez AWS CloudTrail pour vérifier quand le cluster a été arrêté et quand la tâche EmrAddStepsOperator a démarré dans les journaux de tâche. Si la commande de fin du cluster a été déclenchée avant la tâche de démarrage du cluster, l'erreur se produit.

Pour résoudre cette erreur, ajoutez KeepJobFlowAliveWhenNoSteps comme Vrai dans le paramètre JOB_FLOW_OVERRIDES du code DAG. Ce paramètre fait passer le cluster à l’état En attente une fois les étapes terminées.

Exemple :

JOB_FLOW_OVERRIDES = {
    "Name": "Data-Pipeline-" + execution_date,
    "ReleaseLabel": "emr-5.29.0",
    "LogUri": "s3://{}/logs/emr/".format(S3_BUCKET_NAME),
    "Instances": {
        "InstanceGroups": [
            {
                "Name": "Master nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 1
            },
            {
                "Name": "Slave nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "CORE",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 2
            }
        ],
        "TerminationProtected": False,
        "KeepJobFlowAliveWhenNoSteps": True
    }
}

Pour arrêter le cluster une fois les étapes terminées, appliquez l'une des méthodes suivantes :

Utilisez EmrTerminateJobFlowOperator pour créer une tâche en tant que dernière tâche du DAG qui résilie le cluster une fois les étapes terminées :

cluster_remover = EmrTerminateJobFlowOperator(
        task_id='remove_cluster',
        job_flow_id=cluster_creator.output,
        aws_conn_id='aws_default',
    )

Vous pouvez également définir une stratégie de terminaison automatique dans JOB_FLOW_OVERRIDES qui résilie le cluster après un temps d'inactivité spécifié. L'exemple de configuration suivant met fin au cluster après 60 minutes d'inactivité :

JOB_FLOW_OVERRIDES = {
    "Name": "Data-Pipeline-" + execution_date,
    "ReleaseLabel": "emr-<version>",
    "LogUri": "s3://{}/logs/emr/".format(example-s3-bucket-name),
    "Instances": {
        "InstanceGroups": [
            {
                "Name": "Master nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 1
            },
            {
                "Name": "Slave nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "CORE",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 2
            }
        ],
        "TerminationProtected": False,
        "KeepJobFlowAliveWhenNoSteps": True
    },
    "AutoTerminationPolicy": {"IdleTimeout": 3600}
}

Pour plus d'informations, consultez la section Définition des tâches sur le site Web d'Apache Airflow.

Pour afficher d'autres JOB_FLOW_OVERRIDES, consultez la section run_job_flow.

Pour plus d'informations sur la résiliation automatique d'Amazon EMR, consultez la section Utilisation d'une stratégie de résiliation automatique pour le nettoyage du cluster Amazon EMR.

« No module named 'airflow.providers.amazon.aws.operators.emr_add_steps »

Si vous souhaitez que votre environnement Amazon MWAA et votre cluster Amazon EMR se trouvent dans des régions différentes, créez un autre ID de connexion. Assurez-vous d'utiliser la région AWS dans laquelle vous souhaitez lancer le cluster. Si vous souhaitez modifier votre région par défaut, procédez comme suit :

1. Ouvrez l'interface utilisateur d'Apache Airflow.
2. Dans le volet de navigation supérieur, sélectionnez Administrateur, puis Connexions, puis cliquez sur le bouton +.
3. Saisissez un nom pour l'ID de connexion que vous souhaitez utiliser dans le DAG.
4. Sélectionnez Service Amazon Web Services comme Type de connexion.
5. Dans le champ de texte, ajoutez le code JSON suivant :

   {"region_name": "example-region"}

   Remarque : Remplacez example-region par la région que vous souhaitez utiliser comme région par défaut.
6. Assurez-vous d'utiliser votre nouvel ID de connexion pour chaque opérateur Amazon EMR utilisé dans Amazon MWAA.

Erreurs d'importation DAG

Si vous recevez des erreurs d'importation DAG, cela signifie que le package requis n'est pas installé, que le chemin d'importation est différent ou que le nom de l'opérateur est incorrect. Pour résoudre ces erreurs, vérifiez quelle version du package apache-airflow-providers-amazon est installée dans votre environnement Amazon MWAA. Pour ce faire, consultez la page du fournisseur d'interface utilisateur Airflow. Vous pouvez également consulter les packages du fournisseur Apache Airflow installés sur les environnements Amazon MWAA. Puis, vérifiez le chemin d'importation et le nom de l'opérateur dans la documentation du fournisseur Airflow en fonction de la version du package du fournisseur. Par exemple, Amazon MWAA 2.2.2 utilise apache-airflow-providers-amazon 2.4.0. Pour vérifier le chemin d'importation et le nom de l'opérateur pour la version 2.4.0, consultez la page airflow.providers.amazon.aws sur le site Web d'Apache Airflow.

Les chemins d'importation pour chaque opérateur Amazon EMR sont les suivants :

• EmrAddStepsOperator : airflow.providers.amazon.aws.operators.emr_add_steps
• EmrCreateJobFlowOperator : airflow.providers.amazon.aws.operators.emr_create_job_flow
• EmrTerminateJobFlowOperator : airflow.providers.amazon.aws.operators.emr_terminate_job_flow

Remarque : Pour Amazon MWAA 2.8.1 avec apache-airflow-providers-amazon 8.16.0, tous les opérateurs Amazon EMR sont déplacés vers une classe Amazon EMR commune. Pour plus d'informations, consultez la page airflow.providers.amazon.aws.operators.emr sur le site Web d'Apache Airflow.

Pour inclure un chemin pour n'importe quel opérateur Amazon EMR, importez-le depuis airflow.providers.amazon.aws.operators.emr :

from airflow.providers.amazon.aws.operators.emr import EmrAddStepsOperator,
EmrCreateJobFlowOperator, EmrTerminateJobFlowOperator

Remarque : Assurez-vous de modifier les instructions d'importation du DAG en fonction de votre cas d'utilisation.