本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
为 Apache Airflow PythonVirtualenvOperator 创建自定义插件
以下示例显示了如何在 Amazon MWAA 上使用自定义插件修补 Apache Airflow PythonVirtualenvOperator。
版本
先决条件
要使用本页上的示例代码,您需要以下内容:
权限
要求
要使用本页上的示例代码,请将以下依赖项添加到 requirements.txt
。要了解更多信息,请参阅 安装 Python 依赖项。
virtualenv
自定义插件示例代码
Apache Airflow 将在启动时执行插件文件夹中的 Python 文件内容。此插件将在启动过程中修补内置的 PythonVirtualenvOperator
,使其与 Amazon MWAA 兼容。以下步骤介绍了此自定义插件的示例代码。
- Apache Airflow v2
-
-
在命令行提示符中,导航到以上 plugins
目录。例如:
cd plugins
-
复制以下代码示例的内容,并在本地另存为 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'
- Apache Airflow v1
-
-
在命令行提示符中,导航到以上 plugins
目录。例如:
cd plugins
-
复制以下代码示例的内容,并在本地另存为 virtual_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
以下步骤显示如何创建 plugins.zip
。
-
在命令行提示符中,导航到包含以上 virtual_python_plugin.py
的目录。例如:
cd plugins
-
将内容压缩到 plugins
文件夹中。
zip plugins.zip virtual_python_plugin.py
代码示例
以下步骤介绍了如何创建自定义插件的 DAG 代码。
- Apache Airflow v2
-
-
在命令提示符下,导航到存储 DAG 代码的目录。例如:
cd dags
-
复制以下代码示例的内容,并在本地另存为 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,
)
- Apache Airflow v1
-
-
在命令提示符下,导航到存储 DAG 代码的目录。例如:
cd dags
-
复制以下代码示例的内容,并在本地另存为 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_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,
)
Airflow 配置选项
如果您使用的是 Apache Airflow v2,请添加 core.lazy_load_plugins : False
为 Apache Airflow 配置选项。要了解更多信息,请参阅 2 中的使用配置选项加载插件。
接下来做什么?
-
要了解如何将本示例中的 requirements.txt
文件上传到 Amazon S3 存储桶,请参阅 安装 Python 依赖项。
-
要了解如何将本示例中的 DAGD 代码上传到 Amazon S3 存储桶的 dags
文件夹,请参阅 添加或更新 DAG。
-
要了解如何将本示例中的 plugins.zip
文件上传到 Amazon S3 存储桶,请参阅 安装自定义插件。