Apache Airflow TaskFlow API

2025 年 6 月 10 日 | 阅读 8 分钟

引言

Apache Airflow 是一个开源平台，旨在以编程方式编写、调度和监控工作流。在其众多功能中，TaskFlow API 是最强大的工具之一，提供了一种直观、Pythonic 的方式来定义和管理工作流。本指南深入探讨 TaskFlow API 的细节，探索其组件、功能和最佳实践。

什么是 TaskFlow API？

TaskFlow API 在 Apache Airflow 2.0 中引入，旨在简化有向无环图 (DAG) 的创建和管理。它彻底改变了工作流的定义和管理方式，摆脱了对操作符和手动依赖配置的传统依赖。相反，TaskFlow API 使用 Python 函数定义任务，使工作流更加直观、模块化和可维护。

TaskFlow API 的功能

Pythonic 工作流定义：TaskFlow API 允许开发人员使用 Python 函数作为任务。这种方法与 Pythonic 原则相符，使开发人员更容易编写、阅读和维护工作流。
装饰器语法：@task 装饰器简化了任务的定义过程。通过使用 @task 包装 Python 函数，您可以自动将其集成到 Airflow 的 DAG 基础设施中，而无需额外的样板代码。这大大降低了新用户的学习曲线。
数据传递：TaskFlow API 的突出功能之一是其在任务之间无缝传递数据的能力。您无需依赖 XCom（Airflow 在任务之间共享数据的机制），只需使用函数参数和返回值即可实现数据流。这种抽象降低了复杂性和潜在错误。
类型提示：完全支持 Python 类型提示，允许开发人员定义任务的输入和输出类型。类型提示增强了代码可读性，并确保整个工作流的数据一致性。它们还可以作为一种文档形式，帮助其他开发人员理解预期的数据格式。
动态任务映射：动态任务映射允许根据输入数据在运行时创建任务。此功能对于需要处理可变大小数据集或处理依赖于外部动态生成输入的功能的工作流特别有用。
易于调试：TaskFlow API 中的任务是简单的 Python 函数，这意味着它们可以独立于 Airflow 进行测试和调试。此功能对于快速识别和解决问题（尤其是在复杂工作流中）非常宝贵。

传统工作流与 TaskFlow API

在传统的 Airflow 工作流中，任务使用操作符定义，依赖关系使用 >> 或 << 操作符手动设置。虽然功能正常，但这种方法通常会导致冗长且可读性较差的 DAG，尤其对于复杂的管道。由于它们与 Airflow 的执行环境紧密耦合，调试此类工作流也具有挑战性。

将任意对象作为参数传递

在 Apache Airflow 的 TaskFlow API 中，您可以在任务之间传递任意 Python 对象。此功能可在工作流中实现无缝数据流，无需手动序列化或依赖 Airflow 特定的机制（如 XCom）。TaskFlow API 在底层处理序列化和反序列化，确保数据完整性和简单性。

序列化：Airflow 默认使用 JSON 可序列化格式。确保在任务之间传递的对象可以使用 JSON 进行序列化和反序列化。
大对象：应尽可能避免传递大对象，以防止性能瓶颈。对于大型数据集，请使用外部存储机制。
自定义数据类型：如果您的对象不能原生序列化，请实现自定义序列化逻辑，或使用 Airflow 钩子/操作符与外部系统集成。

传递任意对象的示例

以下是使用 TaskFlow API 在任务之间传递复杂 Python 对象的方法

From airflow import, DAG
. decorators import task
from datetime import datetime

# Define a custom class
class CustomData:
    def __init__(self, name, value):
        self.name = name
        self.value = value

    def __repr__(self):
        return f"CustomData(name={self.name}, value={self.value})"

# Define the DAG
default_args = {
    'retries': 2,
    'retry_delay': time delta(minutes=5),
}

dag = DAG(
    dag_id='example_pass_objects',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2023, 1, 1),
    catchup=False
)

@task
def create_data():
    return CustomData(name="Sample", value=42)

@task
def process_data(data: CustomData):
    print(f"Processing: {data}")
    data.value *= 2
    return data

@task
def finalize_data(data: CustomData):
    print(f"Finalized: {data}")

With dag:
    data = create_data()
    processed_data = process_data(data)
    finalize_data(processed_data)

传递任意对象的优势

增强灵活性：简化了需要复杂数据结构的工作流的创建。
改进模块化：将数据封装在对象中，使任务更具可重用性和可维护性。
减少样板：自动处理序列化和反序列化，无需手动干预。
无缝集成：将 Python 的面向对象功能与 Airflow 的编排功能相结合。

自定义对象

自定义对象允许工作流以更结构化的形式封装和管理数据。这些对象可以包括复杂的数据结构或特定领域的信息。使用自定义对象可增强 DAG 定义的模块化、可读性和可伸缩性。

示例：TaskFlow API 中的自定义对象

class Order:
    def __init__(self, order_id, items):
        self.order_id = order_id
        self.items = items

    def __repr__(self):
        return f"Order(id={self.order_id}, items={self.items})"

@task
def create_order():
    return Order(order_id=123, items=["apple," "banana," "cherry"])

@task
def process_order(order: Order):
    print(f"Processing order {order.order_id} with items: {order.items}")

With dag:
    order = create_order()
    process_order(order)

对象版本控制

随着工作流的演变，自定义对象的定义可能会发生变化。对象版本控制可确保工作流不同版本之间的兼容性，尤其是在重新处理历史数据或处理向后兼容性时。

对象版本控制策略

添加元数据：在自定义对象中包含版本信息。

Class Order:	
    def __init__(self, order_id, items, version):
        self.order_id = order_id
        self.items = items
        self.version = version

版本特定逻辑：根据对象版本实现处理逻辑。
序列化注意事项：确保对象结构的更改在序列化时向后兼容。

传感器和 TaskFlow API

传感器是 Airflow 中的特殊任务，它们会等待特定条件满足后才执行下游任务。TaskFlow API 可以将传感器无缝集成到工作流中，使其更具事件驱动性。

TaskFlow API 入门

在深入了解示例之前，请确保您已安装并正确设置了 Airflow。TaskFlow API 需要 Airflow 2.0 或更高版本。

安装 Apache Airflow

您还可以根据需要安装额外的提供程序和依赖项来满足您的工作流需求。

TaskFlow DAG 的基本结构

基于 TaskFlow 的 DAG 通常包括

导入必要的模块。
使用 @task 装饰器定义任务。
通过按所需顺序调用函数来建立任务之间的依赖关系。

下面是一个简单的例子

From airflow import, DAG
.decorators import task
from datetime import datetime, timedelta

# Define the DAG
default_args = {
    'retries': 1,
    'retry_delay': time delta(minutes=5),
}

dag = DAG(
    dag_id="example_taskflow_api",
    default_args=default_args,
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False,
)

@task
def extract():
    return [1, 2, 3]

@task
def transform(data):
    return [x * 2 for x in data]

@task
def load(data):
    print(f"Loaded data: {data}")

# Define the workflow
with dag:
    extracted_data = extract()
    transformed_data = transform(extracted_data)
    load(transformed_data)

TaskFlow API 的组件

@task 装饰器

@task 装饰器将 Python 函数转换为 Airflow 任务。它包装函数以实现与 DAG 和 Airflow 调度机制的无缝集成。

任务之间的数据传递

数据可以使用函数参数和返回值在任务之间传递。在大多数情况下，这消除了对 XCom 的需求，使工作流更清晰、更直观。

它的工作原理

当任务返回一个值时，Airflow 会自动将其输出存储在其元数据数据库中。然后，其他任务可以将其作为输入参数访问。这种机制是无缝的，抽象了手动通过 XCom 推送和拉取数据的需求。

简单性：任务直接通过函数调用进行通信，使依赖关系清晰易懂。
数据验证：利用 Python 的类型提示来确保数据一致性。
改进调试：每个任务仍然是一个独立的 Python 函数，可以独立测试。

示例

@task
def extract_data():
    return ["Airflow," "TaskFlow," "API"]

@task
def transform_data(data: list):
    return [item.upper() for item in data]

@task
def load_data(data: list):
    print(f"Data loaded: {data}")

# Define dependencies
with dag:
    extracted_data = extract_data()
    transformed_data = transform_data(extracted_data)
    load_data(transformed_data)

在此示例中

extract_data 任务生成一个字符串列表。
transform_data 任务使用此列表，对其进行修改，并生成一个新列表。
load_data 任务使用转换后的列表并完成工作流。

在本地测试任务：由于每个任务都是一个独立的 Python 函数，您可以在 Airflow 环境之外对其进行测试

if __name__ == "__main__":
    extracted = extract_data()
    transformed = transform_data(extracted)
    load_data(transformed)

这种本地测试功能对于调试非常宝贵，并确保在部署 DAG 之前您的逻辑是正确的。

数据传递中的错误处理：如果任务遇到错误且未生成输出，则下游函数将不会执行，因为它们依赖于失败任务的输出。此行为可确保数据完整性并尽早发现管道中的问题。

带类型提示和列表的高级示例

From typing import List

@task
def extract_numbers() -> List[int]:
    return [1, 2, 3, 4]

@task
def square_numbers(numbers: List[int]) -> List[int]:
    return [n ** 2 for n in numbers]

@task
def sum_numbers(numbers: List[int]) -> int:
    return sum(numbers)

@task
def print_result(total: int):
    print(f"The sum of squares is: {total}")

# Define dependencies
with dag:
    numbers = extract_numbers()
    squares = square_numbers(numbers)
    total = sum_numbers(squares)
    print_result(total)

此示例说明了

类型提示：确保任务之间传递的数据与预期类型匹配。
链式依赖：每个任务的输出直接馈入下一个任务，形成清晰的逻辑工作流。
易于测试：每个函数都可以独立测试，确保集成前的正确性。

类型提示

TaskFlow API 支持 Python 类型提示，这有助于确保数据完整性并提高代码可读性。

示例

@task
def add(a: int, b: int) -> int:
    return a + b

动态任务映射

动态任务映射允许根据输入数据在运行时动态创建任务。这对于任务数量取决于外部数据的工作流特别有用。

示例

@task
def process_item(item):
    print(f"Processing {item}")

@task
def generate_items():
    return ["item1", "item2", "item3"]

# Dynamically map tasks
with dag:
    items = generate_items()
    process_item.expand(item=items)

任务组

任务组将任务组织成逻辑分组，提高了 DAG 的可读性，尤其对于复杂的工作流。

示例

From airflow. Utils.task_group import TaskGroup

@task
def task_a():
    print("Task A")

@task
def task_b():
    print("Task B")

@task
def task_c():
    print("Task C")

With dag:
    with TaskGroup("group1") as group1:
        task_a() >> task_b()

    task_c() >> group1

高级概念

错误处理和重试

每个任务都可以定义其重试策略，确保在瞬态故障时具有鲁棒性。

示例

from datetime import timedelta
import random

@task(retries=3, retry_delay=timedelta(seconds=10))
def unreliable_task():
    if random.random() < 0.7:
        raise ValueError("Task failed!")
    return "Success"

分支

分支允许工作流根据运行时数据有条件地执行特定任务。

示例

From airflow. Operators. python import branch_python_operator
import random

@task
def choose_branch():
    return "task_a" if random.random() > 0.5 else "task_b"

@task
def task_a():
    print("Branch A")

@task
def task_b():
    print("Branch B")

With dag:
    branch = choose_branch()
    branch >> [task_a(), task_b()]

调试 TaskFlow 工作流

与传统的 Airflow 任务相比，调试 TaskFlow 工作流更简单，因为 Python 函数可以独立于 DAG 进行测试。要在本地测试函数

if __name__ == "__main__":
    result = my_task()
    print(result)

最佳实践

保持任务简单：每个任务都应承担单一职责。
使用类型提示：增强代码可读性并确保数据一致性。
利用任务组：逻辑地组织任务以提高 DAG 的清晰度。
在本地测试函数：在将函数集成到 DAG 之前测试单个任务函数。
文档工作流：使用文档字符串和注释来描述每个任务和整个工作流的目的。
监控性能：使用 Airflow 的内置工具监控任务执行并根据需要进行优化。

TaskFlow API 是 Apache Airflow 的一个游戏规则改变者，它使工作流定义更加直观、可维护和 Pythonic。通过利用其功能（例如 @task 装饰器、动态任务映射和类型提示），开发人员可以轻松构建健壮且可扩展的工作流。无论您是经验丰富的 Airflow 用户还是工作流编排新手，TaskFlow API 都是您工具箱中必备的工具。

下一主题Apache-airflow-tasks-lifecycle

Apache Airflow TaskFlow API

引言

什么是 TaskFlow API？

TaskFlow API 的功能

传统工作流与 TaskFlow API

将任意对象作为参数传递

自定义对象

对象版本控制

对象版本控制策略

传感器和 TaskFlow API

TaskFlow API 入门

TaskFlow DAG 的基本结构

TaskFlow API 的组件

@task 装饰器

任务之间的数据传递

它的工作原理

带类型提示和列表的高级示例

类型提示

动态任务映射

任务组

高级概念

分支

调试 TaskFlow 工作流

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Airflow 教程

Apache Airflow TaskFlow API

引言

什么是 TaskFlow API？

TaskFlow API 的功能

传统工作流与 TaskFlow API

将任意对象作为参数传递

自定义对象

对象版本控制

对象版本控制策略

传感器和 TaskFlow API

TaskFlow API 入门

TaskFlow DAG 的基本结构

TaskFlow API 的组件

@task 装饰器

任务之间的数据传递

它的工作原理

带类型提示和列表的高级示例

类型提示

动态任务映射

任务组

高级概念

分支

调试 TaskFlow 工作流

相关帖子

Apache Airflow 部署与生产

Apache Airflow 数据库迁移参考

Apache Airflow 故障排除

使用公共接口与外部服务和应用程序集成

自定义 Apache Airflow 中的 UI

Apache Airflow 密钥后端

使用 Apache Airflow 扩展机器学习工作流

Apache Airflow 血缘关系

Apache Airflow

Apache Airflow 池

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器