Résolution des problèmes : opérateurs DAGs, connexions et autres problèmes

Découvrez comment créer des clés secrètes pour votre connexion Apache Airflow et des variables dans. Configuration d'une connexion Apache Airflow à l'aide d'un secret AWS Secrets Manager

Apprenez à utiliser la clé secrète d'une variable Apache Airflow (test-variable) dans. Utilisation d'une clé secrète AWS Secrets Manager pour une variable Apache Airflow

Apprenez à utiliser la clé secrète pour une connexion Apache Airflow (myconn) dans. Utilisation d'une clé secrète AWS Secrets Manager pour une connexion Apache Airflow

Testez vos DAGs plugins personnalisés et vos dépendances Python localement à l'aide de aws-mwaa-docker-imageson GitHub.

Ajoutez les entrées suivantes au fichier requirements.txt correspondant à votre environnement.

apache-airflow-providers-snowflake==1.3.0

Ajoutez les importations suivantes à votre DAG :

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

Identifiant du Connecticut : snowflake_conn

Type de connecteur : flocon de neige

Hôte :<my account>. <my region if not us-west-2>.snowflakecomputing.com

Schéma : <my schema>

Identifiant : <my user name>

Mot de passe : ********

Port : <port, if any>

Supplémentaire :

{
						"account": "<my account>",
						"warehouse": "<my warehouse>",
						"database": "<my database>",
						"region": "<my region if not using us-west-2 otherwise omit this line>"
						}

Accédez aux types de connexion fournis par Amazon MWAA dans l'interface utilisateur d'Apache Airflow à l'adresse. Packages du fournisseur Apache Airflow installés sur les environnements Amazon MWAA

Accédez aux commandes pour créer une connexion Apache Airflow dans la CLI à l'adresse. Référence des commandes de la CLI Apache Airflow

Découvrez comment utiliser les modèles de connexion dans l'interface utilisateur d'Apache Airflow de manière interchangeable pour les types de connexion qui ne sont pas disponibles dans l'interface utilisateur d'Apache Airflow sur Amazon MWAA à l'adresse. Vue d'ensemble des types de connexion

Vérifiez les options de configuration d'Apache Airflow. Vérifiez que les paires clé-valeur que vous avez spécifiées en tant qu'option de configuration d'Apache Airflow, par exemple, ont été correctement AWS Secrets Manager configurées. Pour en savoir plus, reportez-vous àJe n'arrive pas à me connecter à Secrets Manager.

Vérifiez lerequirements.txt. Vérifiez que le package « extras » Airflow et les autres bibliothèques répertoriées dans le vôtre requirements.txt sont compatibles avec votre version d'Apache Airflow.

Découvrez comment spécifier les dépendances Python dans un requirements.txt fichier, reportez-vous àGestion des dépendances Python dans requirements.txt.

Vérifiez que votre groupe de sécurité VPC autorise l'accès entrant au port. 5432 Ce port est nécessaire pour se connecter à la base de données de métadonnées Amazon Aurora PostgreSQL de votre environnement. Une fois cette règle ajoutée, accordez quelques minutes à Amazon MWAA pour que l'erreur disparaisse. Pour en savoir plus, reportez-vous àSécurité de votre VPC sur Amazon MWAA.

Si le planificateur n'est pas en cours d'exécution, cela peut être dû à un certain nombre de facteurs tels que l'échec de l'installation des dépendances ou un planificateur surchargé. Vérifiez que vos DAGs plug-ins et vos exigences fonctionnent correctement en accédant aux groupes de CloudWatch journaux correspondants dans Logs. Pour en savoir plus, reportez-vous àSurveillance et métriques pour Amazon Managed Workflows pour Apache Airflow.

Il peut y avoir un grand nombre de DAGs définitions. Réduisez le nombre d'environnements DAGs et mettez-les à jour (par exemple en modifiant le niveau d'un journal) pour forcer une réinitialisation.

1. Airflow analyse DAGs s'ils sont activés ou non. Si vous utilisez plus de 50 % de la capacité de votre environnement, vous risquez de surcharger le planificateur Apache Airflow. Cela entraîne un temps d'analyse total important dans les CloudWatch métriques ou de longs délais de traitement des DAG dans CloudWatch les journaux. Il existe d'autres moyens d'optimiser les configurations d'Apache Airflow qui sortent du cadre de ce guide.

2. Pour en savoir plus sur les meilleures pratiques que nous recommandons pour optimiser les performances de votre environnement, reportez-vous àOptimisation des performances pour Apache Airflow sur Amazon MWAA.

Il se peut que la file d'attente contienne un grand nombre de tâches. Cela se traduit souvent par un nombre important (et croissant) de tâches dans l'NoneÉtat, ou par un nombre important de tâches dans l'État. Queued Tasks and/or Tasks Pending CloudWatch Cela peut se produire pour les raisons suivantes :

1. S'il y a plus de tâches à exécuter que l'environnement n'en a la capacité, and/or un grand nombre de tâches mises en file d'attente avant le dimensionnement automatique ont le temps de les détecter et de déployer du personnel supplémentaire.

2. S'il y a plus de tâches à exécuter qu'un environnement n'en a la capacité, nous vous recommandons de réduire le nombre de tâches que vous DAGs exécutez simultanément, en and/or augmentant le nombre minimum de travailleurs Apache Airflow.

3. Si un grand nombre de tâches ont été mises en file d'attente avant que le scalage automatique n'ait eu le temps de détecter et de déployer des travailleurs supplémentaires, nous vous recommandons d'échelonner le déploiement des tâches en and/or augmentant le nombre minimum de travailleurs Apache Airflow.

4. Vous pouvez utiliser la commande update-environment dans le AWS Command Line Interface (AWS CLI) pour modifier le nombre minimum ou maximum de travailleurs exécutés sur votre environnement.

   aws mwaa update-environment --name MyEnvironmentName --min-workers 2 --max-workers 10

5. Pour en savoir plus sur les meilleures pratiques que nous recommandons pour optimiser les performances de votre environnement, reportez-vous àOptimisation des performances pour Apache Airflow sur Amazon MWAA.

Si vos tâches sont bloquées dans l'état « en cours », vous pouvez également les effacer ou les marquer comme réussies ou échouées. Cela permet au composant de mise à l'échelle automatique de votre environnement de réduire le nombre de travailleurs travaillant sur votre environnement. L'image suivante illustre un exemple de tâche bloquée.

1. Choisissez le cercle pour la tâche bloquée, puis sélectionnez Effacer (comme indiqué). Cela permet à Amazon MWAA de réduire le nombre de travailleurs ; dans le cas contraire, Amazon MWAA ne peut pas déterminer quelles tâches DAGs sont activées ou désactivées, et ne peut pas réduire ses effectifs s'il y a encore des tâches en file d'attente.

   

Pour en savoir plus sur le cycle de vie des tâches d'Apache Airflow, consultez Concepts dans le guide de référence d'Apache Airflow.

Testez vos DAGs plugins personnalisés et vos dépendances Python localement à l'aide de aws-mwaa-docker-imageson GitHub.

Créez votre script de « transformation ».

#!/bin/bash
cp $1 $2

(facultatif) Les utilisateurs de macOS et Linux devront peut-être exécuter la commande suivante pour s'assurer que le script est exécutable.

chmod 777 transform_test.sh

Ajoutez le script à votre fichier plugins.zip.

zip plugins.zip transform_test.sh

Suivez les étapes décrites dans Importer le fichier plugins.zip sur Amazon S3.

Suivez les étapes décrites dans Spécifier la version du fichier plugins.zip sur la console Amazon MWAA.

Créez le DAG suivant.

from airflow import DAG
						from airflow.providers.amazon.aws.operators.s3_file_transform import S3FileTransformOperator
						from airflow.utils.dates import days_ago
						import os
						
						DAG_ID = os.path.basename(__file__).replace(".py", "")
						
						with DAG (dag_id=DAG_ID, schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
						file_transform = S3FileTransformOperator(
						task_id='file_transform',
						transform_script='/usr/local/airflow/plugins/transform_test.sh',
						source_s3_key='s3://amzn-s3-demo-bucket/files/input.txt',
						dest_s3_key='s3://amzn-s3-demo-bucket/files/output.txt'
						)

Suivez les étapes décrites dans la section Chargement du code DAG sur Amazon S3.