Assets

Asset = 工作流产生的数据或结果。 在 Prefect 中，Asset 用来表示：

• 数据（文件、表、模型等）
• 工作流输出
• 数据之间的依赖关系（lineage）

核心概念

Asset Key

每个 Asset 都由一个唯一的 URI 标识:

• s3://data-lake/file.csv
• postgres://db/table

作用：

• 唯一标识资产
• 自动分组（按 scheme）
• 支持层级结构（路径）

Asset 生命周期

Materialization

当 workflow 创建或更新数据时：

from prefect.assets import materialize

@materialize("s3://data/file.csv")
def my_task():
    ...

特点：

• 类似 Task，但用于声明“我要产出一个数据”
• Prefect 会记录这次“产出尝试”
• 成功 / 失败由 Task 状态决定

Reference

当一个 asset 被作为输入使用：

• Prefect 会自动记录依赖关系
• 不需要手动声明（通常）

也可以手动指定：

@materialize(
    "s3://output.csv",
    asset_deps=["s3://input.csv"]
)

依赖建模

Prefect 会构建 数据依赖图，方式有两种：

1. 自动推断（推荐）

   • 基于 Task 输入输出
   • 数据流自动建立依赖

2. 手动声明

   from prefect.assets import materialize
   
   
   @materialize(
       "s3://warehouse/enriched-data.csv",
       asset_deps=["postgres://db/reference-tables", "s3://external/vendor-data.csv"]
   )
   def enrich_data():
       # Explicitly depends on external database and vendor data
       pass
   

   适合：

   • 外部数据源
   • 无法从代码推断的依赖

Metadata

Asset 可以附带元信息：

• Name：人类可读的标识符，用于唯一标识该 asset
• Description：详细文档，支持 Markdown 格式，便于提供丰富的说明与上下文
• Owners：负责方，Prefect 用户和团队在 UI 中有特殊显示
• URL：用于访问或查看该 asset 的网页地址

健康状态

基于最近一次 materialization：

事件系统

Asset 会自动发事件。

事件类型

1. Materialization 事件， 表示数据是否成功产出

   • prefect.asset.materialization.succeeded
   • prefect.asset.materialization.failed

2. Reference 事件， 表示这个数据被使用了

   • prefect.asset.referenced

触发规则

组织方式

Prefect UI 自动组织 assets：

按 URI 分组

s3://
postgres://
snowflake://

按路径层级

s3://data-lake/
    ├── raw/
    └── processed/

支持搜索

基于： - 名称 - 描述 - owner