本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

# 为 Apache Airflow PythonVirtualenvOperator 创建自定义插件
修补 PythonVirtualenvOperator 的自定义插件

以下示例说明了如何在 Amazon Managed Workflows for Apache Airflow 上使用自定义插件修补 Apache Airflow `PythonVirtualenvOperator`。

**Topics**
+ [

## 版本
](#samples-virtualenv-version)
+ [

## 先决条件
](#samples-virtualenv-prereqs)
+ [

## 权限
](#samples-virtualenv-permissions)
+ [

## 要求
](#samples-virtualenv-dependencies)
+ [

## 自定义插件示例代码
](#samples-virtualenv-plugins-code)
+ [

## Plugins.zip
](#samples-virtualenv-pluginszip)
+ [

## 代码示例
](#samples-virtualenv-code)
+ [

## Airflow 配置选项
](#samples-virtualenv-airflow-config)
+ [

## 接下来做什么？
](#samples-virtualenv-next-up)

## 版本


您可以在 [Python 3.10](https://peps.python.org/pep-0619/) 中将本页上的代码示例与 **Apache Airflow v2** 一起使用，在 [Python 3.11](https://peps.python.org/pep-0664/) 中与 **Apache Airflow v3** 一起使用。

## 先决条件


要使用本页上的示例代码，您需要以下内容：
+ [Amazon MWAA 环境。](get-started.md)

## 权限


无需其他权限即可使用本页上的代码示例。

## 要求


要使用本页上的示例代码，请将以下依赖项添加到 `requirements.txt`。要了解更多信息，请参阅 [安装 Python 依赖项](working-dags-dependencies.md)。

```
virtualenv
```

## 自定义插件示例代码


Apache Airflow 将在启动时执行插件文件夹中的 Python 文件内容。此插件将在启动过程中修补内置的 `PythonVirtualenvOperator`，使其与 Amazon MWAA 兼容。以下步骤显示了此自定义插件的示例代码。

1. 在命令提示符下，导航到上一部分中的 `plugins` 目录。例如：

   ```
   cd plugins
   ```

1. 复制以下代码示例的内容，并在本地另存为 `virtual_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'
   ```

## Plugins.zip


以下步骤说明如何创建 `plugins.zip`。

1. 在命令提示符下，导航到上一部分中包含 `virtual_python_plugin.py` 的目录。例如：

   ```
   cd plugins
   ```

1. 将内容压缩到 `plugins` 文件夹中。

   ```
   zip plugins.zip virtual_python_plugin.py
   ```

## 代码示例


以下步骤介绍了如何创建自定义插件的 DAG 代码。

1. 在命令提示符下，导航到存储 DAG 代码的目录。例如：

   ```
   cd dags
   ```

1. 复制以下代码示例的内容，并在本地另存为 `virtualenv_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,
       )
   ```

## Airflow 配置选项


如果您使用的是 Apache Airflow v2，请添加 `core.lazy_load_plugins : False` 为 Apache Airflow 配置选项。要了解更多信息，请参阅 [2 中的使用配置选项加载插件](configuring-env-variables.md#configuring-2.0-airflow-override)。

## 接下来做什么？

+ 要了解如何将本示例中的 `requirements.txt` 文件上传到 Amazon S3 存储桶，请参阅 [安装 Python 依赖项](working-dags-dependencies.md)。
+ 要了解如何将本示例中的 DAG 代码上传到 Amazon S3 存储桶的 `dags` 文件夹，请参阅 [添加或更新 DAG](configuring-dag-folder.md)。
+ 要了解如何将本示例中的 `plugins.zip` 文件上传到 Amazon S3 存储桶，请参阅 [安装自定义插件](configuring-dag-import-plugins.md)。