annotating-task-lineage

用途

本技能指导你使用 inlets 和 outlets 为 Airflow 任务添加人工数据血缘注释，用于标注任务读取和写入的数据集。

主要能力

在操作符上设置 inlets（任务读取的数据）和 outlets（任务写入的数据）。
支持 OpenLineage Dataset 对象（推荐）、Airflow Assets（Airflow 3+）以及 Airflow Datasets（Airflow 2.4+）。
支持为单个任务指定多个输入和多个输出。
支持在自定义算子的 execute 方法中动态设置 inlets/outlets（仅限非 deferrable 算子）。
提供 OpenLineage 命名辅助工具（Snowflake、BigQuery、S3、PostgreSQL）以保证跨平台命名一致性。
在 Astro 上，使用 inlets/outlets 定义的血缘注释会显示在 Astro 增强的 Lineage 标签页中，提供跨 DAG 和跨部署的血缘视图。

使用方式

在操作符上直接传入 inlets 和 outlets 参数，例如 BashOperator 的 inlets=[source_table], outlets=[target_table]。可以使用 OpenLineage 的 Dataset 类定义数据集，指定 namespace 和 name。对于数据库表，命名辅助工具会根据平台生成规范的 namespace 和 name。

在自定义算子中，有两种方式：一是实现 OpenLineage 方法（如 get_openlineage_facets_on_complete），二是直接在 execute 方法中设置 self.inlets 和 self.outlets（仅限非 deferrable 算子）。

适用场景

操作符没有内置 OpenLineage 提取器时。
表级血缘已足够。
需要快速设置血缘而无需编写自定义代码时。
希望血缘注释立即在 Astro UI 中可见时。

限制

仅支持表级血缘，不支持列级血缘。
属于最低优先级的血缘提取方式，若存在 OpenLineage 提取器或方法，则会优先采用后者。
在 DAG 解析时是静态的，若需动态设置需在 execute() 中设置或使用 OpenLineage 方法。
Deferrable 算子会丢失动态血缘，因为 execute() 中设置的属性在 defer 时会丢失。
若提取器或方法存在但未返回数据集，OpenLineage 会检查 hook 级血缘，然后回退到 inlets/outlets。

优先级规则

OpenLineage 的血缘提取优先级为：自定义提取器（最高）、OpenLineage 方法、Hook 级血缘、inlets/outlets（最低）。

最佳实践

始终使用 OpenLineage 命名辅助工具创建数据集。
添加注释说明数据流。
在 SQL 查询变更时更新 inlets/outlets；将 JOIN 中引用的所有表包含为 inlets；将所有写入的表（包括相关临时表）包含为 outlets。
仅输出或仅输入的注释是有效的，鼓励使用单边注释以提高血缘可见性。

技能介绍

用途