Version Prérequis Autorisations Prérequis Exemple de code de plugin personnalisé Plugins.zip Exemple de code Options de configuration du débit d'air Quelle est la prochaine étape ?

Création d'un plugin personnalisé pour Apache AirflowPythonVirtualenvOperator

L'exemple suivant montre comment appliquer un correctif à Apache AirflowPythonVirtualenvOperatoravec un plugin personnalisé sur Amazon Managed Workflows pour Apache Airflow.

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 de code de cette page, vous aurez besoin des éléments suivants :

UnEnvironnement Amazon MWA.

Autorisations

Aucune autorisation supplémentaire n'est requise pour utiliser l'exemple de code de cette page.

Prérequis

Pour utiliser l'exemple de code de cette page, ajoutez les dépendances suivantes à votrerequirements.txt. Pour en savoir plus, consultez Installation des dépendances Python.


virtualenv

Exemple de code de plugin personnalisé

Apache Airflow exécutera le contenu des fichiers Python dans le dossier plugins au démarrage. Ce plugin va patcher le plugin intégréPythonVirtualenvOperatorau cours de ce processus de démarrage pour le rendre compatible avec Amazon MWAA. Les étapes suivantes présentent l'exemple de code pour le plug-in personnalisé.

Apache Airflow v2

Dans votre invite de commandes, accédez aupluginsrépertoire ci-dessus. Par exemple :
```
cd plugins
```

Copiez le contenu de l'exemple de code suivant et enregistrez-le localement sousvirtual_python_plugin.py.


"""
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.plugins_manager import AirflowPlugin
import airflow.utils.python_virtualenv 
from typing import List

def _generate_virtualenv_cmd(tmp_dir: str, python_bin: str, system_site_packages: bool) -> List[str]:
    cmd = ['python3','/usr/local/airflow/.local/lib/python3.7/site-packages/virtualenv', tmp_dir]
    if system_site_packages:
        cmd.append('--system-site-packages')
    if python_bin is not None:
        cmd.append(f'--python={python_bin}')
    return cmd

airflow.utils.python_virtualenv._generate_virtualenv_cmd=_generate_virtualenv_cmd

class VirtualPythonPlugin(AirflowPlugin):                
    name = 'virtual_python_plugin'

Apache Airflow v1

Dans votre invite de commandes, accédez aupluginsrépertoire ci-dessus. Par exemple :
```
cd plugins
```

Copiez le contenu de l'exemple de code suivant et enregistrez-le localement sousvirtual_python_plugin.py.


from airflow.plugins_manager import AirflowPlugin
from airflow.operators.python_operator import PythonVirtualenvOperator

def _generate_virtualenv_cmd(self, tmp_dir):
    cmd = ['python3','/usr/local/airflow/.local/lib/python3.7/site-packages/virtualenv', tmp_dir]
    if self.system_site_packages:
        cmd.append('--system-site-packages')
    if self.python_version is not None:
        cmd.append('--python=python{}'.format(self.python_version))
    return cmd
PythonVirtualenvOperator._generate_virtualenv_cmd=_generate_virtualenv_cmd

class EnvVarPlugin(AirflowPlugin):                
    name = 'virtual_python_plugin'

Plugins.zip

Les étapes suivantes montrent comment créerplugins.zip.

Dans votre invite de commandes, accédez au répertoire contenantvirtual_python_plugin.pyci-dessus. Par exemple :
```
cd plugins
```
Compressez le contenu dans votrepluginsdossier.
```
zip plugins.zip virtual_python_plugin.py
```

Exemple de code

Les étapes suivantes décrivent comment créer le code DAG pour le plug-in personnalisé.

Apache Airflow v2

Dans votre invite de commandes, 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 sousvirtualenv_test.py.


"""
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 airflow.operators.python import PythonVirtualenvOperator
from airflow.utils.dates import days_ago
import os

os.environ["PATH"] = os.getenv("PATH") + ":/usr/local/airflow/.local/bin"

def virtualenv_fn():
    import boto3
    print("boto3 version ",boto3.__version__)

with DAG(dag_id="virtualenv_test", schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
    virtualenv_task = PythonVirtualenvOperator(
        task_id="virtualenv_task",
        python_callable=virtualenv_fn,
        requirements=["boto3>=1.17.43"],
        system_site_packages=False,
        dag=dag,
    )

Apache Airflow v1

Dans votre invite de commandes, 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 sousvirtualenv_test.py.


"""
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 airflow.operators.python_operator import PythonVirtualenvOperator
from airflow.utils.dates import days_ago
import os

os.environ["PATH"] = os.getenv("PATH") + ":/usr/local/airflow/.local/bin"

def virtualenv_fn():
    import boto3
    print("boto3 version ",boto3.__version__)

with DAG(dag_id="virtualenv_test", schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
    virtualenv_task = PythonVirtualenvOperator(
        task_id="virtualenv_task",
        python_callable=virtualenv_fn,
        requirements=["boto3>=1.17.43"],
        system_site_packages=False,
        dag=dag,
    )

Options de configuration du débit d'air

Si vous utilisez Apache Airflow v2, ajoutezcore.lazy_load_plugins : Falseen tant qu'option de configuration d'Apache Airflow. Pour en savoir plus, consultezUtiliser les options de configuration pour charger des plugins en 2.

Quelle est la prochaine étape ?

Découvrez comment charger lerequirements.txtdans cet exemple, dans votre compartiment Amazon S3 dansInstallation des dépendances Python.
Découvrez comment charger le code DAG de cet exemple vers ledagsdossier dans votre compartiment Amazon S3 dansAjout ou mise à jour des DAG.
En savoir plus sur la façon de télécharger leplugins.zipdans cet exemple, dans votre compartiment Amazon S3 dansInstallation de plugins personnalisés.

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

Plugin personnalisé avec Apache Hive et Hadoop

Invoquer des DAG avec Lambda