用途
该技能用于指导创建自定义 OpenLineage 提取器，从 Airflow 算子中捕获数据血缘，主要面向那些没有内置血缘支持的算子。它适用于两种典型需求：需要从不受支持或第三方算子获取血缘，以及需要列级血缘（column-level lineage）的场景。
两种实现路径与选择依据
文档明确指出：应尽可能优先使用 OpenLineage Methods 而非自定义提取器，因为提取器更难编写，在算子行为变更后更容易产生偏离，也更难调试。
选择依据（以场景对照方式给出）：
- 你自己拥有或维护的算子：使用 OpenLineage Methods（推荐，最简单）
- 无法修改的第三方算子：使用 Custom Extractor
- 需要列级血缘：使用 OpenLineage Methods 或 Custom Extractor
- 复杂提取逻辑：使用 OpenLineage Methods 或 Custom Extractor
- 简单表级血缘：使用 Inlets/Outlets（最简单，但优先级最低）
主要能力
一、OpenLineage Methods（推荐）
在自己拥有的算子上直接添加 OpenLineage 方法。示例中定义了一个带有内置 OpenLineage 支持的自定义算子类，并实现了三个方法：
get_openlineage_facets_on_start()：任务进入 RUNNING 时调用，返回已知的输入输出get_openlineage_facets_on_complete(task_instance)：任务成功后调用，添加运行时元数据（如行数统计）get_openlineage_facets_on_failure(task_instance)：任务失败后调用，可选，用于部分血缘
只需实现需要的方法。未实现的方法会回退到 Hook 级血缘或 inlets/outlets。示例代码中演示了通过 from ... import 在方法内部进行延迟导入以避免循环导入，并演示了使用 output_statistics_output_dataset.OutputStatisticsOutputDatasetFacet 添加输出统计 facet。
二、Custom Extractors（自定义提取器）
仅在无法修改算子时使用。基本结构为继承 BaseExtractor，实现：
get_operator_classnames()：类方法，返回该提取器处理的算子类名列表_execute_extraction()：在算子执行前调用，用于已知的输入输出，通过self.operator访问算子属性extract_on_complete(task_instance)：在算子执行成功后调用，用于运行时确定的血缘
文档还说明提取器可实现的三种方法及对应时机：_execute_extraction() 在算子运行前，用于静态/已知血缘；extract_on_complete(task_instance) 在成功后，用于运行时确定的血缘；extract_on_failure(task_instance) 在失败后，用于错误情况下的部分血缘。
使用方式
OperatorLineage 结构
OperatorLineage 包含 inputs（输入数据集列表）、outputs（输出数据集列表）、run_facets（运行元数据，如 SQL facet）、job_facets（作业元数据）。
注册提取器
两种方式：
- 配置文件 airflow.cfg：
[openlineage]
extractors = mypackage.extractors.MyOperatorExtractor;mypackage.extractors.AnotherExtractor
- 环境变量：
AIRFLOW__OPENLINEAGE__EXTRACTORS='mypackage.extractors.MyOperatorExtractor;mypackage.extractors.AnotherExtractor'
文档强调：路径必须能从 Airflow worker 导入。将提取器放置在 DAGs 文件夹或已安装的包中。
在 Astro 上
Astro 内置 OpenLineage 集成，无需额外的传输配置。血缘事件会被自动收集并显示在 Astro UI 的 Lineage 标签页中。部署到 Astro 项目的自定义提取器会被自动发现，因此只需在 airflow.cfg 或环境变量中注册并部署即可。
常见模式
文档给出了三个示例：
- SQL 算子提取器：读取算子的 sql 和 conn_id，解析 SQL 找出表（示例简化，实际建议使用如 sqlglot 的 SQL 解析器），通过
sql_job.SQLJobFacet添加查询元数据。 - 文件传输提取器：示例为 S3 到 Snowflake 的提取器，读取 s3_bucket、s3_key、table、schema，输入使用
s3://命名空间，输出使用snowflake://命名空间。 - 从执行中动态获取血缘：
_execute_extraction()仅返回已知输入，extract_on_complete()通过执行期间设置的算子属性（如created_tables）返回动态输出。
常见陷阱
- 循环导入：在顶层导入 Airflow 模块会导致循环导入，建议在方法内部导入。示例中区分了错误（顶层导入）和正确（方法内导入）两种写法。
- 导入路径错误：提取器路径与实际模块位置不匹配，示例对比了错误路径（
extractors.MyExtractor）与正确完整可导入路径（dags.extractors.my_extractor.MyExtractor）。 - 未处理 None：当算子属性为 None 时提取失败，建议检查必需属性并在缺失时返回 None 跳过提取。
测试
文档给出了使用 pytest 和 unittest.mock 的单元测试示例：用 MagicMock 模拟算子并设置属性，创建提取器实例，调用 _execute_extraction()，断言输入输出数据集的数量与名称。
优先级规则
OpenLineage 按以下顺序检查血缘：
- Custom Extractors（最高优先级）
- 算子上的 OpenLineage Methods
- Hook 级血缘（来自 HookLineageCollector）
- Inlets/Outlets（最低优先级）
如果存在自定义提取器，它会覆盖内置提取和 inlets/outlets。
适用场景与限制
适用于需要从 Airflow 算子收集数据血缘的场景，尤其是第三方或不受支持算子、需要列级血缘、以及需要复杂提取逻辑的情况。限制方面：文档强调应优先使用 OpenLineage Methods，自定义提取器更难编写、更易因算子变更而偏离、更难调试；其路径必须可从 Airflow worker 导入；未处理的 None 属性会导致提取失败；顶层导入可能引发循环导入。相关技能包括 annotating-task-lineage（用于 inlets/outlets 的简单表级血缘）、tracing-upstream-lineage（调查数据来源）、tracing-downstream-lineage（调查数据依赖）。关于收费、价格等信息，资料中暂未提供。