Part6：高级技巧：让你的工作流更上一层楼

导语：

一、回顾：你已经掌握了工作流编排框架的基本用法

• 工作流的概念和应用场景： 工作流是由一系列相互关联的步骤组成的， 用于完成特定的业务目标。 工作流编排框架可以帮助我们自动化执行和管理工作流。

• 工作流编排框架的作用和优势： 自动化执行、提高效率、减少错误、增强可靠性和可维护性。

• Prefect 的基本概念：任务 (Task) 和流程 (Flow)： 任务是最小的执行单元， 流程是任务的容器， 用于组织和编排任务。

• 如何安装和配置 Prefect： 使用 pip 或者 uv 安装 Prefect， 以及如何配置 Prefect Cloud 或 Server (可选)。

• 如何定义任务和流程， 以及如何设置任务的参数和依赖关系： 使用 @task 和 @flow 装饰器， 以及 wait_for 参数。

• 如何调度和监控流程： 使用 deployment， Schedules， Automations 以及 Prefect UI 进行监控。

二、动态映射：处理批量数据不再是难题

from prefect import task, flow

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

@flow
def process_all_items(items):
    for item in items:
        process_item(item)

if __name__ == "__main__":
    items = [1, 2, 3, 4, 5]
    process_all_items(items)

• 代码冗长： 需要手动编写循环逻辑。

• 无法充分利用并发执行的优势： 循环是串行执行的， 无法利用 Prefect 的并发执行能力。

from prefect import task, flow

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

@flow
def process_all_items(items):
    process_item.map(items)  # 使用 map 方法

if __name__ == "__main__":
    items = [1, 2, 3, 4, 5]
    process_all_items(items)

   [1, 2, 3, 4, 5]   <-- 输入数据 (items)
        |
        |  通过 map 方法
        v
   +----+----+----+----+----+
   | 任务 | 任务 | 任务 | 任务 | 任务 |  <-- 动态创建多个任务实例
   +----+----+----+----+----+
      |    |    |    |    |
      |    |    |    |    |   并发执行
      v    v    v    v    v
   [结果1][结果2][结果3][结果4][结果5] <-- 每个任务的执行结果

• 简化代码逻辑： 无需手动编写循环逻辑， 代码更加简洁。

• 提高代码的可读性和可维护性： 代码逻辑更加清晰， 更易于理解和维护。

• 充分利用并发执行的优势： Prefect 会自动并发地执行任务实例， 提高了工作流的执行效率。

三、自定义组件：扩展框架，满足你的特殊需求

• 提高代码的可复用性： 可以将常用的操作封装成自定义组件， 并在不同的流程中重复使用。

• 扩展 Prefect 的功能： 可以根据自己的需求定制 Prefect 的行为， 满足特定的业务需求。

• 让工作流更加灵活和可定制： 可以根据不同的场景选择使用不同的自定义组件。

from prefect import task, flow
from prefect.tasks import Task

class MyCustomTask(Task):
    def __init__(self, message, **kwargs):
        super().__init__(**kwargs)
        self.message = message

    def run(self):
        print(f"Custom task: {self.message}")

@flow
def my_flow():
    custom_task = MyCustomTask(message="Hello from custom task!")
    custom_task()

if __name__ == "__main__":
    my_flow()

四、集成外部服务：与数据库，云平台无缝对接

• Secret：用于存储敏感信息， 如密码、API 密钥等。

• LocalFileSystem：用于操作本地文件系统。

• S3Bucket：用于连接到 AWS S3 云存储。

• KubernetesClusterConfig：用于连接到 Kubernetes 集群。

• ... 还有许多其他类型的 Blocks， 可以参考 Prefect 的官方文档。

from prefect import task, flow
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket

# 提前在 Prefect UI 中创建好 S3Bucket 和 AwsCredentials 的 Block

@task
def upload_file_to_s3(filename, bucket_block_name, credentials_block_name):
    aws_credentials = AwsCredentials.load(credentials_block_name)
    s3_bucket = S3Bucket.load(bucket_block_name, credentials=aws_credentials)
    s3_bucket.upload_from_path(filename, f"s3://{s3_bucket.bucket_name}/{filename}")

@flow
def my_flow():
    upload_file_to_s3("my_file.txt", "my-s3-bucket", "my-aws-credentials")

if __name__ == "__main__":
    # 创建一个测试文件
    with open("my_file.txt", "w") as f:
        f.write("This is a test file.")
    my_flow()

+-----------------+       +-----------------+
|  Prefect 任务   |       |  AWS S3 Bucket |
|                 |       |                 |
|  upload_file    |  ---> |  my-s3-bucket   |
|                 |       |                 |
+-----------------+       +-----------------+
       ^                         ^
       |        +------------+   |
       |        | S3Bucket   |   |
       |        |   Block    |   |
       |        +------------+   |
       |                         |
       +-------------------------+

• 安全地存储和管理连接信息： 避免将敏感信息硬编码在代码中， 提高了安全性。

• 方便地在不同的流程和任务中复用连接信息： 无需在每个任务中都重复配置连接信息。

• 简化与外部服务的交互： Prefect 提供了统一的接口来访问不同的外部服务， 让我们可以更专注于业务逻辑的实现。

五、结果持久化：保存和复用任务的结果

• 中间结果的缓存和复用： 对于一些计算量大、耗时长的任务， 可以将其结果持久化保存下来， 并在后续的流程中复用， 避免重复计算。

• 任务结果的长期保存和分析： 可以将任务的结果保存到数据库、文件系统或者云存储中， 以便进行长期的保存和分析。

• 构建数据管道： 可以将一个任务的结果保存下来， 并作为另一个任务的输入， 从而构建数据管道。

• 本地文件系统 (Local File System)： 可以将任务的结果保存到本地文件系统中。

• 云存储服务： 可以将任务的结果保存到云存储服务中， 例如 AWS S3, Google Cloud Storage, Azure Blob Storage 等等。

from prefect import task, flow
from prefect.filesystems import LocalFileSystem

@task(persist_result=True, result_storage=LocalFileSystem.load("my-result-storage"))
def my_task():
    result = {"message": "This is a task result."}
    return result

@flow
def my_flow():
    my_task()

if __name__ == "__main__":
    # 首先，你需要创建一个名为 `my-result-storage` 的 LocalFileSystem Block， 可以在 Prefect UI 中创建。
    my_flow()

• 选择合适的存储方式， 需要考虑存储成本、访问速度、安全性等因素。

• 注意清理过期的结果， 避免存储空间的浪费。

六、缓存：避免重复执行，提高效率

+-----------------+       +-----------------+
|  任务 (Task)    |       |  缓存 (Cache)  |
|  输入参数       |  ---> |  计算缓存键     |
|  代码           |  ---> |  (Cache Key)   |
+-----------------+       +-----------------+
       |                         ^
       |                         |
       |  检查缓存是否存在       |
       v                         |
+-----------------+       +-----------------+
|  执行任务       |       |  返回缓存结果   |
+-----------------+       +-----------------+

from prefect import task, flow
from datetime import timedelta

@task(cache_key_fn=lambda context, parameters: parameters["item"], cache_expiration=timedelta(minutes=1))
def my_task(item):
    print(f"Processing item: {item}")
    return item * 2

@flow
def my_flow():
    my_task(1)
    my_task(2)
    my_task(1)  # 由于缓存, 该任务不会被执行

if __name__ == "__main__":
    my_flow()

• 确保缓存键的计算方法能够唯一地标识任务的输入和代码。

• 合理设置缓存的过期时间， 避免缓存过期导致的数据不一致问题。

七、 Artifacts: 记录和展示任务执行结果

from prefect import task, flow, artifacts

@task
def my_task():
    result = "This is a task result."
    artifacts.create_markdown_artifact(
        markdown=f"## Task Result\\n\\n{result}",
        key="task-result"
    )

@flow
def my_flow():
    my_task()

if __name__ == "__main__":
    my_flow()

八、最佳实践：分享一些使用工作流编排框架的经验

• 模块化设计： 将复杂的工作流分解成更小的、更易于管理的模块 (任务和流程)。 这样可以提高代码的可读性、可维护性和可复用性。

• 参数化任务： 使用参数来配置任务的行为， 提高任务的灵活性和可复用性。 避免将配置信息硬编码在任务的代码中。

• 充分利用并发： 通过合理地定义任务的依赖关系和使用动态映射， 充分利用 Prefect 的并发执行能力， 提高工作流的执行效率。

• 错误处理： 考虑各种可能出现的错误情况， 并进行相应的处理， 保证工作流的健壮性。 使用 try...except 语句捕获异常， 使用 retries 参数设置重试策略， 使用 on_failure 参数定义任务失败时的回调函数。

• 日志记录： 使用 get_run_logger 记录详细的日志信息， 方便排查问题和跟踪工作流的执行情况。

• 版本控制： 使用版本控制来管理工作流的变更历史， 方便回滚到之前的版本。

• 监控和告警： 设置监控和告警规则， 及时发现和处理问题。 使用 Prefect UI 监控流程的执行状态， 设置 Automations 来发送告警信息。

• 代码规范： 遵循良好的代码规范， 提高代码的可读性和可维护性。 使用有意义的变量名和函数名， 添加必要的注释。

• Task Runner 的选择和使用:
• Prefect 提供了不同类型的 Task Runner， 用于执行任务。 包括 SequentialTaskRunner, ConcurrentTaskRunner, DaskTaskRunner 和 RayTaskRunner (后面两个用于分布式任务)。
• 默认情况下， Prefect 使用 ConcurrentTaskRunner 来并发执行任务。
• 你可以根据任务的特点和资源情况选择合适的 Task Runner。 例如， 如果你的任务是 CPU 密集型的， 可以使用 DaskTaskRunner 或 RayTaskRunner 来利用多核 CPU 的优势。
• 可以在流程上配置 task_runner 属性来使用不同的 Task Runner。

from prefect import flow
from prefect.task_runners import SequentialTaskRunner

@flow(task_runner=SequentialTaskRunner())
def my_flow():
    # ...

• 调试技巧：
• 使用 get_run_logger 记录详细的日志信息， 方便调试。
• 使用 Prefect UI 查看任务的执行状态和日志信息。
• 使用 flow.visualize() 方法可以可视化流程的结构和任务之间的依赖关系。

my_flow.visualize()

• 使用 Python 调试器 (例如 pdb) 来调试任务的代码。

总结：