PythonOperator¶
使用 PythonOperator 来执行 Python 可调用对象。
提示
建议使用 @task 装饰器，而不是传统的 PythonOperator，来执行 Python 可调用对象。
@task(task_id="print_the_context")
def print_context(ds=None, **kwargs):
"""Print the Airflow context and ds variable from the context."""
pprint(kwargs)
print(ds)
return "Whatever you return gets printed in the logs"
run_this = print_context()
def print_context(ds=None, **kwargs):
"""Print the Airflow context and ds variable from the context."""
print("::group::All kwargs")
pprint(kwargs)
print("::endgroup::")
print("::group::Context variable ds")
print(ds)
print("::endgroup::")
return "Whatever you return gets printed in the logs"
run_this = PythonOperator(task_id="print_the_context", python_callable=print_context)
传入参数¶
像普通的 Python 函数一样，向使用 @task 装饰的函数传入额外参数。
# Generate 5 sleeping tasks, sleeping from 0.0 to 0.4 seconds respectively
@task
def my_sleeping_function(random_base):
"""This is a function that will run within the DAG execution"""
time.sleep(random_base)
for i in range(5):
sleeping_task = my_sleeping_function.override(task_id=f"sleep_for_{i}")(random_base=i / 10)
run_this >> log_the_sql >> sleeping_task
# Generate 5 sleeping tasks, sleeping from 0.0 to 0.4 seconds respectively
def my_sleeping_function(random_base):
"""This is a function that will run within the DAG execution"""
time.sleep(random_base)
for i in range(5):
sleeping_task = PythonOperator(
task_id=f"sleep_for_{i}", python_callable=my_sleeping_function, op_kwargs={"random_base": i / 10}
)
run_this >> log_the_sql >> sleeping_task
异步 Python 函数¶
在 3.2 版中添加。
现在已原生支持异步 Python 可调用对象。这意味着我们无需处理事件循环，且可以轻松调用异步 Python 代码和异步 Airflow hook，这些在延迟（deferred）操作符中并不总是可用。
不同于在触发器（triggerer）上执行的延迟操作符，异步操作符在工作节点（workers）上执行。
# Generate 5 sleeping tasks, sleeping from 0.0 to 0.4 seconds respectively
# Asynchronous callables are natively supported since Airflow 3.2+
@task
async def my_async_sleeping_function(random_base):
"""This is a function that will run within the DAG execution"""
await asyncio.sleep(random_base)
for i in range(5):
async_sleeping_task = my_async_sleeping_function.override(task_id=f"async_sleep_for_{i}")(
random_base=i / 10
)
run_this >> log_the_sql >> async_sleeping_task
# Generate 5 sleeping tasks, sleeping from 0.0 to 0.4 seconds respectively
# Asynchronous callables are natively supported since Airflow 3.2+
async def my_async_sleeping_function(random_base):
"""This is a function that will run within the DAG execution"""
await asyncio.sleep(random_base)
for i in range(5):
async_sleeping_task = PythonOperator(
task_id=f"async_sleep_for_{i}",
python_callable=my_async_sleeping_function,
op_kwargs={"random_base": i / 10},
)
run_this >> log_the_sql >> async_sleeping_task
模板化¶
Airflow 还会传入一组额外的关键字参数：针对每个 Jinja 模板变量 的参数以及 templates_dict 参数。
`templates_dict`、`op_args`、`op_kwargs` 参数会进行模板渲染，因此字典中的每个值都会被当作 Jinja 模板 求值。
@task(task_id="log_sql_query", templates_dict={"query": "sql/sample.sql"}, templates_exts=[".sql"])
def log_sql(**kwargs):
log.info("Python task decorator query: %s", str(kwargs["templates_dict"]["query"]))
log_the_sql = log_sql()
def log_sql(**kwargs):
log.info("Python task decorator query: %s", str(kwargs["templates_dict"]["query"]))
log_the_sql = PythonOperator(
task_id="log_sql_query",
python_callable=log_sql,
templates_dict={"query": "sql/sample.sql"},
templates_exts=[".sql"],
)
上下文¶
`Context` 是一个字典对象，包含关于 `DagRun` 环境的信息。例如，选择 `task_instance` 可获取当前运行的 `TaskInstance` 对象。
它可以隐式使用，例如通过 `**kwargs`，也可以显式调用 `get_current_context()`。在这种情况下，可使用类型提示进行静态分析。
PythonVirtualenvOperator¶
使用 PythonVirtualenvOperator 装饰器在新的 Python 虚拟环境中执行 Python 可调用对象。需要在运行 Airflow 的环境中安装 `virtualenv` 包（作为可选依赖 `pip install apache-airflow[virtualenv] --constraint ...`）。
此外，需要使用命令 `pip install [cloudpickle] --constraint ...` 安装 `cloudpickle` 包作为可选依赖。该包取代了当前使用的 `dill` 包。Cloudpickle 侧重于标准的 pickle 协议，具有更高的兼容性和更流畅的数据交换，同时仍能有效处理常见的 Python 对象和函数中的全局变量。
提示
建议使用 `@task.virtualenv` 装饰器，而不是传统的 `PythonVirtualenvOperator`，在新的 Python 虚拟环境中执行 Python 可调用对象。
@task.virtualenv(
task_id="virtualenv_python", requirements=["colorama==0.4.0"], system_site_packages=False
)
def callable_virtualenv():
"""
Example function that will be performed in a virtual environment.
Importing at the module level ensures that it will not attempt to import the
library before it is installed.
"""
from time import sleep
from colorama import Back, Fore, Style
print(Fore.RED + "some red text")
print(Back.GREEN + "and with a green background")
print(Style.DIM + "and in dim text")
print(Style.RESET_ALL)
for _ in range(4):
print(Style.DIM + "Please wait...", flush=True)
sleep(1)
print("Finished")
virtualenv_task = callable_virtualenv()
def callable_virtualenv():
"""
Example function that will be performed in a virtual environment.
Importing at the function level ensures that it will not attempt to import the
library before it is installed.
"""
from time import sleep
from colorama import Back, Fore, Style
print(Fore.RED + "some red text")
print(Back.GREEN + "and with a green background")
print(Style.DIM + "and in dim text")
print(Style.RESET_ALL)
for _ in range(4):
print(Style.DIM + "Please wait...", flush=True)
sleep(1)
print("Finished")
virtualenv_task = PythonVirtualenvOperator(
task_id="virtualenv_python",
python_callable=callable_virtualenv,
requirements=["colorama==0.4.0"],
system_site_packages=False,
)
传入参数¶
像普通的 Python 函数一样，向使用 `@task.virtualenv` 装饰的函数传入额外参数。不幸的是，Airflow 由于底层库的兼容性问题，不支持序列化 `var`、`ti` 和 `task_instance`。对于 Airflow 上下文变量，请确保通过将 `system_site_packages` 设置为 `True` 来访问 Airflow，否则在 `op_kwargs` 中将无法使用大多数上下文变量。如果需要与日期时间对象相关的上下文，例如 `data_interval_start`，可以添加 `pendulum` 和 `lazy_object_proxy`。
重要提示
当需要 Airflow 或提供者包时，必须使用 `pip_install_options` 指定 Airflow 的 约束文件，以避免依赖冲突。
重要提示
要执行的 Python 函数体会被从 DAG 中提取到一个临时文件中，不包含周围的代码。正如示例所示，需要重新添加所有导入，且无法依赖全局 Python 上下文中的变量。
如果想向传统的 PythonVirtualenvOperator 传递变量，请使用 `op_args` 和 `op_kwargs`。
如果需要额外的包安装参数，请通过 `pip_install_options` 参数传入，或使用如下示例中的 `requirements.txt`。
SomePackage==0.2.1 --pre --index-url http://some.archives.com/archives
AnotherPackage==1.4.3 --no-index --find-links /my/local/archives
所有支持的选项列在 requirements 文件格式 中。
模板化¶
Jinja 模板的使用方式与 PythonOperator 中描述的相同。
虚拟环境设置选项¶
虚拟环境基于工作节点上全局的 Python pip 配置创建。可在环境中使用额外的 ENVs，或按 pip 配置 中的说明进行通用 pip 配置调整。
如果希望使用任务特定的私有 Python 仓库来设置虚拟环境，可以传入 `index_urls` 参数，以调整 pip 安装配置。传入的索引 URL 会替代系统默认的索引 URL 设置。为避免在 DAG 代码中直接写入私有仓库的凭据，可使用 Airflow 的 连接与 Hook。此时可以使用连接类型 `Package Index (Python)`。在 `Package Index (Python)` 连接类型中，可指定私有仓库的索引 URL 和凭据。创建好 `Package Index (Python)` 连接后，可通过 `index_urls_from_connection_ids` 参数将连接 ID 传给 `PythonVirtualenvOperator`。`PythonVirtualenvOperator` 会自动将该连接中的索引 URL（包括凭据）追加到 pip 安装器的 `index_urls` 参数中。
列表中的第一个 `index_url` 将作为虚拟环境设置的主索引 URL（pip 使用 `index-url`，uv 使用 `default-index`）。其余 URL 将作为额外索引 URL 添加。如果同时提供 `index_urls` 和 `index_urls_from_connection_ids` 参数，则使用 `index_urls` 中的第一个 URL 作为主索引 URL，其余的作为额外索引 URL。
如果希望在虚拟环境设置时完全禁止远程调用，可将 `index_urls` 设为空列表 `index_urls=[]`，从而强制 pip 使用 `--no-index` 选项。
缓存与复用¶
虚拟环境的设置在每次任务执行时会在临时目录中创建。执行完毕后会删除该虚拟环境。请确保工作节点上的 `$tmp` 文件夹有足够的磁盘空间。通常情况下（若未另行配置），会使用本地 pip 缓存，从而避免每次执行都重新下载包。
但每次都重新创建虚拟环境仍然需要耗时。若需重复执行，可将 `venv_cache_path` 设为工作节点上的某个文件系统路径。这样虚拟环境只会创建一次并被复用。使用虚拟环境缓存时，会根据不同的依赖集合在缓存路径下创建不同的子文件夹。因此，根据系统中 DAG 的变化，需要预留足够的磁盘空间。
请注意，缓存模式下不会自动清理。所有 worker 插槽共享相同的虚拟环境，但如果任务在不同的 worker 上反复调度，可能会在多个 worker 上各自创建虚拟环境。另外，如果 worker 以 Kubernetes POD 方式启动，则重启后会丢失缓存（假设 `venv_cache_path` 未放在持久卷上）。
如果运行时出现损坏的缓存虚拟环境问题，可通过设置 Airflow 变量 `PythonVirtualenvOperator.cache_key` 为任意文本来影响缓存目录的哈希。该变量的内容会用于计算缓存目录键值。
请注意，对已缓存的虚拟环境的任何修改（例如二进制路径下的临时文件、后续安装额外依赖）都可能污染缓存，且该操作符不会维护或清理缓存路径。
ExternalPythonOperator¶
`ExternalPythonOperator` 可帮助您在某些任务中使用与其他任务（以及主 Airflow 环境）不同的 Python 库集合。这可以是一个虚拟环境或任何预装且在运行 Airflow 任务的环境中可用的 Python 安装。该操作符通过 `python` 参数指定 Python 可执行文件。需要注意的是，即使是虚拟环境，`python` 路径也应指向该虚拟环境内部的 Python 可执行文件（通常在虚拟环境的 `bin` 子目录下）。与常规使用虚拟环境不同，无需对环境进行 `activation`，直接使用 `python` 可执行文件即可自动激活。在以下两个示例中，`PATH_TO_PYTHON_BINARY` 就是指向可执行 Python 二进制文件的路径。
使用 ExternalPythonOperator 在预定义的环境中执行 Python 可调用对象。运行 Python 的环境中应预先安装 `virtualenv` 包。如果使用 `dill`，也必须在该环境中预装（版本应与主 Airflow 环境中安装的相同）。
提示
建议使用 `@task.external_python` 装饰器，而不是传统的 `ExternalPythonOperator`，在预定义的 Python 环境中执行代码。
@task.external_python(task_id="external_python", python=PATH_TO_PYTHON_BINARY)
def callable_external_python():
"""
Example function that will be performed in a virtual environment.
Importing at the module level ensures that it will not attempt to import the
library before it is installed.
"""
import sys
from time import sleep
print(f"Running task via {sys.executable}")
print("Sleeping")
for _ in range(4):
print("Please wait...", flush=True)
sleep(1)
print("Finished")
external_python_task = callable_external_python()
def callable_external_python():
"""
Example function that will be performed in a virtual environment.
Importing at the module level ensures that it will not attempt to import the
library before it is installed.
"""
import sys
from time import sleep
print(f"Running task via {sys.executable}")
print("Sleeping")
for _ in range(4):
print("Please wait...", flush=True)
sleep(1)
print("Finished")
external_python_task = ExternalPythonOperator(
task_id="external_python",
python_callable=callable_external_python,
python=PATH_TO_PYTHON_BINARY,
)
传入参数¶
像普通的 Python 函数一样，向使用 `@task.external_python` 装饰的函数传入额外参数。不幸的是，Airflow 由于底层库的兼容性问题，不支持序列化 `var` 与 `ti` / `task_instance`。对于 Airflow 上下文变量，请确保在虚拟环境中也安装了相同版本的 Airflow，否则在 `op_kwargs` 中将无法访问大多数上下文变量。如果需要与日期时间对象相关的上下文，例如 `data_interval_start`，可以在虚拟环境中加入 `pendulum` 与 `lazy_object_proxy`。
重要提示
要执行的 Python 函数体会从 DAG 中提取到一个临时文件中，不包含周围代码。正如示例所示，需要重新添加所有导入，且不能依赖全局 Python 上下文中的变量。
如果想向传统的 ExternalPythonOperator 传递变量，请使用 `op_args` 和 `op_kwargs`。
模板化¶
Jinja 模板的使用方式与 PythonOperator 中描述的相同。
BranchPythonOperator¶
使用 BranchPythonOperator 执行 Python 分支 任务。
提示
建议使用 `@task.branch` 装饰器，而不是传统的 `BranchPythonOperator`，来执行 Python 代码。
@task.branch()
def branching(choices: list[str]) -> str:
return f"branch_{random.choice(choices)}"
branching = BranchPythonOperator(
task_id="branching",
python_callable=lambda: f"branch_{random.choice(options)}",
)
传入参数和模板化¶
参数传递和模板化选项与 PythonOperator 相同。
BranchPythonVirtualenvOperator¶
使用 BranchPythonVirtualenvOperator 装饰器执行 Python 分支 任务，它是 BranchPythonOperator 的变体，在虚拟环境中执行。
提示
建议使用 `@task.branch_virtualenv` 装饰器，而不是传统的 `BranchPythonVirtualenvOperator`，来执行 Python 代码。
# Note: Passing a caching dir allows to keep the virtual environment over multiple runs
# Run the example a second time and see that it reuses it and is faster.
VENV_CACHE_PATH = tempfile.gettempdir()
@task.branch_virtualenv(requirements=["numpy~=1.26.0"], venv_cache_path=VENV_CACHE_PATH)
def branching_virtualenv(choices) -> str:
import random
import numpy as np
print(f"Some numpy stuff: {np.arange(6)}")
return f"venv_{random.choice(choices)}"
# Note: Passing a caching dir allows to keep the virtual environment over multiple runs
# Run the example a second time and see that it reuses it and is faster.
VENV_CACHE_PATH = Path(tempfile.gettempdir())
def branch_with_venv(choices):
import random
import numpy as np
print(f"Some numpy stuff: {np.arange(6)}")
return f"venv_{random.choice(choices)}"
branching_venv = BranchPythonVirtualenvOperator(
task_id="branching_venv",
requirements=["numpy~=1.26.0"],
venv_cache_path=VENV_CACHE_PATH,
python_callable=branch_with_venv,
op_args=[options],
)
传入参数和模板化¶
参数传递和模板化选项与 PythonOperator 相同。
BranchExternalPythonOperator¶
使用 BranchExternalPythonOperator 执行 Python 分支 任务，它是 BranchPythonOperator 的变体，在外部 Python 环境中执行。
提示
建议使用 `@task.branch_external_python` 装饰器，而不是传统的 `BranchExternalPythonOperator`，来执行 Python 代码。
@task.branch_external_python(python=PATH_TO_PYTHON_BINARY)
def branching_ext_python(choices) -> str:
import random
return f"ext_py_{random.choice(choices)}"
def branch_with_external_python(choices):
import random
return f"ext_py_{random.choice(choices)}"
branching_ext_py = BranchExternalPythonOperator(
task_id="branching_ext_python",
python=PATH_TO_PYTHON_BINARY,
python_callable=branch_with_external_python,
op_args=[options],
)
传入参数和模板化¶
参数传递和模板化选项与 PythonOperator 相同。
ShortCircuitOperator¶
使用 ShortCircuitOperator 控制当条件满足或返回真值时，流水线是否继续。
该条件的评估以及真值的获取通过可调用对象的输出完成。如果 callable 返回 True 或真值，流水线将继续，并会推送输出的 XCom。如果输出为 False 或假值，则根据配置的短路行为（后文会详细说明）截断流水线。在下面的示例中，“condition_is_true”任务之后的任务会执行，而“condition_is_false”任务下游的任务则会被跳过。
提示
建议使用 `@task.short_circuit` 装饰器，而不是传统的 `ShortCircuitOperator`，通过 Python 可调用对象实现流水线短路。
@task.short_circuit()
def check_condition(condition):
return condition
ds_true = [EmptyOperator(task_id=f"true_{i}") for i in [1, 2]]
ds_false = [EmptyOperator(task_id=f"false_{i}") for i in [1, 2]]
condition_is_true = check_condition.override(task_id="condition_is_true")(condition=True)
condition_is_false = check_condition.override(task_id="condition_is_false")(condition=False)
chain(condition_is_true, *ds_true)
chain(condition_is_false, *ds_false)
cond_true = ShortCircuitOperator(
task_id="condition_is_True",
python_callable=lambda: True,
)
cond_false = ShortCircuitOperator(
task_id="condition_is_False",
python_callable=lambda: False,
)
ds_true = [EmptyOperator(task_id=f"true_{i}") for i in [1, 2]]
ds_false = [EmptyOperator(task_id=f"false_{i}") for i in [1, 2]]
chain(cond_true, *ds_true)
chain(cond_false, *ds_false)
“短路”行为可以配置为遵循或忽略下游任务的触发规则。如果将 `ignore_downstream_trigger_rules` 设置为 True（默认配置），则所有下游任务都会被跳过，且不考虑任务所定义的 `trigger_rule`。如果将该参数设为 False，则直接下游任务会被跳过，但对其他随后下游任务仍会遵循其指定的 `trigger_rule`。在此短路配置下，操作符假设直接下游任务是明确需要跳过的，而其他后续任务则可能仍需执行。该配置在仅需要**部分**流水线进行短路，而不是跳过所有紧随短路任务的任务时特别有用。
在下面的示例中，注意到 “short_circuit” 任务被配置为尊重下游触发规则。这意味着尽管装饰函数返回 False，导致其后续任务被跳过，但 “task_7” 仍会执行，因为它设置为在上游任务完成（不论状态）后运行（即 `TriggerRule.ALL_DONE` 触发规则）。
[task_1, task_2, task_3, task_4, task_5, task_6] = [
EmptyOperator(task_id=f"task_{i}") for i in range(1, 7)
]
task_7 = EmptyOperator(task_id="task_7", trigger_rule=TriggerRule.ALL_DONE)
short_circuit = check_condition.override(task_id="short_circuit", ignore_downstream_trigger_rules=False)(
condition=False
)
chain(task_1, [task_2, short_circuit], [task_3, task_4], [task_5, task_6], task_7)
[task_1, task_2, task_3, task_4, task_5, task_6] = [
EmptyOperator(task_id=f"task_{i}") for i in range(1, 7)
]
task_7 = EmptyOperator(task_id="task_7", trigger_rule=TriggerRule.ALL_DONE)
short_circuit = ShortCircuitOperator(
task_id="short_circuit", ignore_downstream_trigger_rules=False, python_callable=lambda: False
)
chain(task_1, [task_2, short_circuit], [task_3, task_4], [task_5, task_6], task_7)
传入参数和模板化¶
参数传递和模板化选项与 PythonOperator 相同。